Hive与HBase

hive是基于Hadoop的一个数据仓库工具，可以将结构化的数据文件映射为一张数据库表，并提供完整的sql查询功能，可以将sql语句转换为MapReduce任务进行运行。 其优点是学习成本低，可以通过类SQL语句快速实现简单的MapReduce统计，不必开发专门的MapReduce应用，十分适合数据仓库的统计分析。

Hive与HBase的整合功能的实现是利用两者本身对外的API接口互相进行通信，相互通信主要是依靠hive_hbase-handler.jar工具类

1 .Hadoop和Hbase都已经成功安装了

2 . 拷贝hbase-0.90.4.jar和zookeeper-3.3.2.jar到hive/lib下。

注意：如果hive/lib下已经存在这两个文件的其他版本（例如zookeeper-3.3.2.jar），建议删除后使用hbase下的相关版本。

3. 修改hive/conf下hive-site.xml文件，在底部添加如下内容：

注意：如果hive-site.xml不存在则自行创建，或者把hive-default.xml.template文件改名后使用。

同时修改其中配置

1.单节点启动

#bin/hive -hiveconf hbase.master=master:490001

2 集群启动：

#bin/hive

1.创建hbase识别的数据库：

hbase.table.name 定义在hbase的table名称

hbase.columns.mapping 定义在hbase的列族 

2.查看数据

hive> select * from  hbase_table_1;  

这时可以登录Hbase去查看数据了

#bin/hbaseshell

hbase(main):001:0>describe'sunqi'

hbase(main):002:0>scan'sunqi'

hbase(main):003:0>put'sunqi','100','cf1:qualifier','aaaa'

这时在Hive中可以看到刚才在Hbase中插入的数据了。

使用CREATE EXTERNAL TABLE:

hive> select count(*) from  hbase_table_1;  

Total MapReduce jobs = 1

Launching Job 1 out of 1

Number of reduce tasks determined at compile time: 1

In order to change the average load for a reducer (in bytes):

  set hive.exec.reducers.bytes.per.reducer=<number>

In order to limit the maximum number of reducers:

  set hive.exec.reducers.max=<number>

In order to set a constant number of reducers:

  set mapred.reduce.tasks=<number>

Starting Job = job_201205231120_0004, Tracking URL = http://node1:50030/jobdetails.jsp?jobid=job_201205231120_0004

Kill Command = /home/hadoop/hadoop-0.20.203.0/bin/../bin/hadoop job  -Dmapred.job.tracker=node1:49001 -kill job_201205231120_0004

Hadoop job information for Stage-1: number of mappers: 1; number of reducers: 1

2012-05-23 14:20:35,542 Stage-1 map = 0%,  reduce = 0%

2012-05-23 14:20:41,592 Stage-1 map = 100%,  reduce = 0%

2012-05-23 14:20:53,685 Stage-1 map = 100%,  reduce = 100%

Ended Job = job_201205231120_0004

MapReduce Jobs Launched: 

Job 0: Map: 1  Reduce: 1   HDFS Read: 239 HDFS Write: 2 SUCESS

Total MapReduce CPU Time Spent: 0 msec

OK

1

Time taken: 29.128 seconds

自动会在hadoop上创建Map/Reduce任务，执行统计数据

创建表

hive> CREATE TABLE pokes (foo INT, bar STRING); 

创建表并创建索引字段ds

hive> CREATE TABLE invites (foo INT, bar STRING) PARTITIONED BY (ds STRING); 

显示所有表

hive> SHOW TABLES;

按正条件（正则表达式）显示表，

hive> SHOW TABLES '.*s';

表添加一列 

hive> ALTER TABLE pokes ADD COLUMNS (new_col INT);

添加一列并增加列字段注释

hive> ALTER TABLE invites ADD COLUMNS (new_col2 INT COMMENT 'a comment');

更改表名

hive> ALTER TABLE events RENAME TO 3koobecaf;

删除列

hive> DROP TABLE pokes;

元数据存储

将文件中的数据加载到表中

hive> LOAD DATA LOCAL INPATH './examples/files/kv1.txt' OVERWRITE INTO TABLE pokes; 

加载本地数据，同时给定分区信息

hive> LOAD DATA LOCAL INPATH './examples/files/kv2.txt' OVERWRITE INTO TABLE invites PARTITION (ds='2008-08-15');

加载DFS数据 ，同时给定分区信息

hive> LOAD DATA INPATH '/user/myname/kv2.txt' OVERWRITE INTO TABLE invites PARTITION (ds='2008-08-15');

The above command will load data from an HDFS file/directory to the table. Note that loading data from HDFS will result in moving the file/directory. As a result, the operation is almost instantaneous. 

