Spark任务倾斜实战：从Spark UI定位到两阶段聚合与Join优化

前几天凌晨，一个离线任务在跑大表关联时又卡住了。作业进度条停在某个Stage不动，一个Task跑了快六个小时，其余几十个几十秒就跑完了。这种现象在大数据环境里太常见了：代码没报错，资源也没少给，就是干等。本质只有一个——Spark任务倾斜。

这篇文章不聊教科书定义，我把从定位到解决的完整路径写出来，包括Spark UI怎么看、两阶段聚合怎么写、Join场景怎么处理、以及我踩过的好几个坑。适合刚接手数据开发、被Spark作业杀时间折磨的同学，也适合想系统处理倾斜问题的从业者。我会尽量把每一步的“为什么”讲清楚，而不是丢给你一堆参数让你自己试。

1. 先看现象：一个跑了6小时都不结束的Stage

1.1 分布式计算的木桶效应

Spark任务倾斜的本质，是数据在Partition之间分布严重不均。分布式计算讲的是“分而治之”，一个Job会被拆成多个Task并行执行，而整个Job的运行时间，取决于最慢的那个Task。这就像十个窗口办理业务，九个窗口前面只排了两三个人，最后一个窗口乌泱泱排了几百号人，你的处理速度永远被那个窗口卡住。

很多人在这种情况下第一反应是加资源：加Core、加Executor内存、提升并行度。结果往往很讽刺——加完资源之后，最慢的Task还是那一个，只是其他Task更快地跑完然后空转等待。从Spark UI上看，整个Stage依然维持在最慢Task的运行时长上。资源加了，钱花了，问题纹丝不动，这就是典型的木桶效应。

倾斜导致的直接问题有三个：单个Task处理的数据量超出正常水平，内存压力陡增甚至直接OOM；同一个Stage内Task之间耗时差异巨大，几个小时的运行时间里大部分机器在空转；严重时还会拖垮整批任务，导致下游依赖延迟，甚至触发重试导致重复计算。

1.2 常见的倾斜类型清单

在实际问题排查中，倾斜并不是只有一种表现。我先列一个清单，你对照自己的作业就能猜个八九不离十。

• Key分布不均：最常见的一种。业务表里某个字段值占比过高，比如订单表里的用户ID是“测试用户”，或者某个渠道ID占了半张表。
• Join键膨胀：小表Join大表，当小表的关键值在业务上被大量复制时，关联后中间结果成倍爆炸。典型场景是拉链表与流水表关联。
• 空值聚集：很多人会漏掉这一点。空值或默认字符串会被Shuffle到同一个Partition，形成假性倾斜。它和数据本身的特点无关，纯粹是Shuffle分区算法的锅。
• 数据粒度不均：比如按省份聚合，某几个省份数据量和其他省份完全不在一个量级。这种倾斜在业务上无法通过加盐彻底消失，只能缓解。

我的经验是：拿到一个性能问题，先不要急着调参数或者改业务逻辑，而是先判断它属于哪一类倾斜。定位错了方向，后面所有的优化动作都是在浪费时间和计算资源。

2. 从Spark UI里锁定罪魁祸首：倾斜定位三步法

2.1 第一步：按Duration倒序盯住最慢的Task

Spark UI是排查倾斜的第一现场。打开正在运行的Job，找到耗时最长的Stage，点进Stage详情，看到Task列表后，按Duration做倒序排列。正常情况下，一个健康Stage的Task耗时分布应该是比较集中的，就算有波动也不会出现数量级的差异。

如果某一两个Task的Duration是其他Task的几十倍，整个Stage倒数第二慢的Task也已经结束了，只有它还挂着，说明倾斜的概率非常大。这里要特别强调一个细节：很多人在这个环节看到的是“某个Task运行了6小时”就直接开始改代码，但我建议你还是冷静点，先把鼠标移到那个Task上，看看它的分区编号和Input/Shuffle Read指标。

2.2 第二步：看Shuffle Read指标和中间文件

点开慢Task的详情后，重点看两个指标：Shuffle Read Size / Records。如果一个Task读入的Shuffle数据量是其他Task的几十倍甚至上百倍，那就不只是“运行慢”的问题，而是这个分区在Shuffle阶段就分到了一堆数据。这基本就是倾斜的直接证据。

