Hadoop数据仓库工具hive介绍.pdf

《Hadoop数据仓库工具hive介绍.pdf》由会员分享，可在线阅读，更多相关《Hadoop数据仓库工具hive介绍.pdf（14页珍藏版）》请在淘文阁 - 分享文档赚钱的网站上搜索。

1、百度在线网络技术（北京）有限公司 HIVE 介绍 1 简介 1.1 是什么 hive 是一个基于 hadoop 的数据仓库。使用 hadoop-hdfs 作为数据存储层；提供类似SQL 的语言（HQL），通过 hadoop-mapreduce 完成数据计算；通过 HQL 语言提供使用者部分传统 RDBMS 一样的表格查询特性和分布式存储计算特性。类似的系统有 yahoo 的 pig1，google 的 sawzall2，microsoft 的 DryadLINQ3。1.2 架构 图表 1 hive 架构图4 1、操作界面：CLI，Web，Thrift 2、driver：hive 系统将用户操作

2、转化为 mapreduce 计算的模块（重点）百度在线网络技术（北京）有限公司 3、hadoop：hdfs+mapreduce 4、metastore：存储元数据 1.3 语言 一般有 DDL 和 DML 两种：hive 采用 DDL 方式和少量 DML 方式，类似 sql；pig 使用DML 方式。DDL：data definition language（只讲 definition，不讲实现）create/alter/droptable/view/partition create table as select DML：data manipulation language（有关于实现操作）i

3、nsert overwrite hive 示例 加载 load data local input /logs/urls.txt into table urls partition(ds=2010-01-01);写入 INSERT OVERWRITE TABLE result 操作 SELECT category,AVG(pagerank)FROM urls WHERE pagerank 0.2 GROUP BY category;pig 示例 加载 urls=LOAD /logs/urls.txt USING myLoad()AS(category,pagerank);操作 good_urls

4、=FILTER urls BY pagerank 0.2;groups=GROUP good_urls BY category;output=FOREACH groups GENERATE category,AVG(good_urls.pagerank);写入 STORE output INTO myoutput USING myStore();hive 中使用自定义 map-reduce FROM(FROM pv_users MAP pv_users.userid,pv_users.date USING map_script AS dt,uid CLUSTER BY dt)map_outpu

5、t INSERT OVERWRITE TABLE pv_users_reduced REDUCE map_output.dt,map_output.uid USING reduce_script AS date,count;百度在线网络技术（北京）有限公司 1.4 其他一些功能 1、能够 ALERT 一个 table，主要是 add 一个 column。2、分区（partition）：a)建表的时候指定分区方式：CREATE TABLE invites(foo INT,bar STRING)PARTITIONED BY(ds STRING);b)导入的时候指定分区依据：LOAD DATA LO

6、CAL INPATH./examples/files/kv2.txt OVERWRITE INTO TABLE invites PARTITION(ds=2008-08-15);LOAD DATA LOCAL INPATH./examples/files/kv3.txt OVERWRITE INTO TABLE invites PARTITION(ds=2008-08-08);3、类似 select*from tbl 的查询不需要 MapReduce。4、hive 不只是可以 mapreduce 图表 2 hive 结合 HBase 的逻辑图5”图表 3 reduce 阶段写入 HBase 的

7、方式5”百度在线网络技术（北京）有限公司 图表 4 map-only job 写入 HBase 的方式5”2 实现 2.1 原数据(Metadata)hive 的元数据存储在传统的 RDBMS 中，现在为 mysql 中。采用 JDO（JPOX）。原因：访问这些 Metadata，我们想要“很低的延时”，而存在 hdfs 中是无法满足。（元数据对 hive 是相当重要的，因此一般要求有备份机制）使用：元数据都是在 HQL 语句编译的时候，就被生成一个 xml 文件（包含此次编译所有需要的元数据信息）存储在 hdfs 中，然后运行 mapreduce 时传递给 mapper 和 reducer。

