hive优化分享

粘贴一下我在部门中的一次hive优化的分享。

简述

hive构建在hadoop基础上，利用分布式存储，通过mr引擎实现对大数据的计算。MR会频繁地读写磁盘而且MR任务的启动成本很高。对于hive优化显得尤为重要。而优化的核心就是更好地利用hadoop的分布式特性和hive的有点。本篇从IO、参数设置、案例实战来说明如何优化我们的hive。受限于个人能力，如有不足之处，还望指出，一起沟通讨论。

1、IO

　　A、通过列裁剪，只读取需要的列[对select * 的做法应进行严格要求，甚至禁止]

　　B、 join操作前，应提前处理表，只读取需要的列，并对数据进行梳理。

　　   原：select A.char1,B.char2

　　     from A join B

　　           on A.id=B.id

　　      where A.disabled=0 and B.disabled=0

　　  优：select A.char1,B.char2

　　        from

　　          (

　　            select id,char1

　　      from a

　　            where a.disabled=0

　　           ) A

　　             join

　　        (

　　           select id,char2

　　        from b

　　            where b.disabled=0

　　         ) B

　　            on A.id=B.id

　　C、选择合适的存储类型

　　　　建议使用orc存储格式，作为默认设置[支持切片，有压缩，提高查询速度]。创建表时，指定stored as orcfile

　　D、走分区[可选]

　　　　减少读取更多的分区文件，极大地有帮助！处理分区时，不可对分区字段进行操作，否则将会导致全表扫描。

2、join

　　A、left semi join取代in

　　   根据测试，对于在大数据量中查询某些取值时，有效提升查询效率。[测试：从erp_fct_order表中，查找若干数据。left semi join 使用了100'+,in耗时300'+]

　　 B、map join

　　 当小表与大表进行关联时，通过map join，使左表进入内存，与右表进行关联。这个阶段不涉及reduce。

　　 通过设置set hive.auto.convert.join=true;[默认]

　　     set hive.mapjoin.smalltable.filesize=2500000;[25M,通过配置该属性来确定使用该优化的表的大小，如果表的大小小于此值就会被加载进内存中]

3、设置执行参数

reducer数量

　　我们执行语句时，总会看到这个：

　　 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 mapreduce.job.reduces=<number>

　　Starting Job = job_1515122600431_869586, Tracking URL = http://hadoop04:8088/proxy/application_1515122600431_869586/

　　上面其实就是，hadoop确定reduce的数量的方式。

　　 N=min(hive.exec.reducers.max,总数据量大小/hive.exec.reducers.bytes.per.reducer),默认情况下，hive.exec.reducers.max=999，hive.exec.reducers.bytes.per.reducer=1G;

　　我们也可以强制指定reduce数量，mapreduce.job.reduces=N。

　　需要注意的是，有的reduce为0，有的只为1。

　　为0，即为该任务只有map，没有reduce。

　　    为1，即为全局，如order by、count、笛卡尔积。

mapper的数量

　　map端处理文件时，将对文件进行拆分成若干个task数。而如何拆分，这本身和文件的存储类型相关。hive系统默认的类型为org.apache.hadoop.hive.ql.io.CombineHiveInputFormat[set hive.input.format;]，当我们不指定存储类型时，hive将根据文件本身设置对应的存储类型。如：org.apache.hadoop.mapred.TextInputFormat。

　　继承了FileInputFormat的切分算法：splitSize = max{split.size,min{goalSize,blockSize}}

　　若格式为:HiveInputFormat

　　split.size:mapred.min.split.size,默认为1。切片最小大小，没有什么能比这个更小的了。

　　  goalSize:目标大小，为用户设置的map task数与实际大小决定。goalSize=总大小/设置的mapred.map.tasks的大小[默认为2]

　　blockSize:块大小[set dfs.block.size]，当前我们系统的块为：128M

　　所以，当我们没有设置最小的切片大小时，切片的大小，取决于文件大小、设置的mapred.map.tasks数、块大小.

　　 例如：一个文件如果是300M，则切片大小为：splitSize=max{1,min{300/2,128}}=128M,则3个map对应的文件大小为：128,128,46

　　 若格式为：TextInputFormat：

　　 split.size:mapred.max.split.size，[当前系统为256M],则splitSize为max{256,min{300/2,128}}=256,则只有一个map.

小结

　　mapper的数量合适，单个mapper处理的数据量合适；拆分大小尽量与block数据块相同，避免一个拆分块大小有多个 hdfs 块，且位于不同数据节点，从而降低网络 IO。如果想减少mapper的数据量，一般只要将mapred.max.split.size设置的更大就可以了。

合并小文件

　　hive基于NameNode进行调度管理和任务分发，NameNode本身是有瓶颈的，有固定的大小。在hadoop任务中，NameNode加载必要的信息，其中包括各个文件的元数据（位置、大小、块信息等）。一般情况下，当文件数达到1000W时，所占用的NameNode内存达到3G，将带来性能下降。Hadoop本身对大文件具有很高的处理能力，但在mr任务过程中，map任务和reduce任务都会带来小文件的问题。当小文件作为输入，由map来进行处理时，启用map带来的时间比处理更耗时。此外，reduce数量多的话，产生的小文件也会增多。

