hive sql优化和shuffle过程优化

hive sql优化

sort by代替order by

HiveSQL中的order by与其他SQL方言中的功能一样，就是将结果按某字段全局排序，这会导致所有map端数据都进入一个reducer中，在数据量大时可能会长时间计算不完。

如果使用sort by，那么还是会视情况启动多个reducer进行排序，并且保证每个reducer内局部有序。为了控制map端数据分配到reducer的key，往往还要配合distribute by一同使用。如果不加distribute by的话，map端数据就会随机分配到reducer。举个例子，假如要以UID为key，以上传时间倒序、记录类型倒序输出记录数据：


select uid,upload_time,event_type,record_data
from calendar_record_log
where pt_date >= 20190201 and pt_date <= 20190224
distribute by uid
sort by upload_time desc,event_type desc;

group by代替distinct

当要统计某一列的去重数时，如果数据量很大，count(distinct)就会非常慢，原因与order by类似，count(distinct)逻辑只会有很少的reducer来处理。这时可以用group by来改写：

select count(1) from (
select uid from calendar_record_log
where pt_date >= 20190101
group by uid
) t;

但是这样写会启动两个MR job（单纯distinct只会启动一个），所以要确保数据量大到启动job的overhead远小于计算耗时，才考虑这种方法。当数据集很小或者key的倾斜比较明显时，group by还可能会比distinct慢。那么如何用group by方式同时统计多个列？下面是解决方法：

select t.a,sum(t.b),count(t.c),count(t.d) from (
select a,b,null c,null d from some_table
union all
select a,0 b,c,null d from some_table group by a,c
union all
select a,0 b,null c,d from some_table group by a,d
) t;

group by配置调整

map端预聚合

group by时，如果先起一个combiner在map端做部分预聚合，可以有效减少shuffle数据量。预聚合的配置项是hive.map.aggr，默认值true，对应的优化器为GroupByOptimizer，简单方便。通过hive.groupby.mapaggr.checkinterval参数也可以设置map端预聚合的行数阈值，超过该值就会分拆job，默认值100000。

倾斜均衡配置项

group by时如果某些key对应的数据量过大，就会发生数据倾斜。Hive自带了一个均衡数据倾斜的配置项hive.groupby.skewindata，默认值false。其实现方法是在group by时启动两个MR job。第一个job会将map端数据随机输入reducer，每个reducer做部分聚合，相同的key就会分布在不同的reducer中。第二个job再将前面预处理过的数据按key聚合并输出结果，这样就起到了均衡的效果。但是，配置项毕竟是死的，单纯靠它有时不能根本上解决问题，因此还是建议自行了解数据倾斜的细节，并优化查询语句。

利用map join特性

map join特别适合大小表join的情况。Hive会将build table和probe table在map端直接完成join过程，消灭了reduce，效率很高。

select  a.event_type,b.upload_time
from calendar_event_code a
inner join (
select event_type,upload_time from calendar_record_log
where pt_date = 20190225
) b on a.event_type < b.event_type;

上面的语句中加了一条map join hint，以显式启用map join特性。早在Hive 0.8版本之后，就不需要写这条hint了。map join还支持不等值连接，应用更加灵活。 map join的配置项是hive.auto.convert.join，默认值true，对应逻辑优化器是MapJoinProcessor。还有一些参数用来控制map join的行为，比如hive.mapjoin.smalltable.filesize，当build table大小小于该值就会启用map join，默认值25000000（25MB）。还有hive.mapjoin.cache.numrows，表示缓存build table的多少行数据到内存，默认值25000。

利用map join特性

map join特别适合大小表join的情况。Hive会将build table和probe table在map端直接完成join过程，消灭了reduce，效率很高。

select  a.event_type,b.upload_time
from calendar_event_code a
inner join (
select event_type,upload_time from calendar_record_log
where pt_date = 20190225
) b on a.event_type < b.event_type;

倾斜均衡配置项

这个配置与上面group by的倾斜均衡配置项异曲同工，通过hive.optimize.skewjoin来配置，默认false。

如果开启了，在join过程中Hive会将计数超过阈值hive.skewjoin.key（默认100000）的倾斜key对应的行临时写进文件中，然后再启动另一个job做map join生成结果。通过hive.skewjoin.mapjoin.map.tasks参数还可以控制第二个job的mapper数量，默认10000。再重复一遍，通过自带的配置项经常不能解决数据倾斜问题。join是数据倾斜的重灾区，后面还要介绍在SQL层面处理倾斜的各种方法。

优化SQL处理join数据倾斜

空值或无意义值

这种情况很常见，比如当事实表是日志类数据时，往往会有一些项没有记录到，我们视情况会将它置为null，或者空字符串、-1等。如果缺失的项很多，在做join时这些空值就会非常集中，拖累进度。

因此，若不需要空值数据，就提前写where语句过滤掉。需要保留的话，将空值key用随机方式打散，例如将用户ID为null的记录随机改为负值：

select a.uid,a.event_type,b.nickname,b.age
from (
select(case when uid is null then cast(rand()*-10240 as int) else uid end) as uid,event_type from calendar_record_log
where pt_date >= 20190201
) a left outer join (
select uid,nickname,age from user_info where status = 4
) b on a.uid = b.uid;

单独处理倾斜key

这其实是上面处理空值方法的拓展，不过倾斜的key变成了有意义的。一般来讲倾斜的key都很少，我们可以将它们抽样出来，对应的行单独存入临时表中，然后打上一个较小的随机数前缀（比如0~9），最后再进行聚合。SQL语句与上面的相仿，不再赘述。

不同数据类型

这种情况不太常见，主要出现在相同业务含义的列发生过逻辑上的变化时。举个例子，假如我们有一旧一新两张日历记录表，旧表的记录类型字段是(event_type int)，新表的是(event_type string)。为了兼容旧版记录，新表的event_type也会以字符串形式存储旧版的值，比如'17'。当这两张表join时，经常要耗费很长时间。其原因就是如果不转换类型，计算key的hash值时默认是以int型做的，这就导致所有“真正的”string型key都分配到一个reducer上。所以要注意类型转换：

select a.uid,a.event_type,b.record_data
from calendar_record_log a
left outer join (
select uid,event_type from calendar_record_log_2
where pt_date = 20190228
) b on a.uid = b.uid and b.event_type = cast(a.event_type as string)
where a.pt_date = 20190228;

shuffle优化

1.map端

增大缓冲区大小:100M改为200M

增大缓冲区溢写百分比:0.8改为0.9

2.reduce端

合理设置map和reduce数量:太多会导致争抢资源造成处理超时,太少会导致task等待延长处理时间

增加每个reduce拿取map数据的并行度

3.I/O处理

压缩数据减少io时间

map输出时采用snappy压缩格式