Spark 调优之数据倾斜

什么是数据倾斜？

Spark 的计算抽象如下

Spark 的计算抽象.png

数据倾斜指的是：并行处理的数据集中，某一部分（如 Spark 或 Kafka 的一个 Partition）的数据显著多于其它部分，从而使得该部分的处理速度成为整个数据集处理的瓶颈。

如果数据倾斜不能解决，其他的优化手段再逆天都白搭，如同短板效应，任务完成的效率不是看最快的task,而是最慢的那一个。

数据倾导致的后果：

1. 数据倾斜直接可能会导致一种情况：Out Of Memory 或者GC 超时。
2. 任务不一定失败，但是极端慢。(但是目前我遇到的数据倾斜几乎都失败了)

数据倾斜示意图

image.png.png

如上图所示
个别 ShuffleMapTask2 (98 亿条数据)处理过度大量数据。导致拖慢了整个 Job 的执行时间。
这可能导致该 Task 所在的机器 OOM，或者运行速度非常慢。

倾斜原理：

在进行 shuffle 的时候，必须将各个节点上相同的 key 拉取到某个节点上的一个 task 来进行处理，比如按照 key 进行聚合或 join 等操作。此时如果某个 key 对应的数据量特别大的话，就会发生数据倾斜。

比如大部分 key 对应 几十上百 条数据，但是个别 key 却对应了 成千上万 条数据，那么大部分 task 可能就只会分配到 少量 数据，然后 1 秒钟就运行完了；
但是个别 task 可能分配到了 海量数据，要运行一两个小时。
因此，整个 Spark 作业的运行进度是由运行时间最长的那个 task 决定的。
注：由于同一个 Stage 内的所有 Task 执行相同的计算，在排除不同计算节点计算能力差异的前提下，不同 Task 之间耗时的差异主要由该 Task 所处理的数据量决定。

定位倾斜位置

可能触发的算子

可能触发的算子(不完全)
distinct
groupByKey
reduceByKey
aggregateByKey
join
cogroup
repartition
count

task 内存溢出

这种情况下去定位出问题的代码(触发JOb的Action位)就比较容易了。

可以直接看 yarn-client 模式下本地 log 的异常栈，或者是通过 YARN 查看 yarn-cluster 模式下的 log 中的异常栈。

一般来说，通过异常栈信息就可以定位到你的代码中哪一行(触发JOb的Action位置)发生了内存溢出和溢出的Stage是哪一个。然后在那行代码附近找找，一般也会有 shuffle类算子，此时很可能就是这个算子导致了数据倾斜，但是是经工作中发现，这个定位具体行数还是比较困难，因为日志只会出现触发JOb的Action算子的代码行数，而一个Job可能有多可shuffle阶段，你要很了解任务的划分才有可能找对位置。

要注意的是，出现内存溢出不一定就是倾斜。这只是一种可能而已。

task 执行特别慢的情况

与上面类似，虽然不报错，但是程序就在这里停住了，某部分task一直没有完成。

为了进一步确定是否倾斜，最好的办法是去看web ui,查看当前task 所在Stage的所有task,看看执行特别慢的task 运行时间、所处理的数据量、GC等信息。

如果与其他task差异较大，说明出现了倾斜问题，那我们接下来就该去解决问题了。

key 的数据分布情况

我工作中因为权限、环境等各种问题，无法查看Web UI 所以对于定位GC、OOM的问题特别难受~~~。

所有有时候采用很笨的方法来确定一下是否数据倾斜

上述表格中列举了可能出现倾斜的算子，那么这些我们可以抽样统计一下该算子操作的key对应的数据量。如果key 的分布及不均匀，某种程度上也可以判定是出现了倾斜

