2022-01-03-Spark-42(数据倾斜问题解决)

现象与原理

『不患多而患不均』，这是分布式环境下最大的问题。在进行 shuffle 的时候，不同的key对应的数据量不同导致不同task处理的数据量不同，比如按照key进行聚合和join，这两种其中的某个key数据量特别大，就容易发生数据倾斜

聚合

join

危害

发生OOM
集群崩溃
程序执行缓慢

定位数据倾斜

一个spark job是由多个stage组成的，stage之间具有先后关系，所以是串行执行的，一个stage是由多个task组成的，每个task之间可以并行运行，一个stage的运行时间由耗时最长的那个task来决定的

经验
查看代码中的shuffle算子，如join cogroup reducebykey aggergatebykey repartion groupbykey distinct等，根据代码逻辑判断是否会出现数据倾斜
通过 key 统计，采用抽样的方式，对数据进行抽样，统计出现的次数，根据出现次数大小排序取出前几个
log
log一般会报是在你的哪一行代码，导致了OOM异常。或者看log，看看是执行到了第几个stage。spark代码，是怎么划分成一个一个的stage的。哪一个stage生成的task特别慢，就能够自己用肉眼去对你的spark代码进行stage的划分，就能够通过stage定位到你的代码，到底哪里发生了数据倾斜。
ui
某个stage运行时间过长，shuffle raed 和 shuffle write 的数据量相差巨大，task运行时间过长，读写数据量相差巨大

解决数据倾斜的方案

调整shuffle 并行度
Spark 在做 Shuffle 时，默认使用 HashPartitioner（非 Hash Shuffle）对数据进行分区。如果并行度设置的不合适，可能造成大量不相同的 Key 对应的数据被分配到了同一个 Task 上，造成该 Task 所处理的数据远大于其它 Task，从而造成数据倾斜.dataFrame 和 sparkSql 可以设置 spark.sql.shuffle.partitions=[num_tasks] 参数控制 shuffle 的并发度，默认为200
适用场景:大量不同的 Key 被分配到了相同的 Task 造成该 Task 数据量过大。
优势:实现比较简单，一般如果出现数据倾斜，都可以通过这种方法先试验几次，如果问题未解决，再尝试其它方法。
劣势:治标不治本，只是缓解数据倾斜，没有彻底根除
自定义 Partitioner
适用场景：大量不同的 Key 被分配到了相同的 Task 造成该 Task 数据量过大。使用自定义的 Partitioner 实现类代替默认的 HashPartitioner，尽量将所有不同的 Key 均匀分配到不同的 Task 中。
优势:不影响原有的并行度设计。如果改变并行度，后续 Stage 的并行度也会默认改变，可能会影响后续 Stage
劣势:适用场景有限，只能将不同 Key 分散开，对于同一 Key 对应数据集非常大的场景不适用。效果与调整并行度类似，只能缓解数据倾斜而不能完全消除数据倾斜。而且需要根据数据特点自定义专用的 Partitioner，不够灵活。
两阶段聚合(加盐局部聚合 + 去盐全局聚合)

def antiSkew(): RDD[(String, Int)] = {
    val SPLIT = "-"
    val prefix = new Random().nextInt(10)
    pairs.map(t => ( prefix + SPLIT + t._1, 1))
        .reduceByKey((v1, v2) => v1 + v2)
        .map(t => (t._1.split(SPLIT)(1), t2._2))
        .reduceByKey((v1, v2) => v1 + v2)
}

适用场景：对rdd执行reducebykey等聚合类shuffle或者是group by 语句进行分组聚合的
优势:争对聚合类操作，可以大幅度缓解数据倾斜
劣势:join 不行的，另外进行两次 mapreduce，性能稍微比一次的差些

广播变量(map join)
优势:对join操作导致的数据倾斜，效果非常好，因为根本就不会发生shuffle也就不会发生数据倾斜的
劣势:数据量太大的情况下，广播变量还是需要太大的消耗的
采样倾斜key 拆分 join 再 union all
将有数据倾斜的 RDD 中倾斜 Key 对应的数据集单独抽取出来加上随机前缀，另外一个 RDD 每条数据分别与随机前缀结合形成新的RDD（相当于将其数据增到到原来的N倍，N即为随机前缀的总个数），然后将二者Join并去掉前缀。然后将不包含倾斜Key的剩余数据进行Join。最后将两次Join的结果集通过union all合并，即可得到全部Join结果
适用场景：两张表都比较大，无法使用 Map 端 Join。其中一个 RDD 有少数几个 Key 的数据量过大，另外一个 RDD 的 Key 分布较为均匀。
优势:相对于 Map 则 Join，更能适应大数据集的 Join。如果资源充足，倾斜部分数据集与非倾斜部分数据集可并行进行，效率提升明显。且只针对倾斜部分的数据做数据扩展，增加的资源消耗有限。
劣势:如果倾斜 Key 非常多，则另一侧数据膨胀非常大，此方案不适用。而且此时对倾斜 Key 与非倾斜 Key 分开处理，需要扫描数据集两遍，增加了开销
大表 key 加盐，小表扩大 N 倍 join
如果出现数据倾斜的 Key 比较多，上一种方法将这些大量的倾斜 Key 分拆出来，意义不大。此时更适合直接对存在数据倾斜的数据集全部加上随机前缀，然后对另外一个不存在严重数据倾斜的数据集整体与随机前缀集作笛卡尔乘积（即将数据量扩大N倍）。
其实就是上一个方法的特例或者简化。少了拆分，也就没有 union。
适用场景：一个数据集存在的倾斜 Key 比较多，另外一个数据集数据分布比较均匀。
优势:对大部分场景都适用，效果不错。
劣势:需要将一个数据集整体扩大 N 倍，会增加资源消耗。
过滤异常数据
空值或者异常值，大多是这个原因引起的
无效数据，大量重复的测试数据或是对结果影响不大的有效数据

解决Spark数据倾斜
 Spark 数据倾斜及其解决方案 (qq.com)
spark broadcast join优化