一、数据倾斜产生原因
数据倾斜就是部分task承担了过多的计算任务，导致整个stage都被卡。
可能产生数据倾斜的场景如下
| 操作 | 场景 |
|---|---|
| join | 其中一个表比较小，但key值少 |
| join | 大表与大表，但key值中存在过多的特殊值，如0或null |
| join | on条件包含key值过滤逻辑，导致部分数据被保留，部分被过滤，最终节点分布不均 |
| join | 多对多关系表join导致数据膨胀 |
| group by | 某个组合数量特别多 |
| count distinct | 需要集中最后一个reduce节点处理，特殊值多就会慢 |
| 使用脚本或udf | 会存在强制数据分布在少量节点的可能 |
| distriute by | 存在值不均匀的可能 |
| 读取 | 上游文件数过少，单个文件过大 |
二、数据倾斜优化
2.1 调参优化
| 参数 | 作用 | 备注 |
|---|---|---|
hive.map.aggr=true | map端部分聚合 | 仅适用于hive引擎，不适用于spark。spark本身就支持map端的局部聚合 |
hive.groupby.skewindata=true | 参数设置为true时，生成的查询计划会有2个MR job。第一个MR中，map的输出结果会随机分配到reduce中，每个reduce做部分聚合操作，并输出结果，这样相同的key会被分发到不同的reduce中，起到了负载均衡的目的，第二个MR再根据第一个MR预处理的结果，完成聚合操作 | 仅适用于hive引擎，spark不支持此参数 |
2.2 代码优化
2.2.1 join类操作
- 如果是join操作，那么采用join key分布最均匀的表作为驱动表
- where类的操作在join前进行
2.2.1.1 小表join大表
使用 /*+mapjoin(a)*/ 让小的维度表先进内存，在map端完成reduce
使用此操作需要对小的维度表配置DQC数据量监控，以避免进入内存的维表数据过大
2.2.1.2 中表join大表
在维表大小中等，完全进入内存可能报错；但日志与维表关联部分出现某种行为的维值较少时，可以采用两次 mapjoin 的方式优化倾斜问题。例，有交易日志log，和维表dim，它们2次join示例如下
select /*+mapjoin(b)*/
a.*,
b.*
from log a
left join (
select /*+mapjoin(d)*/
c.*
from dim AS c
join (
select dim_id
from log
group by dim_id
) AS d
on c.dim_id = d.dim_id
) AS b
on a.dim_id = b.dim_id
2.2.1.3 大表join大表
大表join大表一般由于key值存在 null 等特殊值，导致数据过于集中倾斜。这种情况可以使用hash函数将key打散。这里不要用rand，因为fetch failed后，整个job会被kill
2.2.1.4 join条件包含key值过滤逻辑
尽量在join前使用where过滤
2.2.1.5 多对多关系表join
- 去掉热点大key
- 增加关联条件
- 减少数据范围，笛卡尔积结果尽量控制1亿
2.2.2 group by操作
将倾斜数据拿出来单独处理，后面再union 回去
2.2.3 count distinct操作
2.2.3.1 单个 count distinct
可以将它转换为count+group by。如果单个值数据量比较集中，可以过滤该值，结果+1.
下面的示例是从t表中取user_id的uv。其中，user_id有大量为0的异常值
优化前
select count(distinct user_id) as uv from t
优化后
select count(user_id) +1 as uv
from (
select user_id
from t
where user_id != 0
group by user_id
)
2.2.3.1 多个 count distinct
如果多个count distinct，去重的键是一样的，如对用户计算uv和去作弊uv，那么仍可按user_id去重后，根据是否作弊按0,1进行SUM。
如果去重的键是不一样的，那么多个count distinct 可以优化为多个group by再union all的方式。但这会扫描多次表，具体优化效果需根据具体情况测试确定。
2.2.4 使用脚本或udf
这个需要根据具体业务场景优化。如果不方便优化，可以降低 spark.sql.shuffle.partitions的值。
2.2.5 distriute by
distriute by 需要保证键比较均匀。同时，不能使用rand函数，否则shuffle fetch failed后，整个job会被kill
2.2.6 读取操作
读取一般不会倾斜。出现倾斜一般是上游文件数过少，下游处理的executor多。spark默认一个executor对应一个文件，这样会有大量executor空跑。这种情况可以使用 distriute by、repartition 等对文件重新分区。
