1 数据倾斜概述

数据倾斜指参与计算的数据分布不均，即某个key或者某些key的数据量远超其他key，导致在shuffle阶段，大量相同key的数据被发往同一个Reduce，导致该Reduce所需的时间远超其他Reduce，成为整个任务的瓶颈。
Hive中的数据倾斜常出现在分组聚合和join操作的场景中。

2 分组聚合导致的数据倾斜

2.1 优化说明

Hive中未经优化的分组聚合，是通过一个MapReduce Job实现的。Map端负责读取数据，并按照分组字段分区，通过Shuffle，将数据发往Reduce端，各组数据在Reduce端完成最终的聚合运算。
如果group by分组字段的值分布不均，就可能导致大量相同的key进入同一Reduce，从而导致数据倾斜问题。

解决思路：
Map-Site聚合和Skew-GroupBy优化。

1、Map-Side聚合

开启Map-Side聚合后，数据会现在Map端完成部分聚合工作（可以看作大小一样的切片，相当于先完成切片内的聚合工作）。

即便原始数据是倾斜的，经过Map端的初步聚合后，发往Reduce的数据也就不再倾斜了。最佳状态下，Map-端聚合能完全屏蔽数据倾斜问题。

2、Skew-GroupBy优化

Skew-GroupBy专门用于处理由于group by导致数据倾斜的方案。

原理：
启动两个MR任务，第一个MR按照随机数分区，将数据分散发送到Reduce，完成部分聚合；
第二个MR读取第一个Reduce端的数据按照分组字段分区，完成最终聚合。

2.2 案例

1、示例SQL语句

select
    province_id,
    count(*)
from order_detail
group by province_id;

2、优化前
表中的province_id字段是存在倾斜的，若不经过优化，通过观察yarn中reduce任务的执行过程，是能够看出数据倾斜现象的。
hive中的map-side聚合是默认开启的，若想看到数据倾斜的现象，需要先将hive.map.aggr参数设置为false。
3、优化思路
（1）Map-Side聚合
设置参数：

--启用map-side聚合
set hive.map.aggr=true;
--关闭skew-groupby
set hive.groupby.skewindata=false;

执行计划：
在这里插入图片描述
观察yarn中reduce任务的执行过程，很明显可以看到开启map-side聚合后，reduce数据不再倾斜。
（2）Skew-GroupBy优化
设置参数:

--启用skew-groupby
set hive.groupby.skewindata=true;
--关闭map-side聚合
set hive.map.aggr=false;

开启Skew-GroupBy优化后，可以很明显看到该sql执行在yarn上启动了两个mr任务，第一个mr打散数据，第二个mr按照打散后的数据进行分组聚合。
在这里插入图片描述

2.3 总结

Map-site优化优于Skew-GroupBy优化，能做Map-site聚合就做Map-site聚合。

Map-site聚合要在Map端维护一个HashTable，HashTable消耗了内存，即内存不充足的情况下尽量不做Map-site聚合，但也可以做，当HashTable超过设定的内存阈值时，会flush刷新。

当内存很小时会flush很多次导致map端聚合没有解决数据倾斜问题，此时可以用Skew-GroupBy聚合。也就是不管内存足部足够都能把数据打散再做聚合。

3 Join导致的数据倾斜

3.1 优化说明

未经优化的join操作，默认是使用common join算法，也就是通过一个MapReduce Job完成计算。Map端负责读取join操作所需表的数据，并按照关联字段进行分区，通过Shuffle，将其发送到Reduce端，相同key的数据在Reduce端完成最终的Join操作。

如果关联字段的值分布不均，就可能导致大量相同的key进入同一Reduce，从而导致数据倾斜问题。

由join导致的数据倾斜问题，有如下三种解决方案：map join、skew join、调整SQL语句。

1、map join

map join中join操作仅在map端就能完成，没有shuffle操作，没有reduce阶段，自然不会产生reduce端的数据倾斜。该方案适用于大表join小表时发生数据倾斜的场景。

原理：
多个Mapper缓存小表数据，大表数据进行切片（切片只与大小有关，与key无关），均匀切成若干个片，一个map负责处理一个切片（逐条遍历然后找缓存的小表的数据join）再逐条输出，使得每个map处理的数据量一致，解决数据倾斜问题。

2、skew join

skew join解决大表join大表出现数据倾斜的问题。

原理：
为倾斜的大key单独启动一个map join任务进行计算，其余key进行正常的common join。
在这里插入图片描述
相关参数：

--启用skew join优化
sethive.optimize.skewjoin=true;
--触发skew join的阈值，若某个key的行数超过该参数值，则触发（按照行数进行检测）
set hive.skewjoin.key=100000;

这种方案对参与join的源表大小没有要求，但是对两表中倾斜的key的数据量有要求，要求一张表中的倾斜key的数据量比较小（方便走mapjoin）。

3、调整SQL语句

若参与join的两表均为大表，其中一张表的数据是倾斜的，可以对SQL语句进行相应的调整。
假设原始SQL语句如下：A，B两表均为大表，且其中一张表的数据是倾斜的。

select
    *
from A
join B
on A.id=B.id;

在这里插入图片描述
图中1001为倾斜的大key，被发往了同一个Reduce进行处理。

调整SQL语句如下：

select
    *
from(
    select --打散操作，加随机数0、1
        concat(id,'_',cast(rand()*2 as int)) id,
        value
    from A
)ta
join(
    select --扩容操作
        concat(id,'_',0) id,
        value
    from B
    union all
    select
        concat(id,'_',1) id,
        value
    from B
)tb
on ta.id=tb.id;

调整之后的SQL语句执行计划：
在这里插入图片描述

3.2 案例

1、示例SQL语句

select
    *
from order_detail od
join province_info pi
on od.province_id=pi.id;

2、优化前

--关闭Map Join自动转换
set hive.auto.convert.join=false;

--关闭skew join优化（默认为关闭状态）
sethive.optimize.skewjoin=false;

order_detail表中的province_id字段是存在倾斜的，若不经过优化，观察yarn中的reduce任务可以看到数据倾斜的现象。
hive中的map join自动转换是默认开启的，若想看到数据倾斜的现象，需要先将hive.auto.convert.join参数设置为false。
3、优化思路
（1）map join
设置参数：

--启用map join
set hive.auto.convert.join=true;
--关闭skew join
set hive.optimize.skewjoin=false;

可以很明显看到开启map join以后，mr任务只有map阶段，没有reduce阶段，没有数据倾斜发生。
在这里插入图片描述

（2）skew join
设置参数：

--启动skew join
set hive.optimize.skewjoin=true;
--关闭map join
set hive.auto.convert.join=false;

开启skew join后，使用explain查看执行计划：
在这里插入图片描述
skew join生效，任务既有common join，又有部分key走了map join。并且该sql在yarn上最终启动了两个mr任务，而且第二个任务只有map没有reduce阶段，说明第二个任务是对倾斜的key进行了map join。