大数据优化方案----Spark数据倾斜

Posted 2022-12-09 北京小辉

1. 什么是数据倾斜

       数据倾斜是一种很常见的问题（依据二八定律），简单来说，比方WordCount中某个Key对应的数据量非常大的话，就会产生数据倾斜，导致两个后果：
1)OOM（单或少数的节点）；
2)拖慢整个Job执行时间（其他已经完成的节点都在等这个还在做的节点）。

2. 解决数据倾斜需要

1)搞定 Shuffle；
2)搞定业务场景；
3)搞定 CPU core 的使用情况；（这里的core是虚拟的core而不是机器的物理CPU核，可以理解为就是Executor的一个工作线程。）
4)搞定 OOM（内存溢出） 的根本原因等：一般都因为数据倾斜（某task任务的数据量过大，GC压力大，和Kafka不同在于Kafka的内存不经过JVM，其基于Linux的Page）。

3. 导致Spark数据倾斜的本质

       Shuffle时，需将各节点的相同key的数据拉取到某节点上的一个task来处理，若某个key对应的数据量很大就会发生数据倾斜。比方说大部分key对应10条数据，某key对应10万条，大部分task只会被分配10条数据，很快做完，个别task分配10万条数据，不仅运行时间长，且整个stage的作业时间由最慢的task决定。

       数据倾斜只会发生在Shuffle过程，以下算法可能触发Shuffle操作：

distinct：

       distinct的操作其实是把原RDD进行map操作,根据原来的key-value生成为key,value使用null来替换,并对新生成的RDD执行reduceByKey的操作,也就是说,Distinct的操作是根据key与value一起计算不重复的结果.只有两个记录中key与value都不重复才算是不重复的数据。

groupByKey：

       groupByKey会将RDD[key,value] 按照相同的key进行分组，形成RDD[key,Iterable[value]]的形式， 有点类似于sql中的groupby，例如类似于mysql中的group_concat

reduceByKey：

       reduceByKey，就是将key相同的键值对，按照Function进行计算。如代码中就是将key相同的各value进行累加。得到的结果就是类似于[(key2,2), (key3,1), (key1,2)] 形式。

aggregateByKey 函数：

       对PairRDD中相同的Key值进行聚合操作，在聚合过程中同样使用了一个中立的初始值。和aggregate函数类似，aggregateByKey返回值的类型不需要和RDD中value的类型一致。因为aggregateByKey是对相同Key中的值进行聚合操作，所以aggregateByKey’函数最终返回的类型还是PairRDD，对应的结果是Key和聚合后的值，而aggregate函数直接返回的是非RDD的结果。

join:

       join类似于SQL的inner join操作，返回结果是前面和后面集合中配对成功的，过滤掉关联不上的。

cogroup:

       对两个RDD中的kv元素，每个RDD中相同key中的元素分别聚合成一个集合。与reduceByKey不同的是针对两个RDD中相同的key的元素进行合并。

repartition:

       返回一个恰好有numPartitions个分区的RDD，可以增加或者减少此RDD的并行度。内部，这将使用shuffle重新分布数据，如果你减少分区数，考虑使用coalesce，这样可以避免执行shuffle

4. 定位最慢的Task所处的源码位置

       步骤一： 看数据倾斜发生在哪个stage（也就是看以上算子出现在哪个阶段）。yarn-client模式下查看本地log或Spark Web UI中当前运行的是哪个stage；yarn-cluster模式下，通过Spark Web UI查看运行到了哪个Stage。
主要看最慢的Stage各task分配的数据量，来确定是否是数据倾斜。

       步骤二：根据Stage划分，推算倾斜发生的代码（必然有Shuffle类算子）。简单实用方法：只要看到shuffle类算子或Spark SQL的SQL语句会有Shuffle类的算子的句子，就可以知道该地方划分为前后两个Stage。（用Python的PySpark接口，Spark Web UI会查看task在源码中的行数，Java或者Scala 同理。）

5. 解决方案

方案一：使用Hive ETL预处理

场景：若Hive表中数据不均匀，且业务中会频繁用Spark对Hive表分析；
思路：用Hive对数据预处理（对key聚合等操作），原本是Spark对Hive的原表操作，现在就是对Hive预处理后的表操作；
原理：从根源解决了数据倾斜，规避了了Spark进行Shuffle类算子操作。但Hive ETL中进行聚合等操作会发生数据倾斜，只是把慢转移给了Hive ETL；
优点：方便，效果好，规避了Spark数据倾斜；
缺点：治标不治本，Hive ETL会数据倾斜。

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

场景：发生倾斜的key很少且不重要；
思路：对发生倾斜的key过滤掉。比方在Spark SQL中用where子句或filter过滤，若每次作业执行，需要动态判定可使用sample算子对RDD采样后取数据量最多的key过滤；
原理：对倾斜的key过滤后，这些key便不会参与后面的计算，从本质上消除数据倾斜；
优点：简单，效果明显；
缺点：适用场景少，实际中导致倾斜的key很多。

方案三：提高Shuffle操作并行度

