大数据：Spark 数据倾斜怎么定位与解决？热点 Key 采样、加盐、两阶段聚合实战

一、问题的本质：为什么倾斜总是发生在 Shuffle 之后

先把概念钉死。Spark 的“数据倾斜”并不是指数据总量大，而是指同一个 Stage 内不同 Task 处理的数据量或计算量出现数量级差异，导致整个 Stage 的完成时间由最慢的那个 Task 决定。由于 Spark 采用“全 Stage 同步屏障”的调度模型，一个 Stage 里只要有一个 Task 拖尾，其余空闲的 Executor 资源就只能干等。这就是为什么倾斜的代价远大于“数据多”——数据多是线性变慢，倾斜是长尾放大。

倾斜几乎总与 Shuffle 绑定。原因在于：Shuffle 之前，数据按 HDFS Block 或上游分区均匀切分，各 Task 处理量近似均衡；一旦发生 Shuffle，分区函数（默认是 HashPartitioner，即 key.hashCode % numPartitions）会把相同 Key 的记录路由到同一分区。如果某个 Key 的记录数占全表 30%，那么它所在的分区就会承担 30% 的数据量，而其余分区平摊剩余 70%。分区数与数据分布一旦“错配”，倾斜就发生了。

这里有一个常被忽略的细节：倾斜是“相对”而非“绝对”的。同样是 1 亿条记录里有一个 500 万条的 Key，在 200 个分区下它占 2.5 个分区的平均量，尚可忍受；在 2000 个分区下它占 25 个分区的平均量，就会成为明显的长尾。所以讨论倾斜必须同时给出三个量：数据分布、分区数、单 Task 处理能力。脱离分区数谈倾斜是没有意义的。

本文评述：很多团队把“调大 shuffle.partitions”当作万能药，这其实是一种“用空间换长尾”的粗放做法。分区数增大确实能稀释单个热点 Key 的占比，但同时会增加小文件、调度开销与 GC 压力。真正的问题是分区函数没有感知数据分布，单纯增加分区只是把“一个大长尾”切成“若干中长尾”，治标不治本。

从执行计划角度看，倾斜可能出现在三类算子之后：groupByKey / reduceByKey / aggregateByKey 等聚合类 Shuffle、join 类 Shuffle（尤其是 SortMergeJoin 的 shuffle 阶段）、以及 repartition / partitionBy 显式重分区。三类场景的治理手法并不相同，后文会分别展开。

二、定位：从 Stage 到 Task 的可观测体系

2.1 第一层：Spark UI 的“三看”

定位倾斜的第一步不是写代码，而是看 Spark UI 的 Stage 详情页。笔者总结为“三看”：看 Duration 分布、看 Shuffle Read Size 分布、看 Spill 情况。在 Stage 的 Summary Metrics 表格里，如果 Max 列的 Shuffle Read Size 是 Median 的 10 倍以上，基本可以判定存在倾斜；如果 Max 的 Spill (Memory) 显著大于 0 而 Median 为 0，说明热点分区已经大到内存放不下、被迫落盘，性能会进一步恶化。

需要提醒的是，Spark UI 的 Summary Metrics 只展示分位数（25/50/75/100 分位），对于“只有极少数 Task 异常”的场景，75 分位可能完全正常，必须直接看 Max 与 Median 的比值。笔者在实践中常用一个经验阈值：Max/Median > 5 且 Max 绝对值超过 128MB，就值得深入排查。

2.2 第二层：事件日志与指标采集

Spark UI 是事后查看，生产环境更需要自动化采集。Spark 的 EventLog（配置 spark.eventLog.enabled=true）会记录每个 Task 的 shuffleReadMetrics、shuffleWriteMetrics 与 inputMetrics。基于 EventLog 可以构建倾斜检测规则，例如：

# 伪代码：基于 EventLog 的倾斜检测
for stage in event_log.stages:
    reads = [t.shuffleReadBytes for t in stage.tasks]
    if len(reads) < 10: continue
    median = percentile(reads, 50)
    maxv   = max(reads)
    if maxv > 5 * median and maxv > 128 * 1024 * 1024:
        alert(stage.id, ratio=maxv/median, max_bytes=maxv)

这套规则可以接入 Prometheus + Grafana，形成倾斜看板。笔者所在团队曾用类似规则把“事后发现”提前到“任务运行中告警”，平均故障定位时间从小时级降到分钟级。需要说明的是，上述阈值是工程经验值，不同集群、不同业务的数据分布差异很大，建议先用历史任务跑一遍分布统计，再确定阈值，不要直接照搬。

