Hive环境搭建与入门

一、概述
hive是一个基于hadoop的数据仓库基础设施, 提供数据的提取转换加载和查询, 不适于少数据量(比如几亿)的数据查询, 也不适于实时或者在线数据的查询, 是一个非实时的, 离线的数据查询工具. hive没有自己的文件格式, 只需要普通的文件格式或者用户自定义的格式就可以了.

二、环境搭建

hive不需要神马集群配置, 只是需要编译一个环境就好了, 编译的过程, 国人表示很蛋疼, 因为他的ant脚本中, 使用的镜像居然是facebook的, 翻墙了都没用表示很纠结. 所以需要把facebook的镜像改为可用的(最好是国内的, 诸如 人人网, 北京交通大学等等镜像), 镜像的选择可以根据个人喜好和地理位置自由选取, 当然是不能被GFW墙了的.(参考http://www.apache.org/dyn/closer.cgi/hadoop/core/)
1. 下载源代码: svn co http://svn.apache.org/repos/asf/hive/trunk hive

2. 安装ant, 过程简单, 略去 (只要解压就行了)
3. 修改 hive/build.properties 文件, 将
a). hadoop.mirror的值修改为人人网镜像 : hadoop.mirror=http://labs.renren.com/apache-mirror
b). hadoop.security.url的值修改为可用的地址, 在网上找了个: hadoop.security.url=http://people.apache.org/~todd/snapshot-for-hive-build/hadoop-0.20.3-CDH3-SNAPSHOT.tar.gz
(这个东东不是一般的镜像能有的,囧)
c). 将hadoop的版本修改为0.20.2, (貌似0.20.0的地址都不可用了, 难道有大bug?): hadoop.version=0.20.2
4. 修改hive/shims/ 下的ivy.xml 和 build.xml
将对应的hadoop的版本修改为 0.20.2
a) ivy.xml

b) build.xml
(line 61)

<param name="hadoop.version.ant-internal" value="0.20.2" />

5. 编译
进入源代码的根目录hive, 敲下命令: $ant_home/bin/ant package
等待完成… (下载时间可能会有点久, 给点耐心吧).
6. 启动hive, 只需要启动前将 $HADOOP_HOME设置为hadoop的安装目录即可.

 

三、用法和例子

1， 创建表

CREATE TABLE page_view(viewTime INT, userid BIGINT,
page_url STRING, referrer_url STRING,
ip STRING COMMENT 'IP Address of the User')
COMMENT 'This is the page view table'
PARTITIONED BY(dt STRING, country STRING)
STORED AS SEQUENCEFILE;

partitioned by 是创建分区(什么是分区?下边说明)
没有指定字段分隔符的话, 默认是使用^A(ctrl-A)为字段分隔符, 换行为记录分隔符. 指定分隔符用ROW FORMAT row_format 语句

分区: partitioned by , 其实就是在数据的目录下, 用不同目录来区分, 比如, dt, 就是按日期(date)来区分, country 国家, hour 小时等等.对应的会在数据的目录下有分区目录. 可以建双分区, 就是子目录下再分区(其实就是一棵目录树).

参考: http://blog.csdn.net/dajuezhao/archive/2010/07/21/5753055.aspx

高级用法:

CREATE TABLE page_view(viewTime INT, userid BIGINT,
page_url STRING, referrer_url STRING,
ip STRING COMMENT 'IP Address of the User')
COMMENT 'This is the page view table'
PARTITIONED BY(dt STRING, country STRING)
CLUSTERED BY(userid) SORTED BY(viewTime) INTO 32 BUCKETS
ROW FORMAT DELIMITED
FIELDS TERMINATED BY '1'
COLLECTION ITEMS TERMINATED BY '2'
MAP KEYS TERMINATED BY '3'
STORED AS SEQUENCEFILE;

clustered by 是按照某个字段来进行hash集群

2. 查看表和表结构

SHOW TABLES;

列出数据仓库中的所有表

SHOW TABLES 'page.*';

列出所有以”page”为前缀的表, ‘page.*’ 用法和java正则表达式相同