df(dataFrame) 部分数据如下
+--------+-----------+------------+------+--------+----+
|  userid|  zubo_nums|  total_nums|  nums|     day|hour|
+--------+-----------+------------+------+--------+----+
| userid1| zubo_nums1| total_nums1| nums1|20190101|  00|
| userid2| zubo_nums2| total_nums2| nums2|20190101|  00|
| userid3| zubo_nums3| total_nums3| nums3|20190101|  00|
| userid4| zubo_nums4| total_nums4| nums4|20190101|  00|
| userid5| zubo_nums5| total_nums5| nums5|20190101|  00|
| userid6| zubo_nums6| total_nums6| nums6|20190101|  00|
| userid7| zubo_nums7| total_nums7| nums7|20190101|  00|
| userid8| zubo_nums8| total_nums8| nums8|20190101|  00|
| userid9| zubo_nums9| total_nums9| nums9|20190101|  00|
|userid10|zubo_nums10|total_nums10|nums10|20190101|  00|
+--------+-----------+------------+------+--------+----+

logger.info("\n df count=" +df.count())
df.sample(false,0.1).rdd.keyBy(row=>row.getAs("userid").toString).countByKey().foreach(println _)

df count=2058

多次抽样对比
(userid88,3)
(userid99,1)
(userid61,2)
(userid50,2)
(userid34,2)
(userid1,33)
(userid39,4)
(userid83,3)
--------------------
(userid61,1)
(userid50,1)
(userid34,1)
(userid1,35)
(userid83,2)
(userid17,1)
(userid69,2)
---------
(userid99,2)
(userid61,1)
(userid50,2)
(userid34,2)
(userid1,25)
(userid39,1)
(userid83,1)
(userid94,2)
(userid17,1)

从上述抽样结果接可以看出，userid1这个key数量明显多余其他key。
多次抽样也可以看出，这样统计一定程度上可以反应倾斜的问题并且可以确定倾斜的key,这样对于我们后续解决倾斜问题有一定的帮助。

---

解决数据倾斜

从源端数据解决

下面距两个例子说明：

kafka数据源
我们一般通过 DirectStream 方式读取 Kafka数据。

由于 Kafka 的每一个 Partition 对应 Spark 的一个Task（Partition），所以 Kafka 内相关 Topic 的各Partition 之间数据是否平衡，直接决定 Spark处理该数据时是否会产生数据倾斜。

Kafka 某一 Topic 内消息在不同 Partition之间的分布，主要由 Producer 端所使用的Partition 实现类决定。

如果使用随机 Partitioner，则每条消息会随机发送到一个 Partition 中，从而从概率上来讲，各Partition间的数据会达到平衡。此时源 Stage（直接读取 Kafka 数据的 Stage）不会产生数据倾斜。

所以如果业务没有特别需求，我们可以在Producer端的 Partitioner 采用随机的方式，并且可以每个批次数据量适当增加 Partition 的数量，达到增加task目的。

但是很多业务要求将具备同一特征的数据顺序消费，此时就需要将具有相同特征的数据放于同一个 Partition 中。比如某个地市、区域的数据需要放在一个Partition 中，如果此时出现了数据倾斜，就只能采用其他的办法解决了。

hive数据源
如果数据源是来自hive,那么我们可以考虑在hive端就针对该key一次etl处理。

如果评估可行，那我们在Spark就可以在Spark端使用etl后的数据了，也就不用Spark中执行之前倾斜的部分的逻辑了。

优点：实现起来简单便捷，效果不错，完全规避掉了数据倾斜，Spark 作业的性能会大幅度提升。

缺点：治标不治本，我们只是把数据倾斜提前到了hive端，Hive ETL 中还是会发生数据倾斜，所以我们还是避免不了要在hive端处理倾斜问题。

适用情况：
因为本质上没有解决数据倾斜的问题，我们只有解决了Hive端数据倾斜的问题才算真正的解决这个问题。
所以当hive端的数据仅仅被调用一两次的时候，这样做性能提升不大；
但是当频繁的调用相关数据的时候，如果在Spark调用Hive etl后的数据是就不会出现数据倾斜的问题，这样性能的提升就非常可观了

---

调整并行度