8、（减少后期访问）2.2 查询解析(query parser)这一步是实现中最主要的操作，即架构图中 Driver 的大部分。下面将具体介绍其中的每一个小步。2.2.1 解析(parse)使用 antlr 解析 HQL 语句，并产生 AST（abstract syntax tree）。百度在线网络技术（北京）有限公司 2.2.2 类型检测和语义分析 分析所有输入输出的 table，并创建 logical-plan。通过一种中间表示结构 query block（QB）tree，将 AST 转换成 operator-DAG：将嵌套的 queries 变成父子关系的 QB-tree。2.2.3 优化(

9、Optimization)通过 operator-DAG 的中“元素的前后满足关系”生成一些操作（operator）。主要的五个元素为：Node,GrahpWalder,Dispatcher,Rule,Processor：GraphWalker 遍历（walk）DAG 中所有的 Node，并检查一个 Rule 是否满足，在满足的条件下回出发一个对应的 Processor。Dispatcher 则维护 Rule 到 Processor 的映射，并进行Rule 的匹配工作。图表 5 优化过程中的典型转换流图4 百度在线网络技术（北京）有限公司 简单的几个优化步骤 针对优化，这里给出了一些简单的处理

10、方式：1、列裁剪（Column pruning）：只有需要用到的列才进行输出 2、谓词下推（Predicate pushdown）：尽早进行数据过滤(见图表 7 中，下面为先处理的逻辑)，减少后续处理的数据量 3、分区裁剪（Partition pruning）：只读取满足分区条件的文件 4、map-join：对于 join 中一些小文件，可以在 map 阶段进行 join 操作，见 3.2.2 节 map-join 部分 5、join-reordering：将在 reducer 中进行 join 操作时的小 table 放入内存，而大 table 通过stream 方式读取 6、Group-b

11、y 优化：进行局部聚合进行优化（包括 hash-based 和 sort-based），对于 skew的 key（key 的 row num 和 size 在 reduce 时非常不均）可以进行两次 map-reduce 的方式优化 说明：基本上用于优化的提示（hint）都是一些配置项，map-join 除外，需要具体在 HQL直接指定。2.2.4 physical plan 的生成 根据上一步的结果，分解成一些 map/reduce 操作，并将最终结果（即一些 plan 的 xml文件）写入到 hdfs。这里给出一个论文4中的例子：FROM(SELECT a.status,b.school,

12、b.gender FROM status_updates a JOIN profiles b ON(a.userid=b.userid AND a.ds=2009-03-20)subq1 INSERT OVERWRITE TABLE gender_summary PARTITION(ds=2009-03-20)SELECT subq1.gender,COUNT(1)GROUP BY subq1.gender INSERT OVERWRITE TABLE school_summary PARTITION(ds=2009-03-20)SELECT subq1.school,COUNT(1)GROU

13、P BY subq1.school 百度在线网络技术（北京）有限公司 图表 6 有 3 个 job 的多表插入查询的 query-plan(1)百度在线网络技术（北京）有限公司 图表 7 有 3 个 job 的多表插入查询的 query-plan(2)4 简单说明 map1+reduce1 将生成的 数据分别写入 两个临时的 hdfs 文件 tmp1 和 tmp2，map2+reduce2 和 map3+reduce3 就需要等待 tmp1 和 tmp2 的输出才能运行。一些理解和疑问 1、为什么 map-reduce1 中会放入 GroupByOperator 和 FileSinkOpera

14、tor？A：是 predicate pushdown 的结果 2、sink 不知道什么意思？A：我理解成 map/reduce 中 emit 函数的操作 3、中间 selectOperator 和 JoinOperator 操作分成了两步 A：应该为了逻辑上的分开处理 hive 计划 Hive 使用了 rule-based 的优化方案，简单但不够优秀。后期计划是建立 cost-based 的优化方案。百度在线网络技术（北京）有限公司 2.3 执行引擎(Execution Engine)根据 job 间的依赖的顺序执行任务。一个 mapreduce-job 首先是被编写成一个 plan.xml文

15、件，运行时先解析 plan.xml，然后用 hadoop 运行。3 其他说明及优化 3.1 数据模型 Hive 中包含以下数据模型：Table，External Table，Partition，Bucket。1、Hive 中的 Table 和数据库中的 Table 在概念上是类似的，每一个 Table 在 Hive 中都有一个相应的目录存储数据。例如，一个表 pvs，它在 HDFS 中的路径为：/wh/pvs，其中，wh 是在 hive-site.xml 中由$hive.metastore.warehouse.dir 指定的数据仓库的目录，所有的 Table 数据（不包括 External T

16、able）都保存在这个目录中。2、External Table 指向已经在 HDFS 中存在的数据，可以创建 Partition。它和 Table 在元数据的组织上是相同的，而实际数据的存储则有较大的差异。a)Table 的创建过程和数据加载过程（这两个过程可以在同一个语句中完成），在加载数据的过程中，实际数据会被移动到数据仓库目录中；之后对数据对访问将会直接在数据仓库目录中完成。删除表时，表中的数据和元数据将会被同时删除。b)External Table 只有一个过程，加载数据和创建表同时完成（CREATE EXTERNAL TABLE LOCATION），实际数据是存储在 LOCATION

