Spark Join 数据倾斜怎么治

难度与前置

难度：进阶
前置：会提交 Spark 作业；知道 Shuffle Join 与 Broadcast Join 的区别；能打开 Spark UI。
读完能做什么：用 UI 与 SQL 定位倾斜 Key，按场景选加盐 / 广播 / 拆分 / AQE，并用 Task 耗时与结果集验收。

适用场景

当你的批作业出现：

某个 Stage 里少数 Task 跑几十分钟，其余几秒结束
Join / 聚合后墙钟被长尾 Task 拖死
调大 spark.sql.shuffle.partitions 后长尾依旧（只是更碎）

不适用：

整 Stage 均匀慢（更可能是扫盘过大、UDF、GC，不是倾斜）
小表本可 Broadcast，却被误判成「倾斜」去加盐（先试广播）
业务上必须严格按某热点 Key 有序单分区处理（加盐会打乱分区语义，需另设计）

真实场景（一条）

投放 DWD：消耗明细 × 渠道维表，按 channel_id Join。某渠道贡献全库约 40% 明细，Sort-Merge Join 后一个 Task 处理该 Key，Stage 从 8 分钟拖到 55 分钟。

处理：先确认维表可 Broadcast（几十 MB）；开 spark.sql.autoBroadcastJoinThreshold 并 broadcast(dim)。墙钟回到约 9 分钟，且金额汇总与改前一致。

若维表也很大、不能广播，则对热点 channel_id 加盐扩成多分区再 Join，维表侧复制盐值，最后去盐聚合——下面按步骤写清。

怎么做（步骤清单）

1. 先确认是不是倾斜（看 UI，别猜）

信号	看哪里	判断
Task 耗时长尾	Stage → Tasks：max ≫ median	典型倾斜
Shuffle Read 不均	某 Task Shuffle Read 远大于同伴	热点 Key 灌进该分区
Spill / GC 集中	同一批长尾 Task	热点带来内存压力
全员均匀偏慢	各 Task 接近	不是倾斜，换方向

经验阈值：max(task_duration) / median > 5 且 Shuffle Read 同期偏斜，优先按倾斜治。

2. 定位热点 Key

对 Join 键做分布摸底（采样或全量，按数据量选）：

sql

-- 摸底：Top Key 占比（Spark SQL）
SELECT channel_id, cnt, cnt * 1.0 / SUM(cnt) OVER () AS ratio
FROM (
  SELECT channel_id, COUNT(*) AS cnt
  FROM dwd_ad_spend_detail_di
  WHERE dt = '${biz_date}'
  GROUP BY channel_id
) t
ORDER BY cnt DESC
LIMIT 20;

记下：

Top1 / Top5 占比
Join 两侧各自键基数与表体积
是否空值 / 默认值（-1、unknown）扎堆

3. 按场景选策略（决策树）

条件	优先策略
一侧 ≤ 广播阈值（常见数百 MB 内，视集群）	Broadcast Join
仅少量热点 Key，其余均匀	热点拆分：热点单独处理 + 非热点正常 Join，再 UNION
多 Key 倾斜、两侧都大	加盐（salt）扩分区再 Join
先聚合再 Join 且聚合可下推	两阶段聚合（partial + final）
Spark 3.x 且开启 AQE	打开倾斜 Join 优化作兜底，不替代上面主动治理

4. Broadcast（能广播就别加盐）

scala

import org.apache.spark.sql.functions.broadcast

val fact = spark.table("dwd_ad_spend_detail_di").where($"dt" === bizDate)
val dim  = spark.table("dim_ad_channel_df").where($"dt" === bizDate)

val joined = fact.join(broadcast(dim), Seq("channel_id"), "left")

配置参考：

text

spark.sql.autoBroadcastJoinThreshold = 104857600   # 100MB，按集群调

TIP

Broadcast 前确认维表无重复主键，否则事实表会膨胀。先 SELECT channel_id, COUNT(*) ... HAVING COUNT(*) > 1。

5. 加盐 Join（两侧都大时）

思路：事实表热点 Key 扩成 key + salt；维表对同一 Key 复制 0..N-1 盐值；Join 后再按业务键聚合。

sql

-- 事实侧：随机盐 0~15
SELECT
  channel_id,
  CAST(FLOOR(RAND() * 16) AS INT) AS salt,
  bill_id,
  cost_amount
FROM dwd_ad_spend_detail_di
WHERE dt = '${biz_date}';

-- 维表侧：每个 channel_id 复制 16 行
SELECT d.*, s.salt
FROM dim_ad_channel_df d
LATERAL VIEW EXPLODE(SEQUENCE(0, 15)) s AS salt
WHERE d.dt = '${biz_date}';

Join 条件：fact.channel_id = dim.channel_id AND fact.salt = dim.salt。

盐值个数经验：让热点分区体量接近「均匀分区中位数」；常见 8～64，过大则维表复制成本爆炸。

6. 空值与默认值单独处理

NULL / '' / 'unknown' 常被 Hash 到同一分区。做法：

Join 前把空键映射到分散伪键，或过滤后走独立支路
业务上「无渠道」是否允许进指标，写进口径文档

7. AQE 作兜底（Spark 3+）

text

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

AQE 能拆倾斜分区，但不会替你修错误的 Join 键或维表重复。先把模型做对，再开 AQE。

常见坑（至少牢记这些）

把「整 Stage 慢」当成倾斜
全员 Task 都 20 分钟，加盐无效。先看扫字节、UDF、写入小文件。
维表有重复键还 Broadcast
事实行被成倍放大，金额「对上了 UI 变快」但数错了。广播前验唯一性。
盐值过大或只给一侧加盐
维表复制成百倍，或 Join 对不上变成丢数。两侧盐策略必须对称。
盲目加大 spark.sql.shuffle.partitions
倾斜 Key 仍进同一 Hash 分区，分区数再大也救不了该 Key。先打散 Key，再调分区数。
忽略空值扎堆
Top Key 查出来是 NULL，加盐却只处理业务渠道 ID，长尾依旧。

如何验收（可量化）

[ ] Spark UI：问题 Stage 的 max(task_time) / median 降到可接受（目标 < 3，或业务约定阈值）
[ ] Shuffle Read 最大 Task 不再独占数量级差距
[ ] 结果金额 / 行数与改前基准日对齐（diff 可解释，理想为 0）
[ ] 维表主键唯一检查通过；Broadcast 计划在 EXPLAIN 中可见（若走广播）
[ ] 热点 Top Key 占比有文档记录，便于下次回归

sql

-- 改前后同日金额对齐
SELECT
  'before_or_after' AS tag,
  SUM(cost_amount) AS cost_sum,
  COUNT(*) AS row_cnt
FROM ads_or_dwd_join_result
WHERE dt = '${biz_date}';

bash

# 从 Spark History / UI 导出 Stage 任务耗时后，粗算长尾比
# max_ms / median_ms 应明显下降

Spark Join 数据倾斜怎么治 ​

适用场景 ​

真实场景（一条） ​

怎么做（步骤清单） ​

1. 先确认是不是倾斜（看 UI，别猜） ​

2. 定位热点 Key ​

3. 按场景选策略（决策树） ​

4. Broadcast（能广播就别加盐） ​

5. 加盐 Join（两侧都大时） ​

6. 空值与默认值单独处理 ​

7. AQE 作兜底（Spark 3+） ​

常见坑（至少牢记这些） ​

如何验收（可量化） ​

相关面试题 ​