Spark 批任务跑到 99% 卡住，剩下几条 task 遥遥无期——这基本是倾斜的经典画面。概念上都知道「key 不均匀」，落地时要分清：是 join 倾斜、聚合倾斜，还是 读入分区本身不均。处理手段不一样。
下面是一次订单明细关联店铺维表的案例。
案例背景
任务目标：T+1 明细表关联店铺维度，产出宽表。
SELECT /*+ BROADCAST(d) */ -- 起初以为维表不大，加了广播
o.*,
d.shop_name,
d.shop_level,
d.region
FROM dwd.order_detail o
LEFT JOIN dim.shop d
ON o.shop_id = d.shop_id
WHERE o.dt = '2026-03-25';
数据量：当日订单明细约 6.4 亿行；店铺维表约 120 万行。集群资源足够，平时类似任务 45～60 分钟。当天跑了 3 小时仍卡在最后一个 stage，Ganglia / Spark UI 显示 200 个 task 里 196 个已结束，4 个 task 跑了 2 小时+，shuffle read 是其它 task 的几十倍。
业务侧催数，临时把 executor 加了一倍，没本质改善——倾斜任务加资源往往只是陪跑。
定位：先看 UI，再验 key
1. Spark UI
进入卡住的 Stage：
- Tasks 按
Duration、Shuffle Read Size排序。 - 那 4 条 task 的 Shuffle Read 约 18～25 GB，其它多数在 300～800 MB。
- 输入行数也明显偏高。
说明倾斜发生在 shuffle 之后的计算，不是单纯读 HDFS 文件大小不均。
2. 抽样 key 分布
不要猜，直接查：
SELECT shop_id, COUNT(*) AS cnt
FROM dwd.order_detail
WHERE dt = '2026-03-25'
GROUP BY shop_id
ORDER BY cnt DESC
LIMIT 20;
结果触目惊心：
| shop_id | cnt |
|---|---|
| 0 | 约 1.1 亿 |
| 10086 | 约 3200 万 |
| 10010 | 约 2100 万 |
| … | 长尾很快下降 |
shop_id = 0 是历史上「未知店铺」的默认填充；大促活动订单大量落到 0。10086 / 10010 是头部店铺，本身也大，但和 0 比还是小一截。
维表里 shop_id=0 只有一行「未知」。广播 join 理论上该很轻，但当天广播并未生效：维表实际展开后超过 autoBroadcastJoinThreshold（中间还关联了其它属性视图，扫描后远超 10MB 阈值），优化器走了 SortMergeJoin，于是全量按 shop_id shuffle，0 这个 key 把个别 reducer 打死。
处理：分层拆解，而不是只加盐
方案 1：先把「默认值」拆出去（立刻止血）
-- 正常店铺
SELECT o.*, d.shop_name, d.shop_level, d.region
FROM dwd.order_detail o
LEFT JOIN dim.shop d
ON o.shop_id = d.shop_id
WHERE o.dt = '2026-03-25'
AND o.shop_id <> 0
AND o.shop_id IS NOT NULL
UNION ALL
-- 未知店铺：不参与大 join，直接打默认维度
SELECT o.*,
'UNKNOWN' AS shop_name,
'L0' AS shop_level,
'UNKNOWN' AS region
FROM dwd.order_detail o
WHERE o.dt = '2026-03-25'
AND (o.shop_id = 0 OR o.shop_id IS NULL);
0 和 NULL 从 shuffle join 里拿掉后，最重的那几条 task 消失，整体降到约 95 分钟。仍偏慢，因为头部店铺还在。
方案 2：对头部热点 key 加盐
对 top N 热点 shop_id 做随机盐，维表侧做膨胀：
-- 明细侧：热点加盐
SELECT o.*,
CASE WHEN o.shop_id IN (10086, 10010, ...)
THEN concat(o.shop_id, '_', cast(floor(rand()*16) AS int))
ELSE cast(o.shop_id AS string)
END AS shop_id_salt
FROM dwd.order_detail o
WHERE ...
-- 维表侧：热点复制 16 份
SELECT d.*,
concat(d.shop_id, '_', cast(n AS string)) AS shop_id_salt
FROM dim.shop d
LATERAL VIEW explode(sequence(0, 15)) t AS n
WHERE d.shop_id IN (10086, 10010, ...)
UNION ALL
SELECT d.*, cast(d.shop_id AS string) AS shop_id_salt
FROM dim.shop d
WHERE d.shop_id NOT IN (10086, 10010, ...);
再按 shop_id_salt join。热点被打散到 16 个 reducer。盐的份数按热点行数 / 目标每 task 行数估算，不必一次拉到 100。
方案 3：让广播真正生效
梳理维表视图，去掉不必要的宽字段和大扫描，让真正参与 join 的维度子集落到广播阈值内；或显式：
SELECT /*+ BROADCAST(d) */ ...
并确认事件日志里是 BroadcastHashJoin。广播成功后，小表倾斜敏感度会低很多（大表本地 join），但大表自身分区不均仍可能影响读入阶段——那是另一类问题。
方案 4：AQE 辅助
Spark 3 开 AQE 后，对部分倾斜 join 有优化：
spark.conf.set("spark.sql.adaptive.enabled", "true")
spark.conf.set("spark.sql.adaptive.skewJoin.enabled", "true")
AQE 能缓解一部分，但 默认值 key 占比 10%+ 这种极端情况，仍建议显式拆分，别完全赌优化器。
最终效果
组合采用：默认值拆分 + 头部加盐 + 维表精简后广播：
| 版本 | 耗时 | 说明 |
|---|---|---|
| 原始 SortMergeJoin | >3h（未跑完） | shop_id=0 致命 |
| 拆默认值 | ~95min | 头部店铺仍慢 |
| 再加盐 + 广播 | ~48min | 恢复到正常水位 |
另外补了质检：每日 shop_id top10 占比告警，超过阈值邮件通知，避免默认值再次冲量。
排查模板（可复用）
1. UI：卡住 stage 的 task 时长 / shuffle read 是否长尾
2. SQL：group by key order by count 看 topN 占比
3. 分类：
- 默认值 / null 过多 → 拆出去单独处理
- 少数业务热点 → 加盐或单独计算
- 小表误判未广播 → 打薄维表或强制 broadcast
- 聚合倾斜 → 两阶段聚合（先局部再全局）
4. 改完看 stage 的 task 分布是否收敛，不要只看总耗时
聚合倾斜可以怎么做
如果倾斜出在 groupBy / count distinct 而不是 join，常用两阶段：
-- 先局部打散再汇总（示意）
SELECT shop_id, SUM(partial_cnt) AS cnt
FROM (
SELECT shop_id, salt, COUNT(*) AS partial_cnt
FROM (
SELECT shop_id, cast(floor(rand()*8) AS int) AS salt
FROM dwd.order_detail
WHERE dt = '2026-03-25'
) t
GROUP BY shop_id, salt
) s
GROUP BY shop_id;
先把热点 key 拆到多个 partial，再二次聚合。count(distinct user_id) 这类更棘手，可以考虑近似算法（如 HLL）或按业务接受度做预聚合表，这和 join 加盐是不同路线。
小结
这次倾斜的核心不是「Spark 不行」，而是 脏默认值 + 头部店铺 + 广播没生效 叠在一起。加 executor 解决不了单一 key 的堆积。
数据倾斜的处理优先级，个人建议是：
- 先清默认值和脏 key
- 再决定广播 / 加盐 / 两阶段聚合
- 最后才考虑扩容
把 top key 监控做进日常，比每次任务卡死再连夜加盐要舒服得多。