17、 后面指定的 HDFS 路径中，并不会移动到数据仓库目录中。当删除一个 External Table 时，仅删除元数据。3、Partition 对应于数据库中的 Partition 列的密集索引，但是 Hive 中 Partition 的组织方式和数据库中的很不相同。在 Hive 中，表中的一个 Partition 对应于表下的一个目录，所有的 Partition 的数据都存储在对应的目录中。例如：pvs 表中包含 ds 和 city 两 个 Partition，则 对 应 于 ds=20090801,ctry=US 的 HDFS 子 目 录 为：/wh/pvs/ds=20090801/ctr

18、y=US；对应于 ds=20090801,ctry=CA 的 HDFS 子目录为；/wh/pvs/ds=20090801/ctry=CA。PARTITIONED BY 4、Buckets 对指定列计算 hash，根据 hash 值切分数据，目的是为了并行，每一个 Bucket 对应一个文件。如将 user 列分散至 32 个 bucket，首先对 user 列的值计算 hash，对应 hash 值为 0 的 HDFS 目录为：/wh/pvs/ds=20090801/ctry=US/part-00000；hash 值 为 20 的 HDFS 目 录 为：/wh/pvs/ds=20090801/c

19、try=US/part-00020。CLUSTERED BY 关于 HQL 语言使用以及其他 hive 内容，见3 3.2 功能及优化 3.2.1 PARTITION 功能 百度在线网络技术（北京）有限公司 除了在创建 table 时指定 partition，用户可以用 ALTER TABLE ADD PARTITION 来向一个表中增加分区。当分区名是字符串时加引号。也可以用 ALTER TABLE DROP PARTITION 来删除分区。分区的元数据和数据将被一并删除。借鉴 hive 使用的是 Range-partition，也可以参照 MYSQL 的 LIST-partition，也就

20、是将 partition表达式”ds=2010-01-01”的等式表达式变成任意的函数表达式 in_list(list_id)，参考6”7”。3.2.2 JOIN 1、Hive 只支持等值连接（equality joins）、外连接（outer joins）和 left semi joins。Hive 不支持所有非等值的连接，因为非等值连接非常难转化到 map/reduce 任务。join 的实现见图表 8，另外 Hive 支持多于 2 个表的连接。join LEFT/RIGHT OUTER：一定输出左边/右边的每一行对应的结果 left semi join 用于实现 a.key in sel

21、ect key from table b(即 in/exist 功能)2、多个表的 join：多表 join 时，每次 map/reduce 任务的逻辑是这样的：reducer 会缓存 join 序列中除了最后一个表的所有表的记录，再通过最后一个表将结果序列化到文件系统。这一实现有助于在 reduce 端减少内存的使用量。实践中，应该把最大的那个表写在最后（否则会因为缓存浪费大量内存）。多个表的 join key 是同一个时，join 会被转化为单个 map/reduce 任务：Reduce 端会缓存 a 表和 b 表的记录，然后每次取得一个 c 表的记录就计算一次 join 结果 不同 jo

22、in key 时，会被转化为多个 map/reduce 任务：第一次缓存 a 表，用 b 表序列化；第二次缓存第一次 map/reduce 任务的结果，然后用 c 表序列化 3、MAP-JOIN：允许在 map 阶段进行 join 操作，即需要在 map 时先加载一个 join 的数据到内存，然后直接过滤这片 buff，输出结果。实现 i.SELECT/*+MAPJOIN(t2)*/t1.c1,t2.c1 FROM t1 JOIN t2 ON(t1.c2=t2.c2);ii.通过打 tag 来实现同 key 的 join（如果多个 join 都一样的 key，就打多个 tag）iii.小文件被

23、复制多分到每个 split 后，优化 map-join（指定 size 和 num rows），见图表 9 优化 i.bucket-join：在特殊情况下，对大文件 a，小文件 b 进行 bucket 分桶（b 文件不是很小的时候），减少每个 a-split 上都需要一个 b（现在只需要一个 b-bucket 就可以了，但需要按 key 排序分桶的），见图表 10 ii.sort-join:当文件都比较大，可以边读边扔（因为是排序的，有点像 merge sort），就可以处理大文件了。iii.hash-join:在 map 端设计一个 hash，当达到一定大小（比如 50%hash 满），进行