2.3 第三层：Task 级别的“慢任务画像”

知道哪个 Stage 倾斜之后，还要知道是哪个 Key 导致的。这一步需要把 Task 与分区、分区与 Key 对应起来。最直接的办法是在 Shuffle 前对目标列做一次采样统计（见第三节），另一种办法是解析 Shuffle 数据文件的分区索引，统计每个分区的记录数。后者对生产环境更友好，因为它不需要改动业务逻辑，但实现成本较高，通常借助 Spark 的 ShuffleBlockResolver 或直接读取 .data 文件头来完成。

三、热点 Key 采样：把“猜”变成“测”

3.1 为什么要采样，而不是全量统计

全量统计每个 Key 的频次当然最准，但代价是一次额外的全表 Shuffle，对于 TB 级表来说，这个代价可能比倾斜本身还大。采样的价值在于：用 1%～5% 的样本，以可控的误差估计出热点 Key 的排序。我们关心的不是“Key A 到底有 512 万条还是 508 万条”，而是“Key A 是不是 Top 1、量级是不是百万级”，这个精度采样完全能给到。

3.2 三种采样姿势与适用场景

笔者更推荐“随机行采样 + 频次 TopN”的组合：先 sample(0.02) 抽 2%，再 groupBy(key).count().orderBy(desc).limit(50)，得到的 Top50 列表基本能覆盖 95% 以上的倾斜风险。采样率的选择有一个经验公式：若希望以 95% 置信度捕获占比 p 的 Key，样本量 n 需满足 n·p ≥ 3。例如热点 Key 占比 0.5%，则样本量至少 600 条，对应到 1 亿行表就是 0.0006% 的采样率，实际取 1% 已非常充裕。

3.3 采样结果的解读：不要只看 Top1

采样之后，真正有用的信息是“分布形态”而非单个 Key。笔者习惯画一张“频次—排名”的双对数图：如果曲线在头部急剧下降、尾部平缓，说明是“少数超级热点”，适合加盐或两阶段聚合；如果曲线整体平缓、没有明显头部，说明是“长尾均匀但总量大”，此时加盐收益有限，更应考虑增大并行度或改用更合适的 Join 策略。这一步的判读，直接决定后续选哪套治理方案。

笔者认为：采样的真正价值不是“找出那个 Key”，而是“判断倾斜的类型”。超级热点、长尾均匀、双峰分布，对应的治理手段完全不同。很多团队一上来就加盐，结果对长尾均匀型毫无效果，反而增加了 CPU 开销，根源就在于跳过了分布形态判读这一步。

四、加盐打散：原理、边界与踩坑清单

4.1 加盐的本质是“重写分区函数”

加盐（Salting）的做法是给热点 Key 拼接一个随机后缀，例如把 key 变成 key_0 … key_9，从而让原本落到同一分区的数据分散到 10 个分区。从分区函数的角度看，这等价于把 hash(key) % N 替换为 hash(key + salt) % N，人为引入熵，打破“同 Key 必同分区”的约束。

但加盐不是免费的午餐。它把一次聚合拆成了“局部聚合 + 全局聚合”两步，第二步仍然要按原始 Key 聚合，如果第二步不做二次打散，热点会原样复现。所以加盐必须与两阶段聚合配合使用，单独加盐只对“不需要按 Key 聚合、只需打散计算”的场景有效（如某些 map-only 的倾斜）。

4.2 盐粒数的选择：一个可计算的边界

盐粒数 N_salt 太小，打散不充分；太大，则第二阶段的小分组过多，调度开销上升。设热点 Key 记录数为 H，分区数为 P，平均分区数据量为 A = Total/P，则加盐后该 Key 每个子分区的数据量约为 H/N_salt。理想情况下希望 H/N_salt ≈ A，即 N_salt ≈ H/A = H·P/Total。举例：Total=1 亿，P=200，H=2000 万，则 N_salt ≈ 40。实践中取 2 的幂（如 32 或 64）更利于哈希分布均匀。

需要强调的是，这个公式假设热点 Key 的记录在加盐后均匀分布到各子分区，而实际哈希可能仍有偏斜。因此建议在公式结果基础上上浮 20%～50%，并在压测中观察第二阶段的分组数是否超过分区数的 3 倍——超过则说明盐粒过细，需要回调。

4.3 踩坑清单