SHOW PARTITIONS page_view;

列出指定表的所有分区, 如果这个表没有分区, 则显示错误

DESCRIBE page_view;

与普通DBMS的desc一样, 列出指定表的所有字段和字段类型.

DESCRIBE EXTENDED page_view;

列出指定表的所有字段和字段类型, 以及其他所有属性.一般用于调试(显示的格式不友好)

DESCRIBE EXTENDED page_view PARTITION (ds='2008-08-08');

列出指定表指定分区的搜有字段和字段类型, 以及其他所有属性.(一般也是用于调试)

3. 装载数据
装载数据有很多方法:
a. 通过创建一个扩展表, 并指定其location, 然后通过hadoop 的 dfs -put命令将数据放到location指定的位置即可.

CREATE EXTERNAL TABLE page_view_stg(viewTime INT, userid BIGINT,
page_url STRING, referrer_url STRING,
ip STRING COMMENT 'IP Address of the User',
country STRING COMMENT 'country of origination')
COMMENT 'This is the staging page view table'
ROW FORMAT DELIMITED FIELDS TERMINATED BY '44' LINES TERMINATED BY '12'
STORED AS TEXTFILE
LOCATION '/user/data/staging/page_view';

hadoop dfs -put /tmp/pv_2008-06-08.txt /user/data/staging/page_view

b. 使用load语句
可以load 本地文件和hdfs文件

LOAD DATA LOCAL INPATH `/tmp/pv_2008-06-08_us.txt` INTO TABLE page_view PARTITION(date='2008-06-08', country='US')

加载本地文件到指定的表中, 还可以指定表的分区.

LOAD DATA INPATH '/user/data/pv_2008-06-08_us.txt' INTO TABLE page_view PARTITION(date='2008-06-08', country='US')

加载hdfs文件到指定的表中, 还可以指定表的分区.

这里的inpath 后便的参数, 可以普通文件路径(单文件), 也可以是一个目录的路径(该目录下的文件都会被加载, 但要求该目录下没有子目录), 也可以是通配符(如 page_view*.txt, 但通配符匹配的也只是普通文件, 不匹配目录)

4. 简单查询

INSERT OVERWRITE TABLE user_active
SELECT user.*
FROM user
WHERE user.active = 1;

与SQL不同的是, 我们将查询结果插入到一个表中, 后续会讲到如何查看这个结果表的数据, 或者如何dump到本地文件中.
如果是在hive的control端 hive cli中查询并查看数据, 就不需要插入到某个表中了, 可以直接查看, 如:

SELECT user.*
FROM user
WHERE user.active = 1;

5. 基于分区的查询
似乎只能通过where语句来解决

INSERT OVERWRITE TABLE xyz_com_page_views
SELECT page_views.*
FROM page_views
WHERE page_views.date >= '2008-03-01' AND page_views.date <= '2008-03-31' AND
page_views.referrer_url like '%xyz.com';

这里之所以可以这么写, 是因为在定义table的时候, 使用partitioned by 语句指定了分区, 并且分区字段名为 date, 诸如:PARTITIONED BY(date DATETIME, country STRING) , 如果你的名字不叫date, 别指望date能为你做你想要的事.

6. join
这个类似于SQL, 配合关键字on使用

INSERT OVERWRITE TABLE pv_users
SELECT pv.*, u.gender, u.age
FROM user u JOIN page_view pv ON (pv.userid = u.id)
WHERE pv.date = '2008-03-03';

如果想要做外连接, 则可以用 left outer join, right outer join 和 full outer join 等关键字来查询, 例如:

INSERT OVERWRITE TABLE pv_users
SELECT pv.*, u.gender, u.age
FROM user u FULL OUTER JOIN page_view pv ON (pv.userid = u.id)
WHERE pv.date = '2008-03-03';

为了在join时检查一个key在另一个表中的存在性, 可以使用 LEFT SEMI JOIN 关键字来查询, 例如:

INSERT OVERWRITE TABLE pv_users
SELECT u.*
FROM user u LEFT SEMI JOIN page_view pv ON (pv.userid = u.id)
WHERE pv.date = '2008-03-03';