24、一次计算输出 4、其他 join skew-join：针对不清楚 A 还是 B 的某个 key 的 size 小（或者交替），可以的方法有：将 A join B 时 A 的 key的 size 大的保存输出到另一个结果（生成第二个 mapreduce来将 B 的 key放 buf），见图表 12 百度在线网络技术（北京）有限公司 图表 8 join 实现8 图表 9 普通 map-join 实现9 图表 10 bucket map-join9 百度在线网络技术（北京）有限公司 图表 11 sort-based map-join9 图表 12 skew map-join9 3.2.3 GROUP

25、 BY group-by 的实现过程见图表 13。1、局部聚合：a)sort-based(combiner)：一般的 group by 都可以先在 map 端做 combiner 操作(如count 等函数)b)Hash-based：在 map 阶段通过保存 hash 来进行早期的聚合操作（类似 combine，但粒度更小）。配置变量为：hive.map.aggr.hash.percentmemory 2、两次 job（负载均衡）：为了减少一个因 key 分布不均导致某些 key 数据太多，可以要求生成两个 MR-job：第一阶段，随机分布 key（或者选取其他列的分桶方式），进行一次 Gro

26、up-by 后；第二次进行 reducer 时，单个 key 的 rownum 一般就不会很大了。两个百度在线网络技术（北京）有限公司 配置变量为：hive.mapjoin.size.key，hive.mapjoin.cache.numrows 图表 13 group-by 实现8 3.2.4 SORT BY 由于 reduce 是按照 key 分桶排序，当 sort by 为多列时，可能会出现相同第一个列的会在不同 reducer 中（因为多个列总的作为一个 key）。解决方法是使用 DISTRIBUTE BY，指定需要划分在一个 reducer 中的那些列（或叫做子 key）。也就是 pa

27、rtitioner-key 和 sort-key 不相同。3.3 存储格式 hive 允许多种 on-disk 格式(可自定义)：1、file format：row-based，column-based，block-based 2、raw format：text-based，binary-based，custom-based 3、index format？多种 in-memory 格式(可自定义)：Integer，LazyInteger，String，Text 4 参考 1 C.Olston,B.Reed,U.Srivastava,R.Kumar,and A.Tomkins,“Pig lati

28、n,”Proceedings of the 2008 ACM SIGMOD international conference on Management of data-SIGMOD 08,2008,p.1099.2 R.Pike,S.Dorward,and R.Griesemer,Quinlan,“Interpreting the data:Parallel analysis with sawzall,”Scientific Programming Journal,Special Issue on Grids and Worldwide Computing Programming Model

29、s and In,vol.frastructu,p./labs.百度在线网络技术（北京）有限公司 3 M.Isard,M.Budiu,Y.Yu,A.Birrell,and D.Fetterly,“DryadLINQ:A System for General-Purpose Distributed Data-parallel Computing Using a High-Level Language,”ACM SIGOPS Operating Systems Review,vol.41,Jun.2009,p.59.4 A.Thusoo,J.S.Sarma,N.Jain,Z.Shao,P.Chak

30、ka,N.Zhang,S.Antony,and H.Liu,“Hive A Petabyte Scale Data Warehouse Using Hadoop,”Architecture.5 J.Sichi,Facebook Hive team,“Hive/HBase Integration.”6“MYSQL-reference manual.”7“Partition(database).”8 N.Z.Ashish Thusoo,Raghotham Murthy,Joydeep Sen Sarma,Zheng Shao,Namit Jain,Prasad Chakka,Suresh Anth

31、ony,Hao Liu,“Hive-A Petabyte Scale Data Warehouse Using Hadoop,”2010.9 F.H.Team,“Hive New Features and API.”web page(对应前面的 paper 名字的网页)Hive/HBase Integration http:/ Hive-A Petabyte Scale Data Warehouse Using Hadoop http:/ Hive New Features and API http:/ MYSQL-reference manual http:/ Partition(database)http:/en.wikipedia.org/wiki/Partition_(database)附加说明 一些相关的 ppt 和 paper 在：http:/wiki.apache.org/hadoop/Hive/Presentations