原理：调整并行度，分散同一个 Task 的不同 Key 到更多的Task

注意：调整并行度不一定是增加，也可能是减少，目的是为了，分散同一个 Task 中倾斜 Key 到更多的Task，所以如果减少并行度可以实现，也是可以的

对于Spark Sql配置下列参数spark.sql.shuffle.partitions

对于RDD,可以对shuflle算子设置并行度，如

 rdd.map(p=>(p._1,1)).reduceByKey( (c1, c2)=>(c1+c2),1000)
 
 默认使用HashPartitioner，并行度默认为 spark.default.parallelism
 def reduceByKey(func: (V, V) => V, numPartitions: Int): RDD[(K, V)] = self.withScope {
    reduceByKey(new HashPartitioner(numPartitions), func)
  }

优点：实现起来比较简单，理论上可以有效缓解和减轻数据倾斜的影响。

方案缺点：只是缓解了数据倾斜而已，没有彻底根除问题，对于某个key倾斜的情况毫无办法，因为无论你设置并行度为多少，相同的key总会在同一个partition中

一般如果出现数据倾斜，都可以通过这种方法先试验几次，如果问题未解决，再尝试其它方法。

适用场景少，只能将分配到同一 Task 的不同 Key 分散开，但对于同一 Key 倾斜严重的情况该方法并不适用。
并且该方法一般只能缓解数据倾斜，没有彻底消除问题。

根据我工作遇到倾斜问题的来看，这方法有一定效果但是作用不大，还没试过只调整并行度就直接解决的案例。

---

自定义分区函数

原理：使用自定义的 Partitioner（默认为 HashPartitioner），将原本被分配到同一个 Task 的不同 Key 分配到不同 Task。

class CustomerPartitioner(numParts: Int) extends Partitioner{
  override def numPartitions: Int = numParts

  override def getPartition(key: Any): Int = {
    //自定义分区
    val id: Int = key.toString.toInt
    //这里自定义分区的方式比较灵活，可以根据key的分布设计不同的计算方式
    if (id <= 10000) //10000 以内的id容易出现倾斜
      return new java.util.Random().nextInt(10000) % numPartitions
    else
      return id % numPartitions
  }
}

rdd.map(p=>(p._1,1)).groupByKey(new CustomerPartitioner(10))

适用场景：大量不同的 Key 被分配到了相同的 Task 造成该 Task 数据量过大。

优点：不影响原有的并行度设计。如果改变并行度，后续 Stage 的并行度也会默认改变，可能会影响后续 Stage。

缺点：适用场景有限，只能将不同 Key 分散开，对于同一 Key 对应数据集非常大的场景不适用。
效果与调整并行度类似，只能缓解数据倾斜而不能完全消除数据倾斜。
而且不够灵活，需要根据数据特点自定义专用的 Partitioner(即需要非常了解key的分分布)。

---

ReduceJoin转MapJoin（Broadcast ）

原理：如果一个 RDD 是比较小的，则可以采用广播小 RDD 全量数据 +map 算子来实现与 join 同样的效果，也就是 map join，此时就不会发生 shuffle 操作，也就不会发生数据倾斜。

示意图

ReduceJoin转MapJoin.png

优点：对 join 操作导致的数据倾斜，效果非常好，因为根本就不会发生 shuffle，也就根本不会发生数据倾斜。

缺点：要求参与 Join的一侧数据集足够小，并且主要适用于 Join 的场景，不适合聚合的场景，适用条件有限。

通过 Spark 的 Broadcast 机制，将 Reduce 侧 Join 转化为 Map 侧 Join，避免 Shuffle 从而完全消除 Shuffle 带来的数据倾斜。

Web UI的DAG图如下

ReduceJoin.png

MapJoin

MapJoin.png

相关参数：

将 Broadcast 的阈值设置得足够大
SET spark.sql.autoBroadcastJoinThreshold=10485760

---

局部聚合+全局聚合

