Hive数据倾斜

map/reduce程序执行时，reduce节点大部分执行完毕，但是有一个或者几个reduce节点运行很慢，导致整个程序的处理时间很长，这是因为某一个key的条数比其他key多很多（有时是百倍或者千倍之多），这条key所在的reduce节点所处理的数据量比其他节点就大很多，从而导致某几个节点迟迟运行不完，此称之为数据倾斜。

1.万能膏药：hive.groupby.skewindata=true
当选项设定为 true，生成的查询计划会有两个 MR Job。
　　第一个 MR Job 中，Map 的输出结果集合会随机分布到 Reduce 中，每个 Reduce 做部分聚合操作，并输出结果，这样处理的结果是相同的 Group By Key 有可能被分发到不同的 Reduce 中，从而达到负载均衡的目的
　　第二个 MR Job 再根据预处理的数据结果按照 Group By Key 分布到 Reduce 中（这个过程可以保证相同的 Group By Key 被分布到同一个 Reduce 中），最后完成最终的聚合操作。 

1.1.参数调优：hive.map.aggr=true. Map端部分聚合，相当于Combiner 

2. 大小表关联：
可以使用Map Join让小的维度表（1000条以下的记录条数）先进内存。在map端完成reduce.

3. 大表和大表关联：
把空值NULL的key变成一个字符串加上随机数，把倾斜的数据分到不同的reduce上，由于null值关联不上，处理后并不影响最终结果。例如Demo1.空值数据倾斜

4. count distinct大量相同特殊值：
count distinct时，将值为空的情况单独处理。如果是计算count distinct，可以不用处理，直接过滤，在最后结果中加1。如果还有其他计算，需要进行group by，可以先将值为空的记录单独处理，再和其他计算结果进行union。

Demo1.空值数据倾斜
场景：如日志中，常会有信息丢失的问题，比如全网日志中的user_id，如果取其中的user_id和bmw_users关联，会碰到数据倾斜的问题。
解决方法1： user_id为空的不参与关联

Select * From log a 
　　　Join bmw_users b On a.user_id is not nullAnd a.user_id = b.user_id
　Union all Select * from log a　where a.user_id is null;

解决方法2 ：赋予空值新的key值

Select * from log a 
　　left outer Join bmw_users b 
on case when a.user_id is null then concat(‘dp_hive’,rand()) 
　　else a.user_id end = b.user_id; 

结论：方法2比方法1效率更好，不但io少了，而且作业数也少了。
方法1的log读取两次，jobs是2。方法2的job数是1。这个优化适合无效id(比如-99,’’,null等)产生的倾斜问题。
把空值的key变成一个字符串加上随机数，就能把倾斜的数据分到不同的reduce上 ,解决数据倾斜问题。

Demo2.不同数据类型关联产生数据倾斜
场景：一张表s8的日志，每个商品一条记录，要和商品表关联。但关联却碰到倾斜的问题，s8的日志中有字符串商品id,也有数字的商品id,类型是string的，但商品中的数字id是bigint的。
问题原因：把s8的商品id转成数字id做hash来分配reduce，所以字符串id的s8日志，都到一个reduce上了，解决的方法验证了这个猜测。
解决方法：把数字类型转换成字符串类型

Select * from s8_log a
　Left outer join r_auction_auctions b
　　On a.auction_id = cast(b.auction_id as string);

Demo3.大表Join的数据偏斜
MapReduce编程模型下开发代码需要考虑数据偏斜的问题，Hive代码也是一样。数据偏斜的原因包括以下两点：
　　1. Map输出key数量极少，导致reduce端退化为单机作业。
　　2. Map输出key分布不均，少量key对应大量value，导致reduce端单机瓶颈。
Hive中我们使用MapJoin解决数据偏斜的问题，即将其中的某个小表（全量）分发到所有Map端的内存进行Join，从而避免了reduce。这要求分发的表可以被全量载入内存。
极限情况下，Join两边的表都是大表，就无法使用MapJoin。这种问题最为棘手，目前已知的解决思路有两种：
1. 如果是上述情况1，考虑先对Join中的一个表去重，以此结果过滤无用信息。
　　这样一般会将其中一个大表转化为小表，再使用MapJoin 。一个实例是广告投放效果分析，
　　例如将广告投放者信息表i中的信息填充到广告曝光日志表w中，使用投放者id关联。因为实际广告投放者数量很少（但是投放者信息表i很大），因此可以考虑先在w表中去重查询所有实际广告投放者id列表，以此Join过滤表i，这一结果必然是一个小表，就可以使用MapJoin。

select /*+mapjoin(x)*/* from log a left outer join (
　　select /*+mapjoin(c)*/d.*
　　from ( select distinct user_id from log ) c
　　join users d on c.user_id = d.user_id
) x on a.user_id = b.user_id;

2. 如果是上述情况2，考虑切分Join中的一个表为多片，以便将切片全部载入内存，然后采用多次MapJoin得到结果。
一个实例是商品浏览日志分析，例如将商品信息表i中的信息填充到商品浏览日志表w中，使用商品id关联。但是某些热卖商品浏览量很大，造成数据偏斜。例如，以下语句实现了一个inner join逻辑，将商品信息表拆分成2个表：

select * from(
　　select w.id, w.time, w.amount, i1.name, i1.loc, i1.cat
　　　　from w left outer join i sampletable(1 out of 2 on id) i1)
union all(
　　select w.id, w.time, w.amount, i2.name, i2.loc, i2.cat
　　　　from w left outer join i sampletable(1 out of 2 on id) i2);