三亩地 三亩地SAN MU DI · CODE DIARY
交流投稿
ARTICLE DETAIL

日记详情

真实记录编程学习的某一天，欢迎挑你感兴趣的翻一翻。

首页/学习日记/日记详情

Spark数据倾斜诊断与优化：从现象分析到根治方案实战

发布时间：2026/8/12 10:03:31 三亩地 编程学习日记
Spark数据倾斜诊断与优化：从现象分析到根治方案实战

1. 项目概述：当Spark作业“卡”在某个阶段

如果你用过Spark处理过足够多的数据，大概率会碰到一种让人头疼的情况：任务提交后，大部分任务（Task）都飞快地跑完了，但总有那么几个任务，进度条像蜗牛一样，迟迟不动，甚至直接失败重试，最终拖垮整个作业。恭喜你，你大概率是遇到了数据倾斜（Data Skew）这个分布式计算领域的“经典难题”。这就像一条繁忙的高速公路，大部分车道畅通无阻，但有一两个车道因为事故堵得水泄不通，导致整条路的通行效率急剧下降。

数据倾斜的本质是数据分布不均。在Spark的Shuffle阶段（如groupByKey、reduceByKey、join等操作），数据需要根据Key重新分区（Partition）并分发到不同的计算节点。如果某个或某几个Key对应的数据量异常庞大（我们称之为“热点Key”或“大Key”），那么处理这些Key的任务所在的分区就会负载过重，成为整个作业的瓶颈。这不仅导致任务执行缓慢（长尾任务），还可能因为单个任务处理数据量过大，引发GC频繁、OOM（内存溢出）甚至Executor挂掉等问题。

处理数据倾斜，是每个Spark开发者从入门到精通的必经之路。它考验的不仅是对Spark原理的理解，更是对业务数据特性的洞察和解决问题的工程化思维。接下来，我将结合多年踩坑经验，从现象诊断到根治方案，为你系统性地拆解Spark数据倾斜的应对策略。

2. 核心思路：诊断、规避与根治

处理数据倾斜，不能靠“猜”，必须有一套清晰的思路。我的经验是遵循“先诊断，后治疗”的原则，并且治疗方案分为“规避”和“根治”两个层次。

2.1 现象诊断：如何确认是数据倾斜？

在你开始尝试各种“偏方”之前，首先要确诊。以下是一些明确的信号和诊断方法：

  1. Spark UI观察法：这是最直观的方法。打开Spark作业的Web UI，进入Stages页面，找到执行缓慢的那个Stage。

    • 任务执行时间分布：查看该Stage所有Task的“Duration”分布。如果发现绝大多数Task在几秒内完成，而少数几个Task需要几分钟甚至几小时，基本可以断定是数据倾斜。
    • 数据读写量：查看“Shuffle Read Size”或“Input Size”。倾斜的分区其读取的数据量会远远大于其他分区。
    • 任务失败与重试：倾斜分区的任务更容易因为GC或OOM而失败，在“Task Timeline”中你会看到频繁的失败和重试标记。

  2. 日志分析法：在Executor的日志中，你可能会看到大量关于GC的警告，或者直接报出java.lang.OutOfMemoryError错误，并且这些错误总是集中在某几个任务上。

  3. 代码采样推断法：如果你对数据中的Key分布有疑虑，可以在Driver端写一段简单的采样代码来验证。

    val sampledRDD = yourRDD.sample(false, 0.1) // 采样10%的数据 val keyCounts = sampledRDD.map(x => (x._1, 1)).reduceByKey(_ + _).collect() keyCounts.sortBy(-_._2).take(10).foreach(println) // 打印出现次数最多的前10个Key

    如果发现某个Key的计数远远高于其他，倾斜就坐实了。

注意：诊断时一定要定位到具体的Stage和导致Shuffle的算子（如join,groupBy）。不同的算子，倾斜的处理策略侧重点不同。

2.2 解决思路分层：从“绕开”到“解决”

确诊之后，我们的应对策略可以分为两层：

  • 第一层：规避与缓解。当倾斜程度不特别严重，或者业务上可以接受一定误差时，我们可以采用一些通用性较强的“缓兵之计”，快速让作业跑起来。例如：提高并行度、过滤异常Key、使用Map端聚合等。
  • 第二层：根治与优化。当倾斜非常严重，或者业务要求精确结果时，我们需要从数据或逻辑层面进行根本性的改造。这需要更深入的分析和设计，例如：热点Key分离、打散加盐、使用Skew Join等。

下面，我们就深入这两个层次，看看具体有哪些“武器”可以使用。