原理：将原本相同的 key 通过附加随机前缀的方式，变成多个不同的 key，就可以让原本被一个 task 处理的数据分散到多个 task 上去做局部聚合，进而解决单个 task 处理数据量过多的问题。接着去除掉随机前缀，再次进行全局聚合，就可以得到最终的结果。

 rdd1
      .map(s=>(new Random().nextInt(100)+"_"+s._1,s._2))//添加前缀
      .reduceByKey(_+_)//局部聚合
      .map(s=>(s._1.split("_")(1),s._2))//去除前缀
      .reduceByKey(_+_)//全局聚合

局部聚合+全局聚合.png

适用场景：对 RDD 执行 reduceByKey 等聚合类 shuffle 算子或者在 Spark SQL 中使用 group by 语句进行分组聚合时，比较适用这种方案。

优点：对于聚合类的 shuffle 操作导致的数据倾斜，效果是非常不错的。通常都可以解决掉数据倾斜，或者至少是大幅度缓解数据倾斜，将 Spark 作业的性能提升数倍以上。

缺点：仅仅适用于聚合类的 shuffle 操作，适用范围相对较窄。如果是 join 类的 shuffle 操作，还得用其他的解决方案。

---

倾斜 key 增加随机前/后缀

实现原理：将倾斜的key 与非倾斜的key 分别与右表join，得到skewedJoinRDD和unskewedJoinRDD最后unoin得到最终结果

skewedJoinRDD部分实现步骤：

1. 将 rddLeft 中倾斜的 key（即 userid1 与 userid2）对应的数据单独过滤出来，且加上 1 到 n 的随机前缀）形成单独的 left: RDD[(String, Int)]。
2. 将 rddRight 中倾斜 key 对应的数据抽取出来，并通过 flatMap 操作将该数据集中每条数据均转换为 n 条数据（每条分别加上 1 到 n 的随机前缀），形成单独的 right: RDD[(String, String)]。
3. 将 left 与 right 进行 Join，并将并行度设置为 n，且在 Join 过程中将随机前缀去掉，得到倾斜数据集的 Join 结果 skewedJoinRDD。

unskewedJoinRDD部分实现步骤:

1. 将 rddLeft: RDD[(String, Int)] 中不包含倾斜 Key 的数据抽取出来作为单独的 leftUnSkewRDD。
2. 对 leftUnSkewRDD 与原始的 rddRight: RDD[(String, String)] 进行Join，并行度也设置为 n，得到 Join 结果 unskewedJoinRDD。
3. 通过 union 算子将 skewedJoinRDD 与 unskewedJoinRDD 进行合并，从而得到完整的 Join 结果集。

实现代码

  def prix(): Unit = {
    val sc = spark.sparkContext
    val rddLeft: RDD[(String, Int)] = srdd.rdd.keyBy(row => row.getAs("userid").toString).map(p => (p._1, 1))
    val rddRight: RDD[(String, String)] = srdd.rdd.keyBy(row => row.getAs("userid").toString).map(p => (p._1, p._2.getAs("nums").toString))
    val skewedKeySet = Set("userid1", "userid2") //倾斜的key

    val addList: Seq[Int] = 1 to 24 //右表前缀

    val skewedKey: Broadcast[Set[String]] = sc.broadcast(skewedKeySet) //广播倾斜key

    val addLisPrix: Broadcast[Seq[Int]] = sc.broadcast(addList) //广播右表前缀

    val left: RDD[(String, Int)] = rddLeft
      .filter(kv => skewedKey.value.contains(kv._1)) //左表筛选倾斜key
      .map(kv => (new Random().nextInt(24) + "," + kv._1, kv._2)) //倾斜key增加前缀

    val leftUnSkewRDD: RDD[(String, Int)] = rddLeft
      .filter(kv => !skewedKey.value.contains(kv._1)) //左表筛选非倾斜key
    val right: RDD[(String, String)] = rddRight
      .filter(kv => skewedKey.value.contains(kv._1)) //右表筛选倾斜key
      .map(kv => (addLisPrix.value.map(str => (str + "," + kv._1, kv._2)))) //右表倾斜key每个增加1 to 24 的前缀
      .flatMap(kv => kv.iterator)


    val skewedJoinRDD: RDD[(String, String)] = left
      .join(right, 100) //关联操作
      .mapPartitions(kv => kv.map(str => (str._1.split(",")(1), str._2._2))) //去除前缀

    val unskewedJoinRDD: RDD[(String, String)] = leftUnSkewRDD
      .join(rddRight, 100) //非倾斜关联操作
      .mapPartitions(kv => kv.map(str => (str._1, str._2._2)))
    
    //合并倾斜与非倾斜key
    skewedJoinRDD.union(unskewedJoinRDD).collect().foreach(println _)
  }