另外, 像SQL一样, 一次查询可以有多个连接的, 一直join下去都没问题.

7. 聚集函数

INSERT OVERWRITE TABLE pv_gender_sum
SELECT pv_users.gender, count (DISTINCT pv_users.userid)
FROM pv_users
GROUP BY pv_users.gender;

跟SQL类似, 使用distinct关键字可以防止重复计算, 但是在同时使用多个聚集函数时, distinct关键字指定的字段必须一致!

8. 多表/文件 插入
insert语句可以插入到hive的表中, 也可以插入到hadoop的hdfs中, 例如

FROM pv_users
INSERT OVERWRITE TABLE pv_gender_sum
SELECT pv_users.gender, count_distinct(pv_users.userid)
GROUP BY pv_users.gender

INSERT OVERWRITE DIRECTORY '/user/data/tmp/pv_age_sum'
SELECT pv_users.age, count_distinct(pv_users.userid)
GROUP BY pv_users.age;

9. 动态分区插入(hive 0.6.0)
没有使用动态分区插入的情况下:

FROM page_view_stg pvs
INSERT OVERWRITE TABLE page_view PARTITION(dt='2008-06-08', country='US')
SELECT pvs.viewTime, pvs.userid, pvs.page_url, pvs.referrer_url, null, null, pvs.ip WHERE pvs.country = 'US'
INSERT OVERWRITE TABLE page_view PARTITION(dt='2008-06-08', country='CA')
SELECT pvs.viewTime, pvs.userid, pvs.page_url, pvs.referrer_url, null, null, pvs.ip WHERE pvs.country = 'CA'
INSERT OVERWRITE TABLE page_view PARTITION(dt='2008-06-08', country='UK')
SELECT pvs.viewTime, pvs.userid, pvs.page_url, pvs.referrer_url, null, null, pvs.ip WHERE pvs.country = 'UK';

如果想要将计算的分类结果分别保存到不同类别的分区中, 按照上面的写法, 必须有多少分区插入多少次, 这样的写法是不方便的, 而且多个insert操作对应多个map reduce job, 这样对系统的性能也有影响. 在hive的0.6.0以上版本中, 增加了动态分区插入的特性, 在计算过程中, hive会自动识别当前字段属于哪个分区的, 然后插入的时候会插入到对应的分区中(如果该分区还没创建, 那hive会自动创建这个分区的).
使用动态分区插入之后, 只需要一个insert操作, 也就是只对应一个map reduce job

FROM page_view_stg pvs
INSERT OVERWRITE TABLE page_view PARTITION(dt='2008-06-08', country)
SELECT pvs.viewTime, pvs.userid, pvs.page_url, pvs.referrer_url, null, null, pvs.ip, pvs.country

这里再partition定义中, 不指定值, select语句中也少了where语句的判断.
在partition中, country字段没有指定值, 说明这是一个动态的分区字段, 但dt就有一个值, 说明这是静态分区字段. 而且在where语句中不需要增加dt的判断, 因为在partition中已经指定了.目前hive只允动态分区字段出现在partition语句中的最后面(可以多个), 因为分区字段的先后顺序表明了他们之间的层级关系.Note that the dynamic partition values are selected by ordering, not name, and taken as the last columns from the select clause.