3. 通用规避与缓解策略

这些方法通常改动较小，能应对中等程度的数据倾斜，是首先应该尝试的方案。

3.1 提高Shuffle并行度

这是最简单粗暴的方法。通过增加Shuffle后的分区数量，可以让原本集中在一个分区的“大Key”数据被分散到更多的分区中，从而减轻单个分区的压力。

// 设置全局的默认并行度，对RDD和SQL都有效 spark.conf.set(“spark.sql.shuffle.partitions”, “200”) // 默认是200，可以调得更大，比如500-1000 // 或者在特定RDD操作时指定 rdd.reduceByKey(_ + _, 100) // 将shuffle分区数设置为100 df.repartition(200) // 在shuffle前重分区

原理与考量：增加分区数，意味着每个Task处理的数据量变少，降低了OOM风险。但分区数不是越多越好，过多的分区会导致Task调度开销增大，产生大量小文件，可能拖慢速度。通常建议根据数据总量和集群资源来调整，一个分区的数据量在几百MB到1GB之间是比较理想的。

3.2 过滤异常数据（热点Key）

如果业务允许，直接过滤掉导致倾斜的少数几个热点Key，是最有效的办法。例如，在统计用户行为时，如果存在一些爬虫或测试账号（如user_id=0或null）产生了海量日志，可以先将其过滤掉，单独处理或直接丢弃。

val skewedKeys = Set(“hot_key_1”, “hot_key_2”, “”) val filteredRDD = originalRDD.filter { case (key, value) => !skewedKeys.contains(key) } // 对 filteredRDD 进行后续处理 // 对 skewedKeys 对应的数据，可以单独用小批量任务处理，或者直接记录日志忽略

实操心得：这种方法的前提是，这些热点数据对整体分析结果影响不大，或者可以接受其被排除。在过滤前，务必和业务方确认清楚。

3.3 启用Map端聚合（Combiner）

对于reduceByKey、aggregateByKey这类操作，Spark默认会在Map端（Shuffle Write之前）先进行本地聚合（Combiner），这能显著减少Shuffle时需要传输的数据量。对于groupByKey，它不会使用Combiner，所有数据都会通过网络传输，因此在可能的情况下，应优先使用reduceByKey。

// 好的做法：使用reduceByKey，会在map端聚合 rdd.map(x => (x.key, 1)).reduceByKey(_ + _) // 差的做法：使用groupByKey，所有数据都会shuffle rdd.map(x => (x.key, 1)).groupByKey().mapValues(_.sum)

为什么有效：假设一个Key在Map端有1万条记录，经过Combiner本地求和后，可能变成1条记录（一个求和后的值）。这1万到1的压缩，对Shuffle网络和Reduce端压力是巨大的缓解。确保你的聚合操作是可结合和可交换的，这样才能应用Combiner。

4. 高级根治与优化策略

当通用策略无效，或者业务要求必须精确处理所有数据（包括热点Key）时，我们就需要祭出更高级的解决方案。

4.1 两阶段聚合（打散Key/加盐）

这是解决groupBy、reduceByKey等聚合操作倾斜的经典方法，也就是常说的“加盐”（Salting）。其核心思想是：将热点Key附加一个随机前缀（盐），将其数据打散到多个分区进行第一次聚合，然后再去掉前缀进行第二次聚合。

“加盐”是什么意思？简单说，就是给Key“撒点盐”，让它变“咸”（随机），从而分布得更均匀。例如，热点KeyA有1亿条数据。我们给它加上随机前缀，变成A_1,A_2, ...A_n，这样原本属于一个分区的数据，就被分散到n个分区中并行处理了。