举个例子：一个Stage有200个Task，其中199个Task的Shuffle Read Size都在几十MB，只有一个Task读了好几个GB，那不用再做其他分析，问题就出在这个分区上。至于为什么它读得多，可以继续往上找上一个Stage的输出情况，或者直接对数据做一次聚合采样。

有时你还要看Shuffle Write信息。如果某个Task写出的数据量异常大，说明它在下游也会继续产生倾斜，问题会沿着血缘关系往下游传递。这也是我经常强调“倾斜要尽早处理”的原因——它不只影响当前Stage，还会传染给后面的Stage。

2.3 第三步：对Key做一次分布体检

从UI上确认了倾斜现象后，再用SQL对相关表或者DataFrame做一次Key分布体检，这一步能把问题从“怀疑”变成“实证”。

-- 以订单表为例，检查用户ID分布 SELECT user_id, COUNT(*) AS cnt FROM dwd_order_detail GROUP BY user_id ORDER BY cnt DESC LIMIT 20;

跑出来的结果如果出现“某一个user_id的量超过总量10%，或者比第二名多出一个数量级”，那这个Key就是倾斜源。如果分布相对平坦，只是少数分区偶发数据偏多，那可能是分区函数或文件切分导致的问题，两者处理思路完全不同。

这里还要提醒一句：查看Top Key分布时，别只看COUNT，还要看关联之后可能产生的记录数。比如事实表的某个外键只有1万条，但它关联的维表里这个键对应的记录有100万条，Join之后同样会爆炸。所以最好把两张表的关键字段都查一遍，心里有数后再动手。

3. 聚合场景的硬解法：两阶段聚合的完整替换方案

3.1 两阶段聚合的核心思想

如果倾斜发生在单纯的聚合场景，比如groupBy后做sum、count，那么最通用、最稳定的解法就是两阶段聚合。

思路也很直白：既然某个Key太大导致单个Partition扛不住，那么我就在这个Key上加一个随机前缀，把它“打散”到多个Partition里，先做一次局部聚合；等局部聚合完成后，去掉随机前缀，再做一次全局聚合。相当于把一个大部队拆成多个小队分别清点人数，最后再把每个小队的清点结果汇总起来。

你可以理解为：原来所有人都挤在一个窗口排队，现在我在每个人的票据上随手写一个1到10的编号，让他们去10个不同窗口排队。等每个窗口各自处理完后，我再去核对这10组结果的汇总值。业务上原本需要按某个Key精确聚合，加盐只影响Shuffle分配，不影响最终结果。

3.2 完整的Scala实现与注意事项

下面是一个可复制的两阶段聚合示例：

import org.apache.spark.sql.functions._ import org.apache.spark.sql.SparkSession // 假设订单明细表 orders 结构为：order_id, user_id, amount val df = spark.table("dwd_order_detail") // 一阶段：加随机盐，按 user_id + salt 聚合 val saltedAgg = df .withColumn("salt", (rand() * 9).cast("int")) // 0~9，共10个盐 .groupBy("user_id", "salt") .agg(sum("amount").as("part_amount")) // 二阶段：去掉盐，按 user_id 再次聚合 val result = saltedAgg .groupBy("user_id") .agg(sum("part_amount").as("total_amount")) result.write.mode("overwrite").saveAsTable("ads_user_amount")

这里的核心是聚合函数必须满足“可加性”。sum、count、max、min都能安全地拆成两阶段；countDistinct不能直接这样拆，两个分区里可能都有重复值，局部去重之后全局再去重会漏算，需要提前把countDistinct改写为approx_count_distinct，或者用其他专门方案。

盐的个数怎么定？一般来说，先看倾斜Key的数量级再决定。如果倾斜Key只比普通Key多几十倍，盐取10就足够；如果多几百倍，可以取50甚至100。盐越多打散得越均匀，但也要付出额外的序列化和网络传输代价，没必要盲目加大。

3.3 什么时候不适合两阶段聚合

两阶段聚合不是万能药。如果业务本身的聚合结果是精确去重数、或者涉及复杂的自定义聚合函数，强行拆解会导致结果错误。

