spark 数据倾斜优化总结

一、数据倾斜产生原因

数据倾斜就是部分task承担了过多的计算任务,导致整个stage都被卡。
可能产生数据倾斜的场景如下

操作场景
join其中一个表比较小,但key值少
join大表与大表,但key值中存在过多的特殊值,如0或null
joinon条件包含key值过滤逻辑,导致部分数据被保留,部分被过滤,最终节点分布不均
join多对多关系表join导致数据膨胀
group by某个组合数量特别多
count distinct需要集中最后一个reduce节点处理,特殊值多就会慢
使用脚本或udf会存在强制数据分布在少量节点的可能
distriute by存在值不均匀的可能
读取上游文件数过少,单个文件过大

二、数据倾斜优化

2.1 调参优化

参数作用备注
hive.map.aggr=truemap端部分聚合仅适用于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类操作
  1. 如果是join操作,那么采用join key分布最均匀的表作为驱动表
  2. 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
此外,可以使用SMB join,减轻shuffle压力。SMB使用条件如下

条件
开启相应参数 set hive.optimize.bucketmapjoin = true;set hive.auto.convert.sortmerge.join=true;set hive.optimize.bucketmapjoin.sortedmerge = true;set hive.auto.convert.sortmerge.join.noconditionaltask=true;
小表的bucket数=大表bucket数
bucket列 == join列
数据表为分桶表,且分桶列有序
2.2.1.4 join条件包含key值过滤逻辑

尽量在join前使用where过滤

2.2.1.5 多对多关系表join
  1. 去掉热点大key
  2. 增加关联条件
  3. 减少数据范围,笛卡尔积结果尽量控制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 等对文件重新分区。