[Spark] 数据倾斜, 原因确定, 解决方法
判断是否存在倾斜
任务执行时间
- 观察 Spark 任务中各个阶段的执行时间。如果某些任务或阶段的执行时间明显长于其他任务或阶段，可能存在数据倾斜。
- 例如，在 Spark UI 中，如果看到某些 Task 执行时间远远超过平均执行时间，就可能是数据倾斜的迹象。
资源使用情况
- 检查资源使用的不均衡性。例如，某些节点的 CPU 利用率、内存使用量等明显高于其他节点。
- 可以通过监控系统（如 YARN 等）查看节点的资源使用情况。
数据分布分析
- 对数据进行采样或统计，查看数据在各个键或分区上的分布情况。
- 例如，通过简单的代码计算每个键出现的频率，如果某些键的频率远远高于其他键，可能存在倾斜。
查看日志
- Spark 任务的日志中可能会有关于数据倾斜的提示或警告信息。
任务进度
- 观察任务的进度，如果大部分任务已经完成，但仍有少数任务进度缓慢，可能是数据倾斜导致的。
监控指标
- 一些监控工具可以提供关于数据倾斜的指标，如字节倾斜度、记录数倾斜度等。
例如，假设您有一个 Spark 作业处理用户行为数据，通过 Spark UI 发现某个 Stage 中的 Task 执行时间分布极不均匀，大部分 Task 在几分钟内完成，而有几个 Task 却需要数十分钟甚至更长时间，这就很可能是因为处理某些用户的行为数据时出现了数据倾斜。
又如，对数据进行简单的采样统计，发现某个产品的购买量在数据中占比过高，远超过其他产品，这也可能表明在处理该产品相关数据时会有倾斜问题。
解决方法
在代码中处理数据倾斜问题可以采取以下几种常见的方法：
1. 使用随机前缀
在进行关联操作或聚合操作之前，为数据的键添加随机前缀。这样可以将原本集中在某些特定键上的数据分散开，减少倾斜程度。处理完之后再去掉前缀。
示例代码：
import scala.util.Random // 为键添加随机前缀 def addRandomPrefix(key: String): String = { val randomPrefix = Random.nextInt(100).toString randomPrefix + "_" + key } // 去掉随机前缀 def removeRandomPrefix(prefixedKey: String): String = { prefixedKey.split("_").tail.mkString("_") } 2. 二次聚合
先进行局部聚合，再进行全局聚合。这样可以减少数据量，缓解数据倾斜。
二次聚合（局部聚合+全局聚合）通常用于解决在对 RDD 执行 reduceByKey 等聚合类 shuffle 算子或者在 Spark SQL 中使用 GROUP BY 语句进行分组聚合时出现的数据倾斜问题。其核心实现思路是进行两阶段聚合，具体步骤如下：
- 第一次聚合（局部聚合）：先给每个键都打上一个随机数（例如 10 以内的随机数），此时原先相同的键就会变成不同的键，比如
(hello, 1)(hello, 1)(hello, 1)(hello, 1)，可能会变成(1_hello, 1)(1_hello, 1)(2_hello, 1)(2_hello, 1)。接着对打上随机数后的数据，执行reduceByKey等聚合操作，进行局部聚合，局部聚合结果可能会变成(1_hello, 2)(2_hello, 2)。 - 第二次聚合（全局聚合）：将各个键的前缀去掉，得到类似于
(hello, 2)(hello, 2)的结果，然后再次进行全局聚合操作，就可以得到最终结果，比如(hello, 4)。
通过将原本相同的键附加随机前缀的方式，使其变成多个不同的键，这样就可以让原本被一个 task 处理的数据分散到多个 task 上去做局部聚合，进而解决单个 task 处理数据量过多的问题。然后去除随机前缀，再次进行全局聚合，得到最终的结果。
以下是使用 Scala 实现二次聚合的示例代码：
import org.apache.spark.{SparkConf, SparkContext} import scala.util.Random object TwoStageAggregationExample { def main(args: Array[String]): Unit = { val conf = new SparkConf().setAppName("TwoStageAggregationExample").setMaster("local[2]") val sc = new SparkContext(conf) // 准备数据 val array = new Array[Int](10000) for (i <- 0 to 9999) { array(i) = new Random().nextInt(10) } // 生成一个 RDD val rdd = sc.parallelize(array) // 所有 key 加一操作 val mapRdd = rdd.map((_, 1)) // 加随机前缀 val prefixRdd = mapRdd.map(x => { val prefix = new Random().nextInt(10) (prefix + "_" + x._1, x._2) }) // 加上随机前缀的 key 进行局部聚合 val tmpRdd = prefixRdd.reduceByKey(_ + _) // 去除随机前缀 val newRdd = tmpRdd.map(x => (x._1.split("_")(1), x._2)) // 最终聚合 newRdd.reduceByKey(_ + _).foreach(print) } } 上述代码首先生成了一个包含随机整数的 RDD，然后给每个键加上一个随机前缀，进行局部聚合，去掉前缀后再进行全局聚合，从而实现二次聚合的过程。
这种方法对于聚合类的 shuffle 操作导致的数据倾斜效果较好，通常可以解决或大幅度缓解数据倾斜问题，提升 Spark 作业的性能。但它仅适用于聚合类的 shuffle 操作，适用范围相对较窄，如果是 join 类的 shuffle 操作，还需使用其他解决方案。
3. 过滤异常数据
如果数据中存在一些异常值导致数据倾斜，可以在代码中先对这些异常数据进行过滤或单独处理。
示例代码：
val filteredRdd = rdd.filter(row => { // 定义过滤条件 row.getValue < 10000 }) 4. 调整并行度
增加任务的并行度，使数据更均匀地分布在多个任务中。
示例代码：
val newRdd = rdd.repartition(numPartitions) 5. 使用加盐
类似于添加随机前缀，但是盐值更具规律性。
示例代码：
def addSalt(key: String, salt: Int): String = { key + "_" + salt } 处理数据倾斜需要根据具体的数据特点和业务需求选择合适的方法，有时可能需要综合使用多种方法来达到较好的效果。