Hive优化总结

1.Hive表优化

1.1 分区

分区是以字段的形式在表结构中存在，通过describe table命令可以查看到字段存在， 但是该字段不存放实际的数据内容，仅仅是分区的表示（伪列）。

1.1.1静态分区

create table if not exists tablename(id int,name string,tel string)
partitioned by(dt string)
row format delimited
fields terminated by '\t'
stored as textfile;

1.1.2动态分区

使用动态分区前，需要配置以下参数：

hive.exec.dynamic.partition：在DML/DDL中是否支持动态分区，配置为true

hive.exec.dynamic.partition.mode：默认strict，在strict模式下，动态分区的使用必须在一个静态分区确认的情况下，其他分区可以是动态，可以考虑配置成nostrict。

1.2 分桶

使用分桶前，需要配置以下参数：

hive.enforce.bucketing：数据分桶是否被强制执行，默认false，如果开启，则写入table数据时会启动分桶。

hive.enforce.sorting：开启强制排序时，插数据到表中会进行强制排序，默认false；

对于每一个表（table）或者分区， Hive可以进一步组织成桶，也就是说桶是更为细粒度的数据范围划分。Hive也是 针对某一列进行桶的组织。Hive采用对列值哈希，然后除以桶的个数求余的方式决定该条记录存放在哪个桶当中。

2. Hive查询优化

2.1 join优化

2.1.1 和数据倾斜有关的优化：

hive.optimize.skewjoin：是否优化数据倾斜的 Join，对于倾斜的 Join 会开启新的 Map/Reduce Job 处理，默认值为false。

hive.skewjoin.key    倾斜键数目阈值，超过此值则判定为一个倾斜的 Join 查询。默认值为100000

2.1.2和mapjoin有关的优化：

hive.auto.convert.join：是否根据输入小表的大小，自动将 Reduce 端的 Common Join 转化为 Map Join，从而加快大表关联小表的 Join 速度。

hive.mapjoin.smalltable.filesize：输入表文件的mapjoin阈值，如果输入文件的大小小于该值，则试图将普通join转化为mapjoin，默认25MB；

上面的两个参数是自动启用mapjoin，也可以手动启动mapjoin。

2.1.3 和where有关的优化

由于join操作是在where操作之前执行，所以当你在执行join时，where条件并不能起到减少join数据的作用。

优化前：

select o.cid,c.id from order o
join customer c
on o.cid=c.id where o.dt='2013-12-12';

优化后：

select o.cid,c.id from (select cid from order where dt='2013-12-12') o
join customer c
on o.cid=c.id;

优化后会在map端通过where语句过滤掉一些数据，提高了效率

2.2 group by优化

hive.groupby.skewindata：group by操作是否允许数据倾斜，默认是false，当设置为true时，执行计划会生成两个map/reduce作业，第一个MR中会将map的结果随机分布到reduce中，达到负载均衡的目的来解决数据倾斜。

hive.groupby.mapaggr.checkinterval：map端做聚合时，group by 的key所允许的数据行数，超过该值则进行分拆，默认是100000；

2.3 count distinct优化

优化前：

select count(distinct id) from tablename;

在数据量大时，可以使用如下优化：

select count(1) from (selectdistinct id from tablename) tmp;

select count(1) from (select id from tablename group by id) tmp;

优化前查询只会启动一个job来完成，完成数据去重和计数时都只在一个reduce端（即便通过mapred.reduce.tasks设置reduce的数目为多个，但实际执行时仍然只有一个）进行，优化后会启动两个job来完成，同时可以通过mapred.reduce.tasks设定更多的reduce数目，所以适合在数据量很大的情况下使用，因为初始化一个job时花费的时间也会很长。

看一个更复杂的例子：

优化前：

select a,sum(b),count(distinct c),count(distinct d)
from test group by a;

优化后：

from (
select a,0 as b,c,null as d from test group by a,c
union all
select a,0 as b,null as c,d from test group by a,d
union all
select a,b,null as c,null as d from test
) tmp1 group by a;

3. Hive job优化

3.1 并行化执行

每个查询被Hive转化为多个阶段，有些阶段的关联性不大，则可以并行化执行，减少执行时间。

使用并行化执行前，需要配置以下参数：

hive.exec.parallel：hive的执行job是否并行执行，默认不开启false，在很多操作如join时，子查询之间并无关联可独立运行，这种情况下开启并行运算可以大大加速；

hvie.exec.parallel.thread.number：并行运算开启时，允许多少作业同时计算，默认是8；

3.2 本地化执行

使用本地化执行前，需要配置以下参数：

hive.exec.mode.local.auto：是否由hive决定自动在local模式下运行，默认是false。

配置了上面参数后，该job还必须满足以下条件，才能真正的使用本地模式：

1. job的输入数据大小必须小于参数

hive.exec.mode.local.auto.inputbytes.max（默认128MB)

2. job的map数必须小于参数：

hive.exec.mode.local.auto.tasks.max（默认4），我使用的hive版本为1.2.1，貌似已经取消了该配置

3. job的reduce数必须为0或1。

3.3 合并小文件

1. 合并输入小文件

set mapred.max.split.size=256000000;  #每个Map最大输入大小

set mapred.min.split.size.per.node=100000000; #一个节点上split的至少的大小 

set mapred.min.split.size.per.rack=100000000; #一个交换机下split的至少的大小

set hive.input.format=org.apache.hadoop.hive.ql.io.CombineHiveInputFormat;  #执行Map前进行小文件合并

2. 合并输出小文件

hive.merge.mapfiles：在只有map的作业结束时合并小文件，默认开启true；

hive.merge.mapredfiles：在一个map/reduce作业结束后合并小文件，默认不开启false；

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

3.4 JVM重利用

Hadoop中有个参数是mapred.job.reuse.jvm.num.tasks(hadoop2 为：mapreduce.job.jvm.numtasks)，默认是1，表示一个JVM上最多可以顺序执行的task数目（属于同一个Job）是1。也就是说一个task启一个JVM。

适当地增大该值对于有较多task的job是分厂有意义的，但是也有个缺点就是：开启JVM重用将会一直占用使用到的task插槽，以便进行重用，直到任务完成后才能释放。如果某个“不平衡“的job中有几个reduce task 执行的时间要比其他reduce task消耗的时间多得多的话，那么保留的插槽就会一直空闲着却无法被其他的job使用，直到所有的task都结束了才会释放。

3.5 压缩数据

1.中间结果压缩

中间结果压缩就是处理hive查询的多个job之间的数据，对于中间压缩，最好选择一个节省CPU耗时的压缩方式。

hive.exec.compress.intermediate：决定查询的中间 map/reduce job （中间 stage）的输出是否为压缩格式。 

hive.intermediate.compression.codec：中间 map/reduce job 的压缩编解码器的类名（一个压缩编解码器可能包含多种压缩类型），该值可能在程序中被自动设置。           

hive.intermediate.compression.type：中间 map/reduce job 的压缩类型，如 "BLOCK""RECORD"。  

我自己使用的是hive1.2.1版本，但是没有找到hive.intermediate.compression.codec和hive.intermediate.compression.type这两个配置项，如果有朋友跟我的不一样，可以跟我说下。

2.最终结果压缩

hive.exec.compress.output：决定查询中最后一个 map/reduce job 的输出是否为压缩格式。   

mapred.output.compression.codec：最后结果的压缩编解码器的类名

mapred.output.compression.type：最后结果的压缩类型