大数据之Spark：Spark处理数据倾斜有哪几种方式

Posted 2021-06-26 浊酒南街

目录

• 1.前言
• 2.解决方案
• • 解决方案一：聚合原数据
  • 解决方案二：过滤导致倾斜的key
  • 解决方案三：提高shuffle操作中的reduce并行度
  • 解决方案四：使用随机key实现双重聚合
  • 解决方案五：将reduce join转换为map join
  • 解决方案六：sample采样对倾斜key单独进行join
  • 解决方案七：使用随机数以及扩容进行join

1.前言

spark中的数据倾斜问题主要指shuffle过程中出现的数据倾斜问题，是由于不同的key对应的数据量不同导致的不同task所处理的数据量不同的问题；
数据倾斜的发生，一般都是一个或者某几个Key对应的数据过大，导致Task执行过慢，或者内存溢出OOM，一般发生在Shuffle的时候，比如reduceByKey、countByKey、groupByKey，容易产生数据倾斜。

2.解决方案

解决问题：首先看log日志信息，因为log日志报错的时候会提示在哪一行，然后去检查发生Shuffle的地方，这些地方比较容易发生数据

解决方案一：聚合原数据

假设数据一般来源于Hive表，那么在生成Hive表的时候对数据进行聚合，按照Key进行分组，将Key对应的所有values以另一种格式存储，比如拼接成一个字符串，就不用Shuffle了，也就不会出现数据倾斜。
1.避免shuffle过程
2.增大key粒度（减小数据倾斜可能性，增大每个task的数据量；

解决方案二：过滤导致倾斜的key

在不影响业务需求的条件下，过滤掉引起数据倾斜的key；

解决方案三：提高shuffle操作中的reduce并行度

提高reduce端并行度并没有从根本上改变数据倾斜的本质和问题（方案一和方案二从根本上避免了数据倾斜的发生），只是尽可能地去缓解和减轻shuffle reduce task的数据压力，以及数据倾斜的问题，适用于有较多key对应的数据量都比较大的情况。

解决方案四：使用随机key实现双重聚合

通过map算子给每个数据的key添加随机数前缀，对key进行打散，将原先一样的key变成不一样的key，然后进行第一次聚合，这样就可以让原本被一个task处理的数据分散到多个task上去做局部聚合；随后，去除掉每个key的前缀，再次进行聚合；
对于由groupByKey、reduceByKey这类算子造成的数据倾斜由比较好的效果，仅仅适用于聚合类的shuffle操作，适用范围相对较窄；

解决方案五：将reduce join转换为map join

正常情况下，join操作都会执行shuffle过程，并且执行的是reduce join，也就是先将所有相同的key和对应的value汇聚到一个reduce task中，然后再进行join。
如果一个RDD是比较小的，则可以采用广播小RDD全量数据+map算子来实现与join同样的效果，也就是map join，此时就不会发生shuffle操作，也就不会发生数据倾斜。
（注意，RDD是并不能进行广播的，只能将RDD内部的数据通过collect拉取到Driver内存然后再进行广播）

解决方案六：sample采样对倾斜key单独进行join

如果你发现整个RDD就一个key的数据量特别多，那么就可以考虑使用这种方法。
当数据量非常大时，可以考虑使用sample采样获取10%的数据，然后分析这10%的数据中哪个key可能会导致数据倾斜，然后将这个key对应的数据单独提取出来

解决方案七：使用随机数以及扩容进行join

如果在进行join操作时，RDD中有大量的key导致数据倾斜，我们可以考虑对其中一个RDD数据进行扩容，另一个RDD进行稀释后再join。通过flatmap进行扩容，然后再将随机数打入进去进行join；
参考：https://blog.csdn.net/weixin_38747692/article/details/106857625