六、Hive解决数据倾斜具体方法

6.1 场景

6.1.1 group by

注：group by 优于distinct group
情形：group by 维度过小，某值的数量过多
后果：处理某值的reduce非常耗时
解决方式：采用sum() group by的方式来替换count(distinct)完成计算。

6.1.2 count(distinct)

count(distinct xx)
情形：某特殊值过多
后果：处理此特殊值的reduce耗时；只有一个reduce任务
解决方式：count distinct时，将值为空的情况单独处理，比如可以直接过滤空值的行，在最后结果中加1。如果还有其他计算，需要进行group by，可以先将值为空的记录单独处理，再和其他计算结果进行union。

6.1.3 join

情形1：小表与大表join，但较小表key集中
后果：shuffle分发到某一个或几个Reducer上的数据量远高于平均值。想象极端情况，小表的join列全部为一个值，那么shuffle后全部到一个Reducer节点，其他节点无负载。这就是极端的数据倾斜了。
解决方式：mapjoin

情形2：大表与大表join，但是分桶的判断字段0值或空值过多
后果：这些空值/0值都由一个Reducer处理，非常慢
解决方式：把空值的key变成一个字符串加上随机数，把倾斜的数据分到不同的reduce上，由于null值关联不上，处理后并不影响最终结果。

6.1.4 不同数据类型关联产生数据倾斜

情形：比如用户表中user_id字段为int，log表中user_id字段既有string类型也有int类型。当按照user_id进行两个表的Join操作时。
后果：处理此特殊值的reduce耗时；只有一个reduce任务
默认的Hash操作会按int型的id来进行分配，这样会导致所有string类型id的记录都分配到一个Reducer中。
解决方式：把数字类型转换成字符串类型

select * from users a
  left outer join logs b
  on a.usr_id = cast(b.user_id as string)

6.2 调优

6.2.1 hive.map.aggr=true

# 开启map端combiner
set hive.map.aggr=true；

思想
开启map combiner。在map中会做部分聚集操作，效率更高但需要更多的内存。
点评
假如map各条数据基本上不一样, 聚合没什么意义，做combiner反而画蛇添足,hive里也考虑的比较周到通过参数：
hive.groupby.mapaggr.checkinterval = 100000 (默认)
hive.map.aggr.hash.min.reduction=0.5(默认)

6.2.2 hive.groupby.skewindata=true

# 开启数据倾斜时负载均衡
set hive.groupby.skewindata=true;

思想：就是先随机分发并处理，再按照key group by来分发处理。

操作：当选项设定为true，生成的查询计划会有两个MRJob。

第一个MRJob 中，Map的输出结果集合会随机分布到Reduce中，每个Reduce做部分聚合操作，并输出结果，这样处理的结果是相同的GroupBy Key有可能被分发到不同的Reduce中，从而达到负载均衡的目的；

第二个MRJob再根据预处理的数据结果按照GroupBy Key分布到Reduce中（这个过程可以保证相同的原始GroupBy Key被分布到同一个Reduce中），最后完成最终的聚合操作。

点评：它使计算变成了两个mapreduce，先在第一个中在 shuffle 过程 partition 时随机给 key 打标记，使每个key 随机均匀分布到各个 reduce 上计算，但是这样只能完成部分计算，因为相同key没有分配到相同reduce上。

所以需要第二次的mapreduce,这次就回归正常 shuffle,但是数据分布不均匀的问题在第一次mapreduce已经有了很大的改善，因此基本解决数据倾斜。因为大量计算已经在第一次mr中随机分布到各个节点完成。

6.2.3 Join

6.2.3.1 关于驱动表的选取

选用join key分布最均匀的表作为驱动表。

6.2.3.2 做好列裁剪和filter操作

以达到两表做join的时候，数据量相对变小的效果。

6.2.3.3 left semi join

6.2.3.4 大小表Join - MapJoin

思想
小表关联一个超大表时，容易发生数据倾斜，使用 MapJoin把小表全部加载到内存在map端进行join。如果需要的数据在 Map 的过程中可以访问到则不再需要Reduce。

实例分析
原始sql：