另外，如果倾斜Key在总数据量里的占比实在太高，比如某个Key占了全表99%，那打散下去每个盐上仍然数据量巨大，两阶段聚合效果有限。这种情况下要从业务侧入手，确认这个Key是否真的有必要参与聚合，或者是否存在脏数据标识（比如默认值、测试值）可以先过滤。

还有一种情况是，倾斜不是单个Key导致，而是数据分布整体偏斜，少数分区天然承载了大部分数据。这时加盐能缓解，但不能根治。我的建议是，聚合场景优先用加盐两阶段聚合，效果不够再加上过滤和裁剪，必要时调整业务口径。

4. 大表Join才是重灾区：四种倾斜Join对策实测

聚合倾斜还算好解决，Join场景才是真正让人头疼的地方。因为Join不仅要考虑单表的数据分布，还要考虑两张表关联后产生的中间结果量。一个键在小表里只有10行，在大表里被放大到100万行，这个倾倒是数据膨胀型的，加盐也要格外小心。

4.1 方案A：Broadcast小表，让Shuffle彻底消失

如果其中一张表很小，直接使用Broadcast Join，把整张小表复制到每个Executor的内存里，彻底避免Shuffle阶段的倾斜。Spark默认阈值是10MB，你可以通过参数调整：

spark.sql.autoBroadcastJoinThreshold=20MB

我个人不太建议把这个阈值调到很大，比如50MB以上，因为每个Executor都要存一份整表，几百个Executor就是几百份复制，Driver端内存和Executor存储内存都会被压垮。小表Broadcast只能用在小表真的小的情况，一般表体量在50MB以内比较稳妥。

判断哪张表该Broadcast，就是看哪张表是维表或者配置表，数据量小且变化不频繁。把大表留在原处，小表广播过去，Join在Map端完成，Shuffle都没了，倾斜自然也无从谈起。

4.2 方案B：倾斜Key单独走Broadcast路径

倾斜Key只占整体的小部分，大部分Key是正常的。这时不要对全表加盐，只对倾斜Key单独处理，这样代价最小。

思路是三步走：先找到倾斜Key列表，再把这些Key的数据抽取出来单独Join，最后把结果Union回去。

import org.apache.spark.sql.functions._ val skewKeys = List("key_001", "key_002") // 通过分布体检得到 // 大表拆分：倾斜部分 val bigSkew = bigDF.filter(col("key").isin(skewKeys: _*)) val bigNormal = bigDF.filter(!col("key").isin(skewKeys: _*)) // 小表同样拆分 val smallSkew = smallDF.filter(col("key").isin(skewKeys: _*)) val smallNormal = smallDF.filter(!col("key").isin(skewKeys: _*)) // 正常部分走普通Join val normalJoin = bigNormal.join(smallNormal, Seq("key"), "left") // 倾斜部分强制Broadcast小表，且只关联倾斜Key val skewJoin = bigSkew.join(broadcast(smallSkew), Seq("key"), "left") // 合并结果 val result = normalJoin.union(skewJoin)

这个方案的优点很明显：倾斜部分的数据量被限制在很小范围内，即使小表Broadcast也不会撑爆内存。缺点是代码逻辑变复杂，需要维护倾斜Key列表。如果倾斜Key会随业务动态变化，要定期重新计算分布。

4.3 方案C：全局加盐再拆解

当两张表规模都很大，且倾斜Key很多时，可以考虑“大表加随机盐、小表膨胀”的全量处理方案。思路是让小表对倾斜Key做膨胀复制，给每条倾斜Key生成多个随机后缀，同时大表侧也把这些Key打上随机后缀，然后Join时按加盐后的Key做关联。

import org.apache.spark.sql.functions._ val saltNum = 20 // 大表侧：倾斜Key加随机后缀 val bigTableSalted = bigDF .withColumn( "join_key_salt", when(col("key").isin(skewKeys: _*), concat(col("key"), lit("-"), (rand() * saltNum).cast("int"))) .otherwise(col("key")) ) // 小表侧：倾斜Key膨胀saltNum份 val smallTableSalted = smallDF .filter(col("key").isin(skewKeys: _*)) .crossJoin(spark.range(saltNum).toDF("salt")) .withColumn("join_key_salt", concat(col("key"), lit("-"), col("salt"))) // 非倾斜部分保持不变 val smallTableNormal = smallDF .filter(!col("key").isin(skewKeys: _*)) .withColumn("join_key_salt", col("key")) val smallTableAll = smallTableSalted.union(smallTableNormal) val result = bigTableSalted.join(smallTableAll, Seq("join_key_salt"), "left")