• 只对热点 Key 加盐：非热点 Key 不加盐，否则第二阶段分组爆炸。实现上通常用广播变量下发热点 Key 集合，if (hotKeys.contains(k)) k + "_" + rand else k。
• 盐值随机源不可复现：用 rand() 会导致每次运行结果不同，不利于排查。建议用 pmod(hash(key), N_salt) 这种确定性函数。
• 忘记剥离盐值：第二阶段聚合前必须把后缀去掉，否则聚合结果按 key_0、key_1 分开，结果错误。
• UDF 中加盐导致序列化开销：尽量用内置函数实现，避免每行都走 Scala UDF 的序列化路径。

五、两阶段聚合：正确性证明与工程实现

5.1 为什么两阶段聚合是正确的

两阶段聚合（Two-stage Aggregation）的核心思想是：把 groupBy(key).agg(f) 拆成 groupBy(key, salt).agg(f) 与 groupBy(key).agg(f') 两步。其正确性依赖于聚合函数 f 的结合律与交换律：只要 f 满足 f(f(a,b),c) = f(a,f(b,c)) 且 f(a,b)=f(b,a)，分步聚合与一次性聚合结果等价。

sum、count、max、min 都满足；但 avg 不满足——平均值不能直接对子平均值再求平均，必须先转成 (sum, count) 再在第二阶段做除法。同理，去重计数（count distinct）也不能简单两阶段，需要借助 HyperLogLog 或精确的 bitmap。这是工程中最容易出错的地方，笔者见过不止一次“两阶段 avg 算错”的事故。

5.2 完整代码实现（PySpark）

from pyspark.sql import functions as F

SALT = 32
# 1) 采样得到热点 Key（假设已得到 hot_keys 列表）
hot = spark.sparkContext.broadcast(set(hot_keys))

# 2) 第一阶段：热点 Key 加盐，局部聚合
def add_salt(k):
    if k in hot.value:
        return k + "_" + str(abs(hash(k)) % SALT)
    return k

salt_udf = F.udf(add_salt, "string")
stage1 = (df
    .withColumn("salted_key", salt_udf(F.col("key")))
    .groupBy("salted_key")
    .agg(F.sum("amount").alias("partial_sum"),
         F.count("*").alias("partial_cnt")))

# 3) 第二阶段：剥离盐值，全局聚合
stage2 = (stage1
    .withColumn("key", F.split("salted_key", "_")[0])
    .groupBy("key")
    .agg(F.sum("partial_sum").alias("total_amount"),
         F.sum("partial_cnt").alias("total_cnt")))

这段代码有两个工程细节值得注意。第一，split("_")[0] 在 Key 本身含下划线时会出错，生产环境应改用不会冲突的分隔符（如 \u0001）或结构化方式拼接。第二，广播热点 Key 集合的大小要控制，若热点 Key 有几十万个，广播变量会撑爆 Executor 内存，此时应改为“按频次阈值动态判断”而非精确集合。

5.3 性能对比（模拟数据）

注：上表为笔者在 20 节点、每节点 16C64G 测试集群上的模拟压测数据，数据规模 1 亿行、200 分区、热点 Key 占比 20%，仅用于说明量级关系，不代表任何生产环境的绝对性能。

六、AQE 与自适应治理：新版本带来的范式变化

从 Spark 3.0 开始，自适应查询执行（Adaptive Query Execution，AQE）逐步成熟，3.2 之后默认开启。AQE 的核心机制是：在 Shuffle 写阶段收集每个分区的统计信息（大小、记录数），在 Shuffle 读之前根据这些统计动态合并小分区、拆分大分区、甚至切换 Join 策略。这实际上把“运行前静态决定分区”变成了“运行中动态调整分区”。

与倾斜治理直接相关的是 spark.sql.adaptive.skewJoin.enabled 与 spark.sql.adaptive.coalescePartitions.enabled。前者会在检测到 Join 两侧分区大小差异超过阈值（默认 5 倍且大于 256MB）时，把大分区再拆成多个子分区分别与小表分区做 Join，本质上是引擎自动化的“加盐 + 两阶段”。

本文评述：AQE 的出现并没有让手写加盐“过时”，而是改变了它的适用边界。AQE 擅长处理“运行时才暴露的、量级中等的倾斜”，但对“极端超级热点”（单分区数 GB 以上）仍可能力不从心，且其拆分逻辑对聚合类 Shuffle 的支持弱于 Join。笔者的经验是：Join 倾斜优先交给 AQE，聚合倾斜仍以手写两阶段为主，两者叠加使用而非二选一。

需要提醒的是，AQE 的统计信息来自 Shuffle 写阶段的 MapStatus，若上游分区本身极度不均，AQE 也只能“在已知的不均上做二次调整”，无法凭空创造均衡。因此 AQE 不能替代采样与分布分析，它只是把治理动作从“开发期”推迟到了“运行期”。

七、Join 倾斜、窗口倾斜与写倾斜的专门处理

7.1 Join 倾斜：Map Join 与拆分大表

Join 倾斜的经典解法是“小表广播”。当一侧表小于 spark.sql.autoBroadcastJoinThreshold（默认 10MB）时，Spark 会自动转成 BroadcastHashJoin，彻底避免 Shuffle。但生产中常见“两侧都是大表、只有少数 Key 热点”的情况，此时可把大表按热点 Key 与非热点 Key 拆成两部分：热点部分与小表对应部分做广播 Join，非热点部分正常 Shuffle Join，最后 union。这种“拆分 + 分流”的思路在笔者看来比单纯加盐更稳，因为它把倾斜隔离在可控的小范围内。

7.2 窗口倾斜：partitionBy 的陷阱

窗口函数（Window.partitionBy(...).orderBy(...)）会把同一分区的数据全部拉到一个 Task 内排序。如果 partitionBy 的列基数很低（比如按“省份”分区，而某省数据占 40%），倾斜会非常严重。治理手段有限：要么在 partitionBy 中增加一个高基数列做二级分区（会改变语义，需谨慎），要么预先过滤/拆分热点分区单独计算。笔者认为，窗口倾斜的根因往往是“业务上就不该按这个维度做全局窗口”，应优先从需求侧审视。

7.3 写倾斜：小文件与分区键选择

写倾斜常被忽视：partitionBy("dt","hour") 写 Hive 表时，若某小时数据量是其他的 20 倍，会产生一个超大文件加一堆小文件。治理思路是控制单文件大小（spark.sql.files.maxRecordsPerFile）并配合 repartition 按数据量而非行数重分区。这部分与聚合倾斜的技术栈不同，但同属“分区函数与分布不匹配”这一主线。

八、闭环实战：一个电商大促场景的完整治理记录

下面用一个脱敏后的真实场景串起全文方法。某电商大促期间，订单宽表按 user_id 聚合 GMV 的任务从日常 15 分钟劣化到 2 小时 40 分。按本文闭环执行：

1. 定位：Stage 详情显示 Shuffle Read Size 的 Max 为 6.2GB、Median 为 210MB，比值约 29，确认倾斜。
2. 采样：对 user_id 做 2% 采样，Top1 是一个“平台补贴虚拟账号”，占样本 18%，属于超级热点。
3. 诊断：该虚拟账号承载了所有补贴订单，业务上确实需要聚合，不能过滤。
4. 治理：采用两阶段聚合 + 加盐，盐粒数按公式估算约 30，取 32；同时开启 AQE 的 skewJoin 作为兜底。
5. 验证：最长 Task 从 2h40m 降到 11 分钟，总 Shuffle 量增加约 8%，整体任务耗时降到 18 分钟，满足大促 SLA。

这个案例的启示是：倾斜治理的收益主要来自“消除长尾”，而不是“减少总量”。总 Shuffle 量反而略有上升，但任务完成时间由最长 Task 决定，所以整体大幅改善。这也解释了为什么“只调大分区数”往往效果有限——它没有改变长尾的绝对大小。

九、前沿观察与笔者思辨

近年学术界对“倾斜感知的 Shuffle”有不少探索，主流方向包括：基于代价模型的自适应分区（如根据历史统计预分配分区权重）、基于学习的方法预测热点 Key、以及在存储层做“分区内二次分片”。工业界方面，Spark 社区在 AQE 上持续投入，Databricks 的 Photon 引擎也在 Join 重分布上做了向量化优化。笔者认为，这些工作的共同趋势是把“分区”从一个静态配置项变成一个由统计信息驱动的动态决策，这与本文主线完全一致。

但也要清醒看到局限：任何自适应机制都依赖准确的运行时统计，而统计本身有采集开销与滞后性；对“流式写入 + 即时查询”的场景，统计往往来不及反映最新分布。因此笔者认为，未来三到五年内，“手写的、基于业务语义的倾斜治理”仍不会被完全取代，尤其是在金融、电商这类对正确性与时效性要求极高的场景。工程团队应把 AQE 当作“默认底座”，把手写方案当作“精确制导”，两者结合才是务实路径。

笔者认为：数据倾斜治理的成熟标志，不是“掌握了几种加盐技巧”，而是“建立了一套能从指标发现倾斜、用采样判断类型、按类型选择方案、用压测验证收益的闭环”。技巧会过时，闭环不会。

十、参考资料与拓展阅读