场景：任何场景都可以，优先选择的最简单方案；
思路：
对RDD操作的Shuffle算子传入一个参数，也就是设置Shuffle算子执行时的Shuffle read task数量。
对于Spark SQL的Shuffle类语句（如group by，join）即spark.sql.shuffle.partitions，代表shuffle read task的并行度，默认值是200可修改；
原理：增大shuffle read task参数值，让每个task处理比原来更少的数据；
优点：简单，有效；
缺点：缓解的效果很有限。

方案四：两阶段聚合（局部聚合+全局聚合）

场景：对RDD进行reduceByKey等聚合类shuffle算子，SparkSQL的groupBy做分组聚合这两种情况
思路：首先通过map给每个key打上n以内的随机数的前缀并进行局部聚合，即(hello, 1) (hello, 1) (hello, 1) (hello, 1)变为(1_hello, 1) (1_hello, 1) (2_hello, 1)，并进行reduceByKey的局部聚合，然后再次map将key的前缀随机数去掉再次进行全局聚合；
原理：对原本相同的key进行随机数附加，变成不同key，让原本一个task处理的数据分摊到多个task做局部聚合，规避单task数据过量。之后再去随机前缀进行全局聚合；
优点：效果非常好（对聚合类Shuffle操作的倾斜问题）；
缺点：范围窄（仅适用于聚合类的Shuffle操作，join类的Shuffle还需其它方案）。

方案五：将reduce join转为map join

场景：对RDD或Spark SQL使用join类操作或语句，且join操作的RDD或表比较小（百兆或1,2G）；
思路：使用broadcast和map类算子实现join的功能替代原本的join，彻底规避shuffle。对较小RDD直接collect到内存，并创建broadcast变量；并对另外一个RDD执行map类算子，在该算子的函数中，从broadcast变量（collect出的较小RDD）与当前RDD中的每条数据依次比对key，相同的key执行你需要方式的join；
原理：若RDD较小，可采用广播小的RDD，并对大的RDD进行map，来实现与join同样的效果。简而言之，用broadcast-map代替join，规避join带来的shuffle（无Shuffle无倾斜）；
优点：效果很好（对join操作导致的倾斜），根治；
缺点：适用场景小（大表+小表），广播（driver和executor节点都会驻留小表数据）小表也耗内存。

方案六：采样倾斜key并分拆join操作

场景：两个较大的（无法采用方案五）RDD/Hive表进行join时，且一个RDD/Hive表中少数key数据量过大，另一个RDD/Hive表的key分布较均匀（RDD中两者之一有一个更倾斜）；
思路：
    1. 对更倾斜rdd1进行采样（RDD.sample）并统计出数据量最大的几个key；
    2. 对这几个倾斜的key从原本rdd1中拆出形成一个单独的rdd1_1，并打上0~n的随机数前缀，被拆分的原rdd1的另一部分（不包含倾斜key）又形成一个新rdd1_2；
    3. 对rdd2过滤出rdd1倾斜的key，得到rdd2_1，并将其中每条数据扩n倍，对每条数据按顺序附加0~n的前缀，被拆分出key的rdd2也独立形成另一个rdd2_2； 
    【个人认为，这里扩了n倍，最后union完还需要将每个倾斜key对应的value减去(n-1)】
    4. 将加了随机前缀的rdd1_1和rdd2_1进行join（此时原本倾斜的key被打散n份并被分散到更多的task中进行join）； 
    【个人认为，这里应该做两次join，两次join中间有一个map去前缀】
    5. 另外两个普通的RDD（rdd1_2、rdd2_2）照常join；
    6. 最后将两次join的结果用union结合得到最终的join结果。
原理：对join导致的倾斜是因为某几个key，可将原本RDD中的倾斜key拆分出原RDD得到新RDD，并以加随机前缀的方式打散n份做join，将倾斜key对应的大量数据分摊到更多task上来规避倾斜；
优点：前提是join导致的倾斜（某几个key倾斜），避免占用过多内存（只需对少数倾斜key扩容n倍）；
缺点：对过多倾斜key不适用。

方案七：用随机前缀和扩容RDD进行join

场景：RDD中有大量key导致倾斜；
思路：与方案六类似。 
1. 查看RDD/Hive表中数据分布并找到造成倾斜的RDD/表； 
2. 对倾斜RDD中的每条数据打上n以内的随机数前缀； 
3. 对另外一个正常RDD的每条数据扩容n倍，扩容出的每条数据依次打上0到n的前缀； 
4. 对处理后的两个RDD进行join。
原理：与方案六只有唯一不同在于这里对不倾斜RDD中所有数据进行扩大n倍，而不是找出倾斜key进行扩容（这是方案六）；
优点：对join类的数据倾斜都可处理，效果非常显著；
缺点：缓解，扩容需要大内存。 
【个人认为，这里和方案六一样，也需要对扩容的key对应的value最后减去(n-1)，除非只需大小关系，对值没有要求】

方案八：多种方案组合

实际中，需综合着对业务全盘考虑，可先用方案一和二进行预处理，同时在需要Shuffle的操作提升Shuffle的并行度，最后针对数据分布选择后面方案中的一种或多种。实际中需要对数据和方案思路理解灵活应用。

原文参考：

• spark-数据-方法
• 数据倾斜是多么痛？spark作业/面试/调优必备秘籍