select c.channel_name,count(t.requesturl) PV
 from ods.cms_channel c
 join
 (select host,requesturl from  dms.tracklog_5min where day='20151111' ) t
 on c.channel_name=t.host
 group by c.channel_name
 order by c.channel_name;

以上为小表join大表的操作，可以使用mapjoin把小表c放到内存中处理，语法很简单只需要增加 /*+ MAPJOIN(小标) */，把需要分发的表放入到内存中。

select /*+ MAPJOIN(c) */
c.channel_name,count(t.requesturl) PV
 from ods.cms_channel c
 join
 (select host,requesturl from  dms.tracklog_5min where day='20151111' ) t
 on c.channel_name=t.host
 group by c.channel_name
 order by c.channel_name;

6.2.3.5 大表Join大表 - skewjoin

当key值都是有效值时可使用hive配置：

set hive.optimize.skewjoin=true;
指定是否开启数据倾斜的join运行时优化，默认不开启即false。

set hive.skewjoin.key=100000;
判断数据倾斜的阈值，如果在join中发现同样的key超过该值，则认为是该key是倾斜key。

默认100000。一般可以设置成处理的总记录数/reduce个数的2-4倍。

set hive.optimize.skewjoin.compiletime=true;
指定是否开启数据倾斜的join编译时优化，默认不开启即false。

具体来说，会基于存储在原数据中的倾斜key，来在编译时为导致倾斜的key单独创建执行计划，而其他key也有一个执行计划用来join。然后，对上面生成的两个join执行后求并集。因此，除非相同的倾斜key同时存在于这两个join表中，否则对于引起倾斜的key的join就会优化为map-side join。

此外，该参数与hive.optimize.skewjoin之间的主要区别在于，此参数使用存储在metastore中的倾斜信息在编译时来优化执行计划。如果元数据中没有倾斜信息，则此参数无效。一般可将这两个参数都设为true。如果元数据中有倾斜信息，则hive.optimize.skewjoin不做任何操作。

6.2.3.6 小结

以上方式，都是根据数据倾斜形成的原因进行的一些变化。要么将 reduce 端的隐患在 map 端就解决，要么就是对 key 的操作，以减缓reduce 的压力。了解了原因再去寻找解决之道就相对思路多了些，方法肯定不止这几种。

6.2.4 先group再count

能先进行 group 操作的时候先进行group操作，把 key 先进行一次 reduce，之后再进行 count 或者 distinct count 操作。

6.2.5 控制空值分布

将为空的key转变为字符串加随机数或纯随机数，将因空值而造成倾斜的数据分不到多个Reducer。

注：对于异常值如果不需要的话，最好是提前在where条件里过滤掉，这样可以使计算量大大减少

实践中，可以使用case when对空值赋上随机值。此方法比直接写is not null更好，因为前者job数为1，后者为2.

使用case when实例1：

select userid, name
fromuser_info a
join (
select case
when userid is null  then  cast (rand(47)* 100000 as int )
else userid
end
from user_read_log
) b
on a.userid = b.userid

使用case when实例2：

select
    '${date}' as thedate,
    a.search_type,
    a.query,
    a.category,
    a.cat_name,
    a.brand_id,
    a.brand_name,
    a.dir_type,
    a.rewcatid,
    a.new_cat_name,
    a.new_brand_id,
    f.brand_name as new_brand_name,
    a.pv,
    a.uv,
    a.ipv,
    a.ipvuv,
    a.trans_amt,
    a.trans_num,
    a.alipay_uv
from fdi_search_query_cat_qp_temp a
left outer join brand f
on
     f.pt='${date}000000'
    and case when a.new_brand_id is null then concat('hive',rand() ) else a.new_brand_id end = f.brand_id

如果上述的方法还不能解决，比如当有多个JOIN的时候，建议建立临时表，然后拆分HIVE SQL语句。

6.2.6 压缩

设置map端输出、中间结果压缩。（不完全是解决数据倾斜的问题，但是减少了IO读写和网络传输，能提高很多效率）

6.2.7 增加Reuducer个数

默认是由参数hive.exec.reducers.bytes.per.reducer来推断需要的Reducer个数。

可通过mapred.reduce.tasks控制，默认-