这个方法有一个经典风险：小表膨胀后，Join结果的总行数并没有变化，但Shuffle和中间计算量变大。盐个数越大，倾斜Key被分散得越均匀，小表膨胀倍数也越高，内存压力也越大。一般我建议盐数先取20，观察结果后再决定是否调整。

还要注意Join后可能会产生重复匹配，尤其是非倾斜部分和倾斜部分Union时如果键值重叠，需要保证两边数据隔离干净，最好在倾斜部分和非倾斜部分之间加一个业务维度标志，避免串数据。

4.4 方案D：Spark 3.x的AQE自动倾斜Join优化

如果你用的是Spark 3.2以上版本，还有一个“躺平”选项：开启AQE（Adaptive Query Execution），让Spark自动检测倾斜Partition并拆分。

spark.sql.adaptive.enabled=true spark.sql.adaptive.skewJoin.enabled=true spark.sql.adaptive.skewJoin.skewedPartitionFactor=5 spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes=256MB

skewedPartitionFactor的意思是，当一个分区的数据量是中位数的5倍以上时判定为倾斜；skewedPartitionThresholdInBytes是触发判定的大小阈值。开启后，Spark会在运行时自动把倾斜的分区分成多个子分区处理，而不需要你手动改代码。

实测下来，AQE对简单Join场景的效果不错，但对有复杂业务约束的作业仍然可能不够，而且它会额外消耗运行时计算资源。我的建议是：把AQE作为默认开启项，但它解决不了所有问题，尤其是业务层面的膨胀型倾斜，最终还是需要手动方案。手动优化的作业，尽量把AOE关闭或者在测试环境对比验证，防止两者互相干扰。

4.5 四种方案的选型对比

我把四种方案放在一张表里，方便你快速决策。

不要一上来就选最重的方案。先确认倾斜Key的分布集中度，再决定是走方案B还是方案C。很多场景其实是方案B三行代码就能解决的事，犯不上全局大改。

5. 倾斜问题的前置预防：分区设计与写入优化

5.1 分区键的选取决定了后续80%的命运

做数据开发时间越长，我越觉得“后期解决”不如“前期预防”。很多倾斜问题在建模和写入阶段就已经埋下了，只是到跑数那天才暴露。

分区的选择非常关键。如果你的业务表固定按某个字段做分区或分桶，而这个字段本身存在极度不均匀的分布，那么无论后面怎么优化代码，倾斜都很难绕开。比如按渠道ID分区，某个渠道的活动流量可能是其他渠道的几百倍，那这个分区天然就是倾斜的。

我更推荐把分区键挑在相对均匀的业务字段上，比如日期、自然增长的ID、按业务语义分拆的枚举值。遇到必须用高基数字段做分桶的场景，也要先做一次分布抽样，确认没有单点过载后再定方案。

另外，在建表时合理使用分桶表（Bucket表）可以在源头控制数据分布。分桶数应该结合数据量预估，让每个桶的数据量落在合理范围，比如每个桶几十MB到几百MB之间。分桶太多会产生大量小文件，分桶太少又会重新制造倾斜，需要拿历史数据多测几个值。

5.2 写入阶段的动态分区、小文件与shuffle并行度

写入阶段对倾斜的影响也经常被低估。动态分区写入时，Spark会按照分区字段值将数据划分，如果某个分区值的数据量过大，写入这个分区的Task就会拖慢整个Job。同时，动态分区还容易产生小文件，小文件反过来又导致下游读取时出现新的数据倾斜。

控制这个链条的关键参数是spark.sql.shuffle.partitions。很多人贪心地把并行度开到2000，想让数据尽量分散，结果每个Task只处理很少的数据，调度开销和网络开销反而成了主要成本，作业并没有变快。我说一个相对通用的经验：并行度不是越大越好，而是要和单任务处理数据量匹配。

