Spark join数据倾斜调优

Spark中常见的两种数据倾斜现象如下

• stage部分task执行特别慢

一般情况下是某个task处理的数据量远大于其他task处理的数据量，当然也不排除是程序代码没有冗余，异常数据导致程序运行异常。

• 作业重试多次某几个task总会失败
  

常见的退出码143、53、137、52以及heartbeat timed out异常，通常可认为是executor内存被打满。

RDD调优方法

1. 查看数据分布
   Spark Core中shuffle算子出现数据倾斜时，可在Spark作业中加入查看key分布的代码，也可以将代码拆解出来使用spark-shell做测试

val rdd = sc.parallelize(Array("hello", "hello", "hello", "hi")).map((_,1))// 数据量较少
rdd.reduceByKey(_ + _)
.sortBy(_._2, false)
.take(20)
// 数据量较大, 用sample采样后在统计
rdd.sample(false, 0.1)
.reduceByKey(_+_)
.sortBy(_._2, false)
.take(20)

2. 调整shuffle并行度
   原理：Spark在做shuffle时，默认使用HashPartitioner（非Hash Shuffle）对数据进行分区。如果并行度设置的不合适如比较小，可能造成大量不相同的key对应的数据被分配到了同一个task上，造成该task所处理的数据远大于其它task，从而造成数据倾斜
   

调优建议：

• 使用spark.default.parallelism调整分区数，默认值200建议500或更大
• 在shuffle的算子上直接设置分区数，如：a.join(b, 500)、rdd.reduceByKey(_ + _, 500)

3. reduce join转map join
   原理：不使用join算子直接进行连接操作，而使用broadcast变量与map类算子实现join操作，进而完全规避掉shuffle类的操作，彻底避免数据倾斜的出现
   

调优建议：

• broadcast的数据量不要超过500M, 过大driver/executor可能会oom

// 1.broadcast小表
val rdd1Broadcast = sc.broadcast(rdd1.collect())
// 2.map join
rdd2.map { x =>val rdd1DataMap = rdd1Broadcast.value.toMaprdd1DataMap.get(x._1) match {case Some(v) => (x._1, (x._2, v))case None => (x._1, (x._2, null))}
}
// 2.或者直接
rdd2.join(rdd1Broadcast)

4. 分拆join在union
   原理：将有数据倾斜的RDD1中倾斜key对应的数据集单独抽取出来加盐（随机前缀），另外一个RDD2每条数据分别与所有的随机前缀结合形成新的RDD（相当于将其数据增到到原来的N倍，N即为随机前缀的总个数），然后将二者join之后去掉前缀；然后将不包含倾斜key的剩余数据进行join；最后将两次join的结果集通过union合并，即可得到全部join结果。
   

调优建议：

// 1.统计数量最大的key
val skewedKeySet = rdd1.sample(false, 0.2).reduceByKey(_ + _).sortBy(_._2, false).take(10).map(x => x._1).toSet// 2.拆分异常的rdd, 倾斜key加上随机数
val rdd1_1 = rdd1.filter(x => skewedKeySet.contains(x._1)).map { x =>val prefix = scala.util.Random.nextInt(10).toString(s"${prefix}_${x._1}", x._2)
}
val rdd1_2 = rdd1.filter(x => !skewedKeySet.contains(x._1))// 3.正常rdd存在倾斜key的部分进行膨胀
val rdd2_1 = rdd2.filter(x => skewedKeySet.contains(x._1)).flatMap { x =>val list = 0 until 10list.map(i => (s"${i}_${x._1}", x._2))}val rdd2_2 = rdd2.filter(x => !skewedKeySet.contains(x._1))// 4.倾斜key的rdd进行join
val skewedRDD = rdd1_1.join(rdd2_1).map(x => (x._1.split("_")(1), x._2))
// 5.普通key的rdd进行join
val sampleRDD = rdd1_2.join(rdd2_2)
// 6.结果union
skewedRDD.union(sampleRDD)

SQL调优方法

1. 查看数据分布
   统计某个查询结果或表中出现次数超过200次的key

WITH a AS (${query})
SELECT k，s
FROM (SELECT ${key} AS k，count(*) AS sFROM aGROUP BY ${key}
)
WHERE s > 200

2. 自动调整shuffle并行度
   原理：自适应执行开启的前提下（AQE），假设我们设置的shuffle partition个数为5，在map stage结束之后，我们知道每一个partition的大小分别是70MB，30MB，20MB，10MB和50MB。假设我们设置每一个reducer处理的目标数据量是64MB，那么在运行时，我们可以实际使用3个reducer。第一个reducer处理partition 0 (70MB)，第二个reducer处理连续的partition 1 到3，共60MB，第三个reducer处理partition 4 (50MB)
   

Spark参数：

参数	说明	推荐值
spark.sql.adaptive.enabled	开启自适应执行	线上默认值true
spark.sql.adaptive.coalescePartitions.minPartitionNum	自适应执行中使用的最小shuffle后分区数，默认值executor*core数	无
spark.sql.adaptive.coalescePartitions.initialPartitionNum	合并前的初始shuffle分区数量，默认值spark.sql.shuffle.partitions	无
spark.sql.adaptive.advisoryPartitionSizeInBytes	合并小分区到建议的目标值, 默认256m	无
spark.sql.shuffle.partitions	join等操作分区数，默认值200	推荐500或更大

3. 自动优化Join
   原理：自适应执行开启的前提下（AQE），我们可以获得SortMergeJoin两个子stage的数据量，在满足条件的情况下，即一张表小于broadcast阈值，可以将SortMergeJoin转化成BroadcastHashJoin
   

参数	说明	推荐值
spark.sql.adaptive.enabled	开启自适应执行	线上默认值true
spark.sql.autoBroadcastJoinThreshold	默认10M，设置为-1可以禁用广播；实际根据hive表存储的统计信息或文件预估大小与此值做判断看是否做broadcast，由于文件是压缩格式一般情况下此参数并不可靠	建议膨胀系数spark.sql.sources.fileCompressionFactor=10推荐此参数保持默认，调整自适应的broadcast参数
spark.sql.adaptive.autoBroadcastJoinThreshold	此参数仅影响自适应执行阶段join优化时broadcast阈值；设置为-1可以禁用广播；默认值spark.sql.autoBroadcastJoinThreshold	自适应执行得到的数据比较准确，driver内存足够的前提下可以将此值调大如200M

4. 自动处理数据倾斜
   原理：自适应执行开启的前提下（AQE），我们可以在运行时很容易地检测出有数据倾斜的partition。当执行某个stage时，我们收集该stage每个mapper 的shuffle数据大小和记录条数。如果某一个partition的数据量或者记录条数超过中位数的N倍，并且大于某个预先配置的阈值，我们就认为这是一个数据倾斜的partition，需要进行特殊的处理
   

参数	说明	推荐值
spark.sql.adaptive.enabled	开启自适应执行	线上默认值true
spark.sql.adaptive.skewJoin.enabled	开启自动解决数据倾斜，默认值true	无
spark.sql.adaptive.skewJoin.skewedPartitionFactor	影响因子，某分区数据大小超过所有分区中位数与影响因子乘积，才会被认为发生了数据倾斜	无
spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes	视为倾斜分区的分区数据最小值	无