操作步骤：

  1. 打散阶段：对原始RDD，给每个Key加上一个随机前缀（如0~n的随机数）。
    val saltNum = 10 // 假设我们打散成10份 val saltedRDD = originalRDD.map { case (key, value) => val salt = (key.hashCode % saltNum + saltNum) % saltNum // 生成0-9的随机盐，这里用hashCode模拟 (s”${key}_${salt}”, value) }
  2. 局部聚合：对打散后的Key进行聚合操作。
    val aggregatedRDD = saltedRDD.reduceByKey(_ + _) // 第一次聚合，在多个分区并行处理
  3. 去盐聚合：将Key的前缀去掉，再次聚合，得到最终结果。
    val finalRDD = aggregatedRDD.map { case (saltedKey, value) => val originalKey = saltedKey.split(“_”)(0) // 去掉随机后缀 (originalKey, value) }.reduceByKey(_ + _) // 第二次聚合，此时每个Key的数据量已经大大减少

注意事项：

  • 盐值数量选择：盐值数量（saltNum）需要根据热点Key的数据量来估算，要保证打散后每个分区的数据量降到可接受范围。通常可以先采样估算热点Key的数据量。
  • 适用场景：主要用于聚合类操作。对于join操作，如果只有一侧有热点Key，也可以使用类似思路，即“倾斜Key分离+随机前缀扩容”。

4.2 处理Join操作的数据倾斜

Join是数据倾斜的重灾区，尤其是当两张表的大小差异很大（大小表Join）且大表存在热点Key时。

4.2.1 使用广播Join（Broadcast Join）

当一张表足够小（通常建议小于几十MB，可通过spark.sql.autoBroadcastJoinThreshold参数控制），可以将其广播到所有Executor节点。这样，每个Task本地都有一份完整的小表数据，直接进行Map端Join，完全避免了Shuffle。

import org.apache.spark.sql.functions.broadcast val largeDF: DataFrame = ... val smallDF: DataFrame = ... val resultDF = largeDF.join(broadcast(smallDF), “key”)

这是最高效的Join方式，应优先考虑。

4.2.2 Skew Join（倾斜Join）

这是Spark 3.0+ 版本中引入的利器。当明确知道某些Key是倾斜的，可以提示Spark对这些Key使用特殊的处理策略。

— 在Spark SQL中，可以使用HINT SELECT /*+ SKEW(‘table_name’, ‘column_name’, (skew_value1, skew_value2…)) */ * FROM table_name; — 或者动态检测倾斜（AQE功能） SET spark.sql.adaptive.skewJoin.enabled = true; SET spark.sql.adaptive.skewJoin.skewedPartitionFactor = 5; // 分区大小中位数5倍以上视为倾斜 SET spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes = 256MB; // 分区大于256MB视为倾斜

启用自适应查询执行（AQE）和Skew Join后，Spark会在运行时自动检测倾斜的分区，并将其拆分成多个子分区，分别与另一侧的表进行Join，最后合并结果。这大大简化了手动处理倾斜Join的复杂度。

4.2.3 手动分离热点Key + 广播/打散

如果版本较低不支持AQE，可以手动实现：

  1. 识别并分离：从大表中将热点Key的数据过滤出来，形成skewDF；剩余正常数据形成normalDF。
  2. 分别处理：
    • 将小表中与热点Key关联的数据也过滤出来（smallSkewDF）。由于热点Key数量少，smallSkewDF通常也很小，可以将其广播，与skewDF进行广播Join。
    • normalDF与过滤掉热点Key后的小表剩余部分（smallNormalDF）进行普通的Shuffle Hash Join或Sort Merge Join。
  3. 合并结果：将两部分Join的结果进行union。
// 1. 识别热点Key (假设已知热点Key是 “A”) val hotKey = “A” // 2. 分离大表 val skewBigDF = bigDF.filter($”key” === hotKey) val normalBigDF = bigDF.filter($”key” =!= hotKey) // 3. 分离小表 val skewSmallDF = smallDF.filter($”key” === hotKey) // 通常很小 val normalSmallDF = smallDF.filter($”key” =!= hotKey) // 4. 分别Join val resultSkew = skewBigDF.join(broadcast(skewSmallDF), Seq(“key”)) // 广播Join val resultNormal = normalBigDF.join(normalSmallDF, Seq(“key”)) // 普通Join // 5. 合并 val finalResult = resultSkew.union(resultNormal)

4.3 优化数据结构与序列化

有时，数据倾斜会加剧由数据本身带来的开销。例如，Key是非常复杂的对象（如嵌套的Case Class），或者Value是巨大的字符串/容器。

  • 使用更紧凑的数据类型：能用Int就不用String，能用数组就不用集合类。这减少了Shuffle过程中数据的序列化/反序列化开销和网络传输量。
  • 使用高效的序列化器：如Kryo。在Spark配置中设置spark.serializer为org.apache.spark.serializer.KryoSerializer并注册需要序列化的类，可以显著提升性能，减少内存占用。
    conf.set(“spark.serializer”, “org.apache.spark.serializer.KryoSerializer”) conf.registerKryoClasses(Array(classOf[MyClass1], classOf[MyClass2]))

5. 实战案例：一个倾斜Join的完整处理过程

假设我们有一个用户行为日志表user_logs（超大，TB级）和一个用户信息维表user_info（较小，GB级）。我们需要根据user_id关联，统计每个城市用户的行为总数。发现user_id为NULL或0（未登录用户）的记录量异常巨大，导致Join倾斜。

步骤1：诊断确认通过采样SQLSELECT user_id, count(1) FROM user_logs GROUP BY user_id ORDER BY count(1) DESC LIMIT 10;确认NULL和0是热点Key。

步骤2：方案设计（采用热点Key分离+广播Join）

  1. 从user_logs中分离出热点Key数据 (log_hot) 和正常数据 (log_normal)。
  2. 从user_info中分离出对应热点Key的数据 (info_hot)。由于热点Key只有两个，info_hot极小。
  3. log_hot与广播的info_hot进行Map端Join。
  4. log_normal与user_info进行普通的Shuffle Join。
  5. 合并两部分结果。

步骤3：代码实现

// 初始化SparkSession val spark = SparkSession.builder().appName(“SkewJoinDemo”).getOrCreate() import spark.implicits._ // 读取数据 val userLogsDF = spark.read.parquet(“hdfs://path/to/user_logs”) val userInfoDF = spark.read.parquet(“hdfs://path/to/user_info”) // 定义热点Key val hotUserIds = Seq(null, “0”).map(_.asInstanceOf[String]) // 注意null的处理 // 分离热点日志数据 val logHotDF = userLogsDF.filter($”user_id”.isInCollection(hotUserIds) || $”user_id”.isNull) val logNormalDF = userLogsDF.filter(!$”user_id”.isInCollection(hotUserIds) && $”user_id”.isNotNull) // 分离热点用户信息 (可能不存在，需要处理) val infoHotDF = userInfoDF.filter($”user_id”.isInCollection(hotUserIds) || $”user_id”.isNull) // 假设user_info中城市信息字段为city，未登录用户城市记为‘UNKNOWN’ val infoHotBroadcast = broadcast(infoHotDF.withColumn(“city”, when($”user_id”.isNull || $”user_id” === “0”, lit(“UNKNOWN”)).otherwise($”city”))) // 处理热点部分Join val resultHotDF = logHotDF.join(infoHotBroadcast, Seq(“user_id”), “left”) .groupBy(“city”).agg(count(“*”).as(“action_count”)) // 处理正常部分Join (启用AQE优化) spark.conf.set(“spark.sql.adaptive.enabled”, “true”) spark.conf.set(“spark.sql.adaptive.skewJoin.enabled”, “true”) val resultNormalDF = logNormalDF.join(userInfoDF, Seq(“user_id”), “left”) .groupBy(“city”).agg(count(“*”).as(“action_count”)) // 合并结果 val finalResultDF = resultHotDF.union(resultNormalDF) finalResultDF.write.parquet(“hdfs://path/to/output”)

步骤4：效果验证提交作业后，观察Spark UI。原先卡在某个Task的Stage应该被分解成多个快速完成的小Task。整体作业时间从小时级下降到分钟级。

6. 配置、监控与预防

6.1 关键配置参数

除了前面提到的，还有一些配置有助于预防和缓解倾斜：

  • spark.sql.adaptive.enabled:强烈建议设置为true。启用自适应查询执行，Spark可以动态调整分区数量、处理倾斜Join、优化Shuffle读写。
  • spark.sql.adaptive.coalescePartitions.enabled: 合并过小的分区，避免任务调度开销。
  • spark.shuffle.spill: 允许将内存中的数据溢写到磁盘，防止OOM。确保设置为true。
  • spark.executor.memoryOverhead: 为堆外内存（如Native堆、线程栈）预留的空间。如果遇到容器被杀（YARN）或Executor丢失，可以适当调大此值。

6.2 监控与告警

建立作业监控体系，关注以下指标：

  • Stage最大/最小任务耗时比。
  • Shuffle读写数据量的分布。
  • Task失败重试次数。
  • GC时间占比。 当这些指标出现异常时，可以及时发出告警，介入分析。

6.3 数据治理预防倾斜

  • ETL过程去重：在数据接入层（ODS）就对明显异常的热点数据进行清洗或采样。
  • 业务逻辑优化：与业务方沟通，是否某些统计可以换一种维度，避免在极易倾斜的Key上聚合。
  • 数据分桶：对于常作为Join Key的字段，可以在数仓层就进行分桶存储，这样Join时可以利用桶的元数据优化。

处理Spark数据倾斜没有银弹，它是一个结合了监控、诊断、调优和业务理解的综合性工作。核心在于，不要一上来就尝试最复杂的方案，而是从最简单的增加并行度、过滤数据开始，逐步深入。理解你的数据，理解你的作业，配合Spark UI这个强大的工具，你就能将这条计算“高速公路”上的堵点一一疏通。

编程学习 学习日记 实战经验
← 返回列表