SQL 操作

按先件查询

hive> SELECT a.foo FROM invites a WHERE a.ds='<DATE>';

将查询数据输出至目录

hive> INSERT OVERWRITE DIRECTORY '/tmp/hdfs_out' SELECT a.* FROM invites a WHERE a.ds='<DATE>';

将查询结果输出至本地目录

hive> INSERT OVERWRITE LOCAL DIRECTORY '/tmp/local_out' SELECT a.* FROM pokes a;

选择所有列到本地目录 

hive> INSERT OVERWRITE TABLE events SELECT a.* FROM profiles a;

hive> INSERT OVERWRITE TABLE events SELECT a.* FROM profiles a WHERE a.key < 100; 

hive> INSERT OVERWRITE LOCAL DIRECTORY '/tmp/reg_3' SELECT a.* FROM events a;

hive> INSERT OVERWRITE DIRECTORY '/tmp/reg_4' select a.invites, a.pokes FROM profiles a;

hive> INSERT OVERWRITE DIRECTORY '/tmp/reg_5' SELECT COUNT(1) FROM invites a WHERE a.ds='<DATE>';

hive> INSERT OVERWRITE DIRECTORY '/tmp/reg_5' SELECT a.foo, a.bar FROM invites a;

hive> INSERT OVERWRITE LOCAL DIRECTORY '/tmp/sum' SELECT SUM(a.pc) FROM pc1 a;

将一个表的统计结果插入另一个表中

hive> FROM invites a INSERT OVERWRITE TABLE events SELECT a.bar, count(1) WHERE a.foo > 0 GROUP BY a.bar;

hive> INSERT OVERWRITE TABLE events SELECT a.bar, count(1) FROM invites a WHERE a.foo > 0 GROUP BY a.bar;

JOIN

hive> FROM pokes t1 JOIN invites t2 ON (t1.bar = t2.bar) INSERT OVERWRITE TABLE events SELECT t1.bar, t1.foo, t2.foo;

将多表数据插入到同一表中

FROM src

INSERT OVERWRITE TABLE dest1 SELECT src.* WHERE src.key < 100

INSERT OVERWRITE TABLE dest2 SELECT src.key, src.value WHERE src.key >= 100 and src.key < 200

INSERT OVERWRITE TABLE dest3 PARTITION(ds='2008-04-08', hr='12') SELECT src.key WHERE src.key >= 200 and src.key < 300

INSERT OVERWRITE LOCAL DIRECTORY '/tmp/dest4.out' SELECT src.value WHERE src.key >= 300;

将文件流直接插入文件

hive> FROM invites a INSERT OVERWRITE TABLE events SELECT TRANSFORM(a.foo, a.bar) AS (oof, rab) USING '/bin/cat' WHERE a.ds > '2008-08-09';

This streams the data in the map phase through the script /bin/cat (like hadoop streaming). Similarly - streaming can be used on the reduce side (please see the Hive Tutorial or examples) 

实际示例

创建一个表

CREATE TABLE u_data (

userid INT,

movieid INT,

rating INT,

unixtime STRING)

ROW FORMAT DELIMITED

FIELDS TERMINATED BY '\t'

STORED AS TEXTFILE;

下载示例数据文件，并解压缩

wget http://www.grouplens.org/system/files/ml-data.tar__0.gz

tar xvzf ml-data.tar__0.gz

加载数据到表中

LOAD DATA LOCAL INPATH 'ml-data/u.data'

OVERWRITE INTO TABLE u_data;

统计数据总量

SELECT COUNT(1) FROM u_data;

现在做一些复杂的数据分析

创建一个 weekday_mapper.py: 文件，作为数据按周进行分割 

import sys

import datetime

for line in sys.stdin:

line = line.strip()

userid, movieid, rating, unixtime = line.split('\t')

生成数据的周信息

weekday = datetime.datetime.fromtimestamp(float(unixtime)).isoweekday()

print '\t'.join([userid, movieid, rating, str(weekday)])

使用映射脚本

//创建表，按分割符分割行中的字段值

CREATE TABLE u_data_new (

userid INT,

movieid INT,

rating INT,

weekday INT)

ROW FORMAT DELIMITED

FIELDS TERMINATED BY '\t';

//将python文件加载到系统

add FILE weekday_mapper.py;

将数据按周进行分割

INSERT OVERWRITE TABLE u_data_new

SELECT

TRANSFORM (userid, movieid, rating, unixtime)

USING 'python weekday_mapper.py'

AS (userid, movieid, rating, weekday)

FROM u_data;

SELECT weekday, COUNT(1)

FROM u_data_new

GROUP BY weekday;