跑Spark作业，最怕的就是那种“永远跑不完”的task。明明集群有几百个executor，可某个task卡在那里半小时，其他task几秒就结束了。这不是集群坏了，多半是数据倾斜。数据倾斜说白了就是Shuffle阶段分区分配不均，某些key的数据量远大于其他key，导致对应的task承担了绝大多数计算，拖累整个Stage。这个问题解决不了，集群再大也会被几个热点key卡死。
这套内容是我结合自己给多个业务方排查慢任务的实战经验整理的。不管你是刚接触Spark的开发者，还是已经被这类问题折磨过一阵的数据工程师，都可以照着文中的定位思路和方案组合去处理。本文会从倾斜的识别开始讲，再逐步展开两阶段聚合、加盐扩容、Spark自带动态优化等手段，最后附上高频问题的排查速查表。希望能帮你把“看到慢任务就懵”变成“打开UI定位、量化、选方案、落地”一套清晰流程。
1. 数据倾斜的识别与定位
1.1 倾斜产生的根因：Shuffle与分区分配
要解决倾斜，先得清楚它是怎么冒出来的。Spark任务执行到宽依赖算子时，比如groupByKey、reduceByKey、join，都会触发Shuffle。Shuffle过程会把数据按key重新分配到下游的各个分区，默认使用的分区器是HashPartitioner，计算逻辑大致是key.hashCode % numPartitions。这个公式的潜台词是：同一个key，永远只会落到同一个分区，无论它在全集群有多少份。
打个比方，一条单车道本来每分钟能过50辆车，结果有段时间所有车都挤到这条道上，剩下几条车道空空荡荡。数据倾斜就是“热门key把单车道堵死”的情况。常见的热门key有大客户ID、爆款商品ID、某个地域代码，甚至是一大堆null值。当某个key的数据量占到整体数据量的百分之十几甚至几十时，它所在的分区对应的task就要处理海量数据，其他task却在“看戏”。任务能不能跑完，取决于那个最惨的task，而不是平均时间。
需要注意的是，Spark本身并没有把任务“故意”分配给某个task，纯粹是HashPartitioner按key均匀分区的假设失效了。你写代码时的逻辑没有错，分布劣化是数据导致的。所以解决方案也集中在“让热点key不再扎堆”上，这也是后面所有方案的核心逻辑。
1.2 三分钟定位慢Task：Spark UI实战
很多人在集群空闲、资源也够的情况下，任务依然卡住，却不知道问题出在哪个Stage、哪个task，只能盲目加资源。其实Spark UI已经把答案摆在面前了，只是很少有人仔细看。
定位步骤很简单：任务运行中或运行结束后，打开Spark Application的Web UI，切到“SQL”页签。当天跑的SQL或DataFrame任务会按Execution顺序列出，点开最慢的那一个，会看到它拆成了好几个Stage。挨个点进去，重点看两个指标——Task的Duration列和Shuffle Read Size列。正常情况下，所有task的耗时和读写量分布应该比较均匀；如果出现某个task的Shuffle Read达到几十GB，而其他task只有几百MB，或者某个task耗时为其他task的几十倍，基本可以锁死倾斜发生的位置。
更细一点，可以点进某个慢task查看Event Timeline，确认是执行时间长还是GC停顿时间长。如果GC时间占比很高，说明数据量确实大到内存扛不住，这往往也是倾斜导致的；如果GC几乎为零，纯粹是task耗时长，那大概率是计算逻辑本身重，不一定和倾斜相关。用这套方法，我一般几分钟内就能判断任务该不该走倾斜优化，而不是一上来就调并行度。
1.3 先分清“真倾斜”和“假慢任务”
有几年经验的人都明白，不是所有慢任务都是倾斜。常见的“假慢”我见过三类：
第一类是全局资源瓶颈。所有task耗时都偏高，没有明显长尾，这种情况是集群资源不足或executor规格不够，和倾斜无关。第二类是单点故障。某个executor所在节点的磁盘IO异常或网络抖动，导致它上面的task普遍慢，其他节点正常，这类问题重启或换executor就能缓解。第三类是数据爆炸。比如因为过滤条件失效，或者关联用了笛卡尔积，所有task处理的数据量整体上涨，看起来都慢，但分配依然均匀。
判断方法是先看Shuffle Read Size的峰值和均值。如果峰值远大于均值，比如差10倍以上，那是倾斜，走后面的优化方案。如果所有task的数据量同步上涨，说明是数据量或资源问题，优先调整过滤逻辑、增加节点资源、或者考虑缓存中间结果。别把所有慢任务都归因于倾斜，不然会把本来不需要动的代码改出新的问题。
2. 局部聚合与全局聚合：应对groupBy类倾斜
2.1 两阶段聚合的核心原理
定位到某个Stage是groupBy或reduceByKey导致的倾斜后，最常用的一套方案就是两阶段聚合，也叫局部聚合加全局聚合。思路简单说：别让某个热门key直接落进同一个分区，先给它加一个随机前缀，把同一个key拆成多个“临时key”。
拿一个订单表为例，假设product_id = sku_001的记录有1亿条，其他商品各几万条。默认200个分区，sku_001几乎全部进同一个分区，这个分区的task要处理上亿行。给它加一个0到19的随机盐，得到sku_001_3、sku_001_17这样20个新key，它们被均匀打散到20个分区，每个分区只处理五百万分之一的数据，压力瞬间降下来。这个阶段叫局部聚合，得到的结果是20组“sku_001_盐值：局部汇总值”。第二阶段把盐值去掉，恢复成sku_001，再做一次聚合，把20个局部汇总值累加，得到最终结果。
这套思路的好处是逻辑简单，几乎不依赖额外配置，坏处是同一份中间结果会多一次Shuffle。但比起让单个task卡半小时，多一次Shuffle完全值得。
2.2 实操代码与参数设计
落地时我用PySpark写过两阶段聚合，大致是这样：
from pyspark.sql import SparkSession from pyspark.sql.functions import rand, concat, lit, col, regexp_extract spark = SparkSession.builder \ .appName("two_phase_agg") \ .enableHiveSupport() \ .getOrCreate() # 原始订单表 orders = spark.table("orders").select("product_id", "amount") # 1. 加盐：生成 0 到 salt_num-1 的随机整数 salt_num = 20 salted = orders.withColumn("salt", (rand() * salt_num).cast("int")) \ .withColumn("salted_key", concat(col("product_id"), lit("_"), col("salt"))) # 2. 局部聚合 partial = salted.groupBy("salted_key").agg({"amount": "sum"}) \ .withColumnRenamed("sum(amount)", "partial_sum") # 3. 去掉盐前缀 grouped = partial.withColumn( "product_id", regexp_extract(col("salted_key"), r"^(.*)_[0-9]+$", 1) ) # 4. 全局聚合 result = grouped.groupBy("product_id").sum("partial_sum")有几个细节要特别提示。第一，盐值数量salt_num不是越大越好。它决定了第一阶段能把热点key拆成多少份，同时也决定了中间数据会被放大多少倍。按集群并行度和热点key规模来定，通常取并行度的2到5倍就差不多了。比如shuffle分区是200，盐值取400到1000区间比较合理，太大会让Shuffle数据量暴涨，反而拖慢整体。第二，对sum、count、max、min这类可累加的聚合，两阶段聚合直接相加就行；但avg要先算sum和count，最后用总sum除以总count。第三，count distinct不能直接套这个模式，因为第一阶段的局部去重无法在第二阶段直接累加，必须先按key + value做去重，或者考虑用近似去重函数。
2.3 不适合两阶段聚合的情况
两阶段聚合不是银弹，我踩过几次坑之后总结出几个不适合的场景。
第一种是倾斜极其严重的场景，比如某个key占了全量数据的50%以上。第一阶段虽然把它拆散了，但拆出来的每个分区依然很大，第二阶段把这些局部结果汇总时，同样会汇聚到一个分区处理。这个最终分区可能不卡了，但shuffle阶段的内存和网络压力依然极高。第二种是join场景。两阶段聚合只适用于按key做聚合的场景，join本身不能通过“先聚合再聚合”解决，需要换加盐扩容的方案，这在下一节讲。第三种是聚合语义比较特殊，比如需要精确的全局去重、精确的百分位，这类操作在两阶段聚合时容易丢精度。
一个常见的失误是把随机盐加到了所有key上，而不是只加热点key。如果数据里本来只有一两个热点key，给全量数据加盐等于把整份数据人为放大20倍，网络和磁盘IO都会翻数倍，最后可能比倾斜前还慢。更好做法是先做一次groupBy key count拿到热点key清单，只在热点key上加盐，其他key原样处理。这也是为什么我说第一步永远是数据分布量化，方案要跟着分布来。
3. 加盐扩容：搞定Join场景的倾斜
3.1 加盐到底做了什么
聚合场景有两阶段聚合，那join场景怎么办？同样是热门key，两张大表或者一张大表和一张小表做join时，热点key会导致某个分区要处理巨量关联数据。常规解法是加盐扩容。
加盐的核心逻辑是：给大表的热点key加随机前缀，把它拆成N份；同时把维表（通常是小表）中对应key复制成N份，也加上相同的前缀。join时，大表拆出来的“sku_001_0”只跟小表里“sku_001_0”这一份关联，“sku_001_1”跟“sku_001_1”关联，大表的热点key就被分散到了N个分区并行处理。
可以想象一下，原来是一个明星窗口被一万人排队，加盐之后开成了十个窗口，每个窗口只排一千人，处理完再把各窗口结果汇总。对于join来说，不需要像聚合那样做两阶段汇总，每行关联结果独立，因此实现起来更直接。
3.2 条件加盐的实操方案
写代码时，我会把加盐限定在热点key上，避免全量膨胀。下面是一个DataFrame实现。
from pyspark.sql import SparkSession from pyspark.sql.functions import col, lit, concat, when, array, explode spark = SparkSession.builder.appName("salted_join").getOrCreate() big = spark.table("fact_orders").select("sku_id", "user_id", "amount") dim = spark.table("dim_sku").select("sku_id", "sku_name") salt_num = 20 # 热点key清单，由倾斜检测SQL得到 hot_keys = ["sku_001", "sku_002"] salt_cols = [lit(str(i)) for i in range(salt_num)] # 大表：热点key加盐，非热点key盐固定为0 big_salted = big.withColumn( "candidate_salt", when(col("sku_id").isin(hot_keys), array(*salt_cols)).otherwise(array(lit("0"))) ).withColumn("salt", explode(col("candidate_salt"))) \ .drop("candidate_salt") \ .withColumn("join_key", concat(col("sku_id"), lit("_"), col("salt"))) # 小表：同样只对热点key扩容 dim_salted = dim.withColumn( "candidate_salt", when(col("sku_id").isin(hot_keys), array(*salt_cols)).otherwise(array(lit("0"))) ).withColumn("salt", explode(col("candidate_salt"))) \ .drop("candidate_salt") \ .withColumn("join_key", concat(col("sku_id"), lit("_"), col("salt"))) # 执行join result = big_salted.join(dim_salted, "join_key", "inner") \ .drop("salt", "join_key")这段代码的关键在于，热点key拆成20份，非热点key只保留一份，join后的总行数和原始大表行数一致，不会因为非热点key的膨胀导致shuffle量爆炸。实际生产中，热点key往往就那么几个，这样条件加盐的收益非常明显。
3.3 外连接、小表重复数据与膨胀风险
加盐join最容易翻车的地方是join类型和数据的“一对多”关系。
先说外连接。inner join用加盐没问题，因为两边都扩容后，大表拆出的每一份总能匹配到小表对应的一份。left join要小心，大表加盐后如果不匹配，左表会保留但右表字段为空，这本身没问题；但如果右表（小表）的同一个key本来就有多条记录，扩容后每条记录都复制成N份，左表的每一条会匹配到N份右表记录，结果行数直接放大好几倍甚至几十倍。
举个例子，sku_001在小表里有3条价格历史记录，大表有1亿条订单。不加盐join，每笔订单会关联到3条记录，结果是3亿行；加盐20份后，小表变成60条记录，每笔订单会匹配到60条，结果是60亿行。这种膨胀往往比倾斜更可怕。
解决方式是在加盐前先对维表做一次key粒度的聚合，比如按sku_id取最新的记录，或者把多条记录聚合成一条。如果业务上必须保留多行，就只能接受膨胀，但至少要把数据量级算清楚，别等任务跑到一半才发现结果比预期大了几十倍。
还有一点要注意：加盐后join_key从sku_id变成了sku_id_盐值，如果后续逻辑还依赖原始sku_id，需要提前保存好原字段，不要在join后再尝试从join_key里解析回来，否则遇到key本身带下划线的情况很容易解析错。
4. 用好Spark自带的动态优化能力
4.1 并行度不是万能钥匙
很多人的第一反应是调大spark.sql.shuffle.partitions，从默认200调到2000。这个操作对于“每个分区都均匀但分区数太少导致shuffle文件巨大”的任务是有效的，但对倾斜来说基本没用。原因还是HashPartitioner：同一个key永远落在同一个分区，你把分区数从200调到2000，热点key仍然在同一个分区里，只是其他1999个分区更空闲而已。
如果你用的是RDD API，调spark.default.parallelism或reduceByKey(numPartitions)，效果同理。对于倾斜任务，真正有用的并行度调整是把热点key“人为”拆开后再分散，也就是前面说的加盐思路。调大并行度只能作为辅助手段，比如让其他相对不热的数据更分散，减少整体Shuffle压力，但指望它解决核心热点，不现实。
写代码时的合理姿势是用repartition显式指定分区策略，并在必要场景使用自定义Partitioner。比如你知道某个key会在两个分区之间拉扯，可以按业务维度分组后再hash，但自定义Partitioner只针对RDD层有效，DataFrame层还是建议用加盐方案。
4.2 AQE SkewJoin的正确理解
Spark 3.0之后引入了Adaptive Query Execution（AQE），其中有一个内置的倾斜join优化。它会在运行时根据Shuffle阶段的统计信息，自动检测到某个分区明显大于其他分区，然后把这个倾斜分区拆成多个小分区，同时把另一侧对应的key复制多份，相当于在框架层自动完成了“加盐”。
配置大致如下：
spark.sql.adaptive.enabled=true spark.sql.adaptive.coalescePartitions.enabled=true spark.sql.adaptive.skewJoin.enabled=true spark.sql.adaptive.skewJoin.skewedPartitionFactor=5 spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes=256MBskewedPartitionFactor=5表示某个分区大小至少是其他分区中位数的5倍时才认为是倾斜；skewedPartitionThresholdInBytes=256MB表示分区还得超过256MB才算。这两个条件同时满足才会触发自动优化。
但AQE不是万能的。第一，它只对Spark SQL和DataFrame的join生效，对RDD手动写的join无能为力。第二，它对groupBy聚合不生效，聚合倾斜依然需要手动两阶段聚合。第三，触发条件比较苛刻，如果你的数据倾斜没有达到默认阈值，或者倾斜比例没那么夸张，AQE可能静悄悄不干活。所以理解原理还是有必要的，不能光把开关打开就以为完成任务。
4.3 广播变量与配置的正确姿势
除了AQE，另一个容易被忽略的优化是广播小表。小表join大表时，Spark默认会尝试Broadcast Hash Join，把小表广播到各executor，避免大表的shuffle。默认阈值spark.sql.autoBroadcastJoinThreshold是10MB，如果你的维表只有几十MB，直接调整阈值就可以：
spark.sql.autoBroadcastJoinThreshold=52428800也可以在SQL里显式加hint：
SELECT /*+ BROADCAST(d) */ f.order_id, d.sku_name FROM fact_orders f JOIN dim_sku d ON f.sku_id = d.sku_id关键禁忌是不要把阈值设得过大。你为了省一次shuffle把阈值调到1GB，小表广播到每个executor的内存开销可能直接压垮driver或executor，尤其在executor数量多的时候。广播的本质是用内存换网络IO，小表加广播只有在数据量可控时才是优解。
配置还有一个习惯问题：不要动不动改spark-defaults.conf里的全局配置。同一个集群跑不同类型任务，有的任务需要大shuffle分区，有的需要大广播阈值，全局改会影响彼此。我一般在代码里用SparkSession.builder.config()设置任务级参数，或者在spark-submit时通过--conf指定，这样更隔离，也不会坑到别人。
5. 方案选型：不同倾斜程度怎么组合落地
5.1 先量化倾斜程度
在动手优化之前，我强烈建议先量化一下key分布。不要靠猜，直接在数据上跑一条SQL就能看到热点key：
SELECT key, COUNT(*) AS cnt FROM fact_table GROUP BY key ORDER BY cnt DESC LIMIT 20;如果TOP1占比超过总数据量的10%，几乎可以认定存在倾斜。想更细一点，可以算一下各个key出现次数的分位数：
SELECT percentile_approx(cnt, 0.5) AS p50, percentile_approx(cnt, 0.95) AS p95, percentile_approx(cnt, 0.99) AS p99 FROM ( SELECT key, COUNT(*) AS cnt FROM fact_table GROUP BY key ) t;如果p99是p50的几十倍甚至上百倍，说明长尾非常严重。这个统计结果直接决定了用哪个方案：轻度倾斜可能只调并行度就行，中度倾斜用两阶段聚合或条件加盐，重度倾斜得把加盐、广播、过滤脏数据等组合起来。
5.2 技术方案对照表
我把常见的方案对照整理如下，方便你按场景对号入座：
| 场景 | 方案 | 优势 | 注意点 |
|---|---|---|---|
| groupBy/reduceByKey，倾斜比例中等 | 两阶段聚合 | 实现简单，对聚合类任务通用 | 中间结果多一次Shuffle，盐值要控制 |
| join大表和小表，热点key明确 | 加盐扩容 | 热点被拆散，并行度高 | 外连接和一对多场景要小心放大 |
| 小表join大表，小表真的很小 | 广播join | 完全避免大表shuffle | 广播阈值不能设太大，否则OOM |
| Spark 3+，join倾斜且满足阈值 | AQE SkewJoin | 框架自动处理 | 对聚合无效，需要开启adaptive |
| 倾斜key来自null或脏数据 | 数据清理/过滤 | 从源头消除热点 | 需要确认业务上能否过滤 |
| 单纯分区数不够 | 调大shuffle partitions | 最简单 | 热点key不拆开就没用 |
做选型时我会先看有没有脏数据导致的倾斜，再看是不是聚合场景，再看是不是join场景，最后看是否需要组合多个方案。这个顺序能省掉不少无用功。
5.3 组合使用的实际经验
我处理过一个订单分析任务，慢的原因都不止一层：先是一个join阶段被热点商品卡住，后面又叠加了一个group by导致的任务积压。当时没有只用一种方案，而是分了两步处理。第一步通过条件加盐把join拆开，第二步对聚合阶段用两阶段聚合，同时开启AQE兜底。最终运行时间从半小时压到了五分钟以内。
组合使用的关键点在于知道每个优化的作用边界。比如AQE开启后，它只会在join时自动拆分区，但对你的聚合阶段完全无能为力；广播只对“小表join大表”有效，如果两个都是大表，把广播阈值调大反而增加driver内存负担。把这些边界搞清楚，你才不会出现“开了AQE还是很慢”“调了并行度还是卡”的困惑。
另外，优化后一定要保留一个“中间结果落盘”的意识。比如加盐join结果可能后续还会被重复使用，用cache()或checkpoint把中间结果持久化，能避免后面每次跑都重新走一次耗时的shuffle。我当时就是先在加盐join后落了一份中间表，再做后续聚合，整个链路才稳定下来。
6. 高频问题排查实录
6.1 加盐或随机前缀后反而更慢
这是一个非常典型的现象：加了随机盐，任务不但没快，反而更慢。原因几乎都是数据膨胀失控。最常见的就是对所有key统一加盐，热点key只占全量的10%，加盐20倍后全量数据变成了原来的20倍，shuffle和磁盘压力陡增，得不偿失。
解决思路是只对热点key加盐。先跑一遍倾斜检测SQL，把热点key装进一个列表，再用条件分支给这些key单独加盐。如果你用的是两阶段聚合，同样只对TOP N热点key加盐，其他key的盐可以固定为0，这样整体数据量不会暴涨。
6.2 聚合或join结果行数不对
两阶段聚合结果不对，多半出在“解析原始key”这步。比如原始key本身包含下划线，regexp_extract的贪婪匹配很容易把a_b_1解析成a_b而不是a。为了规避，建议不要从带盐的key里解析原始key，而是在加盐时就把原始key单独保留一列，局部聚合时分组字段包含原始key列，第二阶段再按原始key列聚合。
join结果膨胀则往往是小表一对多数据没有先收敛。加盐前先对小表按key做一次去重或聚合，能有效避免复制后产生的多倍膨胀。如果业务上必须保留多行，那就要接受结果会放大，至少要提前估算好放大倍数。
6.3 AQE没有生效怎么办
看到AQE没生效，先确认三件事：第一，是不是Spark 3.0以上版本，只有3.0之后才有AQE，3.2之后默认开启。第二，是不是在SQL或DataFrame层面运行RDD的话不生效。第三，看执行计划里是否出现了AdaptiveSparkPlan字样。
还有一种情况是条件不满足。skewedPartitionFactor=5和skewedPartitionThresholdInBytes=256MB是同时生效的，如果你的总数据量本身不大，单个分区连256MB都不到，AQE自然不会触发。这时候就考虑手动加盐，或者降低两个阈值。
如果任务跑到一半抛出OOM，优先去看executor日志，确认是哪个Stage哪个task的内存溢出。不要一上来就调spark.memory.fraction，先判断是不是某个task读取了超大分区数据。如果是，按倾斜处理；如果不是，再考虑调整executor内存规格。
6.4 别忽略数据质量这个源头
有不少倾斜问题的根源不是业务热点，而是脏数据。比如null key、空字符串、非法日期值，它们会被HashPartitioner统一分配到一个分区，形成事实上的“热点”。这种问题用加盐方案也能缓解，但最合理的做法是在源头清理。
我的习惯是先把null和空字符串单独拉出来，统计一下占比。如果就是它们导致的倾斜，直接补上WHERE key IS NOT NULL AND key != ''，或者用一个业务上可以接受的默认值替代，比如__UNKNOWN__。这样既解决了倾斜，又提升了数据质量。从另外一个角度看，倾斜排查本身也是数据质量检查的一部分，很多业务上的统计异常，都是通过这类慢任务的排查被提前发现的。
最后再分享一个我自己的习惯：任务一慢，不要急着套方案。先去Spark UI把“最慢的几个task”和“Shuffle Read最大的task”截图，再跑一遍key分布统计，十分钟之内就能判断该走哪条路。数据倾斜本质上是数据分布的幂律现象，几乎无法彻底避免，但只要你掌握了定位方法和常用手段，就不会再被它搞得半夜爬起来调任务。希望这篇整理能帮你少踩几个坑，多在正常作息时间收工。