用场景：两张表都比较大，无法使用 Map 侧 Join。其中一个 RDD 有少数几个 Key 的数据量过大，另外一个 RDD 的 Key 分布较为均匀。

优点：相对于 Map 侧 Join，更能适应大数据集的 Join。
如果资源充足，倾斜部分数据集与非倾斜部分数据集可并行进行，效率提升明显。
且只针对倾斜部分的数据做数据扩展，增加的资源消耗有限。

缺点：如果倾斜 Key 非常多，则另一侧数据膨胀非常大，此方案不适用。
而且此时对倾斜 Key 与非倾斜 Key 分开处理，需要扫描数据集两遍，增加了开销。

---

倾斜表随机添加n种随机前缀，小表扩大n倍

原理：将包含倾斜 key 的rdd通过附加随机前缀 1 to n 变成不一样的 key，然后就可以将这些处理后的 “不同key” 分散到多个 task 中去处理。通过每条记录增加前缀 1 to n 扩容非倾斜 rdd ，然后再join

(此方法还有一个变体，就是将倾斜的key拉出来添加n种随机前缀，小表扩大n倍，倾斜与非倾斜分开来，类似上一个例子)

实现原理

  def prixAndMul(): Unit = {
    val sc = spark.sparkContext
    val rddLeft: RDD[(String, Int)] = srdd.rdd.keyBy(row => row.getAs("userid").toString).map(p => (p._1, 1))
    val rddRight: RDD[(String, String)] = srdd.rdd.keyBy(row => row.getAs("userid").toString).map(p => (p._1, p._2.getAs("nums").toString))
    val skewedKeySet = Set("userid1", "userid2") //倾斜的key

    val addList: Seq[Int] = 1 to 24 //右表前缀

    val addLisPrix: Broadcast[Seq[Int]] = sc.broadcast(addList) //广播右表前缀

    val left: RDD[(String, Int)] = rddLeft
      .map(kv => (new Random().nextInt(24) + "," + kv._1, kv._2)) //倾斜key增加前缀
    
    val right: RDD[(String, String)] = rddRight
      .map(kv => (addLisPrix.value.map(str => (str + "," + kv._1, kv._2)))) //右表倾斜key每个增加1 to 24 的前缀
      .flatMap(kv => kv.iterator)
    
    val resultRDD: RDD[(String, String)] = left
      .join(right, 100) //关联操作
      .mapPartitions(kv => kv.map(str => (str._1.split(",")(1), str._2._2))) //去除前缀
    
    resultRDD.collect().foreach(println _)
  }
  
  

优点：对 join 类型的数据倾斜基本都可以处理，而且效果也相对比较显著，性能提升效果非常不错。

缺点：该方案更多的是缓解数据倾斜，而不是彻底避免数据倾斜。而且需要对整个 RDD 进行扩容，对内存资源要求很高。

该方案至少能保证程序能够运行完成，速度的话看实际情况了，毕竟先跑通再优化。

---

过滤少数导致倾斜的 key

对于数据要求不是很严谨的情况，可以通过抽样获取倾斜key ，然后直接过滤掉

---

关于数据倾斜，没有一个固定的解决办法，要根据数据的实际情况，灵活采用各种方案解决