可以在写入前用repartition()显式控制分区数，而不是只依赖全局并行度。比如对最终结果做一次按日期字段或业务主键的repartition，再写入目标表。这样既控制了文件大小，也防止写数据时出现尾差倾斜。

5.3 把“后置兜底”变成“前置设计”

我见过很多团队，把大量精力花在跑完任务后调优救火，却很少在设计阶段做数据分布评估。这里我建议在数据开发流程里加一个固定节点：建表和编写核心ETL之前，先做数据分布健康检查。

用一条简单的SQL就能评估：

SELECT count(*) AS total_cnt, count(DISTINCT key) AS distinct_keys, max(cnt_per_key) AS max_key_cnt FROM ( SELECT key, count(*) AS cnt_per_key FROM your_table GROUP BY key ) t;

如果max_key_cnt / total_cnt超过0.1，或者和均值的比值超过一个数量级，那就要在设计阶段优化了。提前发现问题，比事后加盐、改代码、调资源要省事一百倍。

6. 几个容易反复踩的坑和我的最终建议

6.1 盲目调大并行度与内存的误区

我见过不止一次，有人面对倾斜问题，第一反应是把spark.sql.shuffle.partitions从200调到800，或者把Executor内存调大两三倍。结果呢？慢的Task依然慢，只是其他Task跑得更快，空转更多，最终作业时间几乎没变。

原因很简单：倾斜问题是数据分布问题，不是并发度和资源问题。你把并行度调大，倾斜的那个Key还是会在某个分区上集中，除非分区数超过那个Key的记录数并做了合适的Salt分配，否则问题不会消失。内存调大，最多是让原本会OOM的Task不至于挂掉，但计算时间还是被卡住。先定位数据分布，再改代码，这是最省时间的路径。

6.2 加盐之后结果错误：一道隐藏的算术题

两阶段聚合和加盐Join最怕什么？最怕结果对不上，而且你在生产环境里很难发现，因为数据量大，没人一行一行去核对。我就遇到过countDistinct加盐后结果少了几万条的情况，排查了半天才发现是聚合函数不可拆分导致的。

再提醒一次：sum、count、max、min可以合并，countDistinct不行。如果你非要用加盐方案做精确去重，要先把去重逻辑手动拆成“先按盐分组去重，再全局去重”的语义，相当于自己实现一个可累加的去重器，复杂度会明显上升。大多数业务其实能接受approx_count_distinct的误差率，直接换函数反而更安全。

6.3 空值带来的假倾斜现象

空值导致的倾斜是最容易被误判的。早期我做排查时，看到一个Task的Shuffle Read数据量特别大，第一反应是某些业务Key分布不均，结果查了半天，发现是user_id字段有大量空字符串和NULL，这些值在哈希分区时全部落到了同一个分区。

处理起来很简单：对NULL或默认值单独过滤，然后在业务逻辑允许的情况下直接排除；如果业务需要保留这些数据，就给它们加随机盐，让它们分散到不同分区。但要注意，如果下游还要按这些Key聚合，加盐后要记得把它们聚回来。

6.4 我个人总结的排查清单

文章最后，分享一个我自己一直在用的处理顺序，虽然不算什么高深方法论，但真能帮你在紧急排查时少走弯路。

• 先看Spark UI，找出运行时间异常长的Stage和Task，确认是最慢点。
• 再看Task的Shuffle Read Size，和同Stage其他Task做对比。
• 用SQL对关联键和聚合键做分布体检，拿到Top Key占比。
• 区分场景：纯聚合走两阶段聚合，Join场景按倾斜Key占比选择方案A到D。
• 参数调整放在最后一步，改完必须对比前后两个版本的“Stage执行时间”和“Task最大耗时”。
• 生产环境上线前，至少在一份采样数据上验证结果集数量是否一致，尤其是加盐和拆key后。

我现在处理任何Spark性能问题的顺序，永远是先看UI、再算分布、然后才动代码。前两步花不了10分钟，却能让你在最后一小时里不白忙活。如果你也被Spark任务倾斜折磨过，试着按这条链路重新走一遍，多半能少掉几根头发。