　　输入时合并

　　set mapred.max.split.size=256000000;-- 每个Map最大输入大小，决定合并后的文件数

　　set mapred.min.split.size.per.node=100000000; -- 一个节点上split的至少的大小 ，决定了多个data node上的文件是否需要合并 [当前设置为:1，默认设置。设置该参数，一定要设置下方的rack参数，否则报错。]

　　set mapred.min.split.size.per.rack=100000000; -- 一个交换机下split的至少的大小，决定了多个交换机上的文件是否需要合并 [当前设置为:1，默认设置]

　　set hive.input.format=org.apache.hadoop.hive.ql.io.CombineHiveInputFormat; -- 执行Map前进行小文件合并[经过测试，设置上面的参数就能够让小文件进行合并了。]

　　输出时合并

　　set hive.merge.mapfiles=true; -- 在map-only job后合并文件，默认true

　　set hive.merge.mapredfiles=true; -- 在map-reduce job后合并文件，默认false

　　set hive.merge.size.per.task=256000000; -- 合并后每个文件的大小，默认256000000

　　set hive.merge.smallfiles.avgsize=16000000; -- 平均文件大小,当结果文件小于该值时，将进行合并

数据倾斜

原因：当我们进行join操作时，会在map端根据key的hash值，shuffle到某一个reduce上去，在reduce端做join连接操作，内存中缓存join左边的表，遍历右边的表，一次做join操作。所以在做join操作时候，将数据量多的表放在join的右边。当数据量比较大，并且key分布不均匀，大量的key都shuffle到一个reduce上了，就出现了数据的倾斜。

场景

key为空:当join的key存在空值时，这些值会被hash到同一个reduce中，导致数据倾斜。遇到这样的情况，建议区分空和非空

count(distinct):数据聚合类的操作sum、count，因为已经在map端做了聚合操作了，到reduce端的数据相对少一些[主要设置参数为:1.set hive.map.aggr=true;默认值，是否在map端进行聚合;2.set hive.groupby.mapaggr.checkinterval = 100000;默认值，在 Map 端进行聚合操作的条目数目],不会出现这个问题。当distinct时，会对group by后的字段和distinct的字段进行hash，如果存在大量重复值，则会导致数据倾斜。

遇到倾斜时set hive.groupby.skewindata = true;[负载均衡，默认为false。查询计划会有两个 MR Job。第一个 MR Job 中，Map 的输出结果集合会随机分布到 Reduce 中，每个 Reduce 做部分聚合操作，并输出结果，这样处理的结果是相同的 Group By Key 有可能被分发到不同的 Reduce 中，从而达到负载均衡的目的；第二个 MR Job 再根据预处理的数据结果按照 Group By Key 分布到 Reduce 中（这个过程可以保证相同的 Group By Key 被分布到同一个 Reduce 中），最后完成最终的聚合操作]

常用参数设置及说明

--类型：输入时合并小文件

　　set hive.input.format=org.apache.hadoop.hive.ql.io.CombineHiveInputFormat; ----可在map端进行合并，减少小文件

　　set mapred.max.split.size=256000000;  ----设置一个合理的数值，最大的切片数据。影响map的个数

　　set mapred.min.split.size.per.node=134217728;  ----一个节点上的最小切片大小。设置为block大小

　　set mapred.min.split.size.per.rack=256000000;  ----一个交换机上的最小切片大小

--类型：资源占用量

　　set mapreduce.map.memory.mb=5120;

　　set mapreduce.reduce.memory.mb=5120;

--类型：设置最大值

　　set mapreduce.job.reduces=80;  ----用来对reduer数量进行控制，但设置的过大，不设置效果更好

　　set hive.exec.reducers.max=80; ----最大的reduce数

--类型：mapjoin

　　set hive.auto.convert.join=true;

　　set hive.mapjoin.smalltable.filesize=2500000;

--类型：输出时合并

　　set hive.merge.mapfiles=true; -- 在map-only job后合并文件，默认true

　　set hive.merge.mapredfiles=true; -- 在map-reduce job后合并文件，默认false

　　set hive.merge.size.per.task=256000000; -- 合并后每个文件的大小，默认256000000

　　set hive.merge.smallfiles.avgsize=16000000; -- 平均文件大小,当结果文件小于该值时，将进行合并

--类型：并发及并发数

　　set hive.exec.parallel=true;

　　set hive.exec.parallel.thread.number=16;

补充

在写sql的过程中，有时候会用到排序。但是在hive中用order by排序，将带来一个问题。只有一个reduce。往往带来性能的较差。Hive为查询提供了其他排序方式。

　　Order by：全局，一个reduce【不建议】

　　Distribute by ：map的输出排序，排序后，将对应的key通过hash放入到reduce中

　　Sort by：局部排序，在每个reduce中实现排序。当reduce数只有一个时，sort by和order by结果没有什么差异【建议】

Cluster by：分桶中用到，用来指定分桶的字段。cluster by  key = distribute by key sort by key desc  ，是一个简写。cluster by只能降序排序。

原创博客，转载请注明出处！欢迎邮件沟通：[e-mail]