insert语句中几个语义上要注意的地方
a.如果查询结果对应的分区原来存在的话, 将被覆盖, 如果原来存在的分区, 但是查询结果中没有属于这个分区的, 将不被覆盖.
b.partition语句中的值最终对应的是hdfs上的path, 所以必须这些值必须满足path的命名规范, 否则将被urlencode.
c.如果输入的字段不是string类型, 将会被强制先装换为string类型.
d.如果输入的字段值是NULL或者空(empty),记录将会被放到一个特殊的分区, 这个分区在hive的配置参数中可以得到.
e.动态分区插入的时候可能会在短时间内创建大量的分区, 如果需要自己控制, 可以配置hive.exec.max.dynamic.partitions和hive.exec.max.dynamic.partitions.pernode这两个参数, 前者是配置动态分区时最多产生的分区数, 后者是在每个M/R job中产生的最多的分区数, 但计算过程中超过这些数时, 将会产生致命性的错误, 并停止job的运行.
f.在动态分区插入中, 有一种情况是搜索的partition参数都是动态的, 这样是没有意义的, 所以hive中提供了一个参数来限制这种事的发生:hive.exec.dynamic.partition.mode=strict/nonstrict 默认值strict就是要求至少有一个是静态的分区字段.另外还有一个参数来配置是否支持动态分区插入 hive.exec.dynamic.partition=true/false , 默认是关闭的(false)
g.目前, 如果hive开启了hive.merge.mapfiles=true 或者 hive.merge.mapredfiles=true 会导致不支持动态分区插入, 因为开启了其中一个在job执行的时候就会把文件合并为一个, 这样不利于我们的分区, 因为分区其实也是对应到hadoop的hdfs上的.

10. 插入到本地文件

INSERT OVERWRITE LOCAL DIRECTORY '/tmp/pv_gender_sum'
SELECT pv_gender_sum.*
FROM pv_gender_sum;

11. 抽样
抽样语句允许用户抽取样品数据而不是整个表的数据来进行查询, 抽样语句只适用于在表创建时使用bucketed on 语句进行分桶的表, 例如:

INSERT OVERWRITE TABLE pv_gender_sum_sample
SELECT pv_gender_sum.*
FROM pv_gender_sum TABLESAMPLE(BUCKET 3 OUT OF 32);

抽取总共32桶中的第三桶
抽样语句的语法如下:
TABLESAMPLE(BUCKET x OUT OF y)
其中, x必须比y小, y必须是在创建表的时候bucket on的数量的因子或者倍数, hive会根据y的大小来决定抽样多少, 比如原本分了32分, 当y=16时, 抽取32/16=2分, 这时TABLESAMPLE(BUCKET 3 OUT OF 16) 就意味着要抽取第3和第16+3=19分的样品. 如果y=64， 这要抽取 32/64=1/2份数据, 这时TABLESAMPLE(BUCKET 3 OUT OF 64) 意味着抽取第3份数据的一半来进行.

12. Union all
类似于SQL

INSERT OVERWRITE TABLE actions_users
SELECT u.id, actions.date
FROM (
SELECT av.uid AS uid
FROM action_video av
WHERE av.date = '2008-06-03'

UNION ALL

SELECT ac.uid AS uid
FROM action_comment ac
WHERE ac.date = '2008-06-03'
) actions JOIN users u ON(u.id = actions.uid);

13. 数组操作.
但一个字段的数据类型时数组时, 可以通过数组的索引来访问该字段的某个索引值.

SELECT pv.friends[2]
FROM page_views pv;

另外还提供了一个函数 size, 可以求出数组的大小

SELECT pv.userid, size(pv.friends)
FROM page_view pv;

14. Map(关联性数组)操作
map的访问类似于 php中对数组的访问, 直接用key作为索引来访问数组即可.

INSERT OVERWRITE page_views_map
SELECT pv.userid, pv.properties['page type']
FROM page_views pv;

与数组类似, 也提供了一个求大小的函数 size

SELECT size(pv.properties)
FROM page_view pv;

15. 自定义map/reduce脚本 & cogroups, 太高级了, 还没看懂, 先晾着.

16. 变更表(altering table)
a, ALTER TABLE old_table_name RENAME TO new_table_name; 表重命名
b, ALTER TABLE old_table_name REPLACE COLUMNS (col1 TYPE, …); 字段重命名
c, ALTER TABLE tab1 ADD COLUMNS (c1 INT COMMENT ‘a new int column’, c2 STRING DEFAULT ‘def val’); 增加新的字段.

16. 删除表和分区(drop)
a, DROP TABLE pv_users; 删除表
b, ALTER TABLE pv_users DROP PARTITION (ds=’2008-08-08′)删除分区
操作不可恢复

以上手册本人参考和翻译自 http://wiki.apache.org/hadoop/Hive/Tutorial