数据倾斜怎么解决？——加盐、AQE、两阶段聚合的场景差别

数仓开发手册第 6 题约 11 分钟读完

本文目录

先说结论
业务场景：我在某出海硬件品牌的测试设备倾斜
段落一：解法一——AQE（Adaptive Query Execution）
段落二：解法二——两阶段聚合（Group By 类倾斜）
段落三：解法三——加盐 Join（Join 类倾斜）
段落四：解法四——倾斜 key 单独处理（最暴力也最快）
段落五：四种方案对比
面试高频追问
实战踩坑：Null 值倾斜是最隐蔽的
总结

先说结论

数据倾斜的本质：某个或某几个 key 的数据量远大于其他 key，导致 Shuffle 之后某个 reducer 干了全部人的活，其他 reducer 已经闲着。表现是 99% 的 task 几分钟完成，1% 的 task 跑几十分钟。

四种标准解法的边界：AQE（自动、低成本、Spark 3.x 首选）、两阶段聚合（Group By 类倾斜首选）、加盐（Join 类倾斜首选）、倾斜 key 单独处理（只有几个已知大 key 时最快）。

先诊断再动手：不要一上来就加盐。先 COUNT 看分布、Spark UI 看 task 耗时分布，定位到具体是哪几个 key 在倾斜，再选方法。

先把这段讲顺；面试官追问时，再展开下面的原理和细节。

详细讲解

数据倾斜是数仓面试里被挖得最深的题之一。上一题讲 SQL 慢排查时，数据倾斜只是一条分支；这一题面试官会单独再挖一次，要求你讲出 3-4 种解决方案、各自的适用场景、以及你亲手处理过哪些倾斜。能把『AQE、加盐、两阶段聚合、倾斜 key 单独处理』这四种手段的边界讲清楚，这题基本稳了。

业务场景：我在某出海硬件品牌的测试设备倾斜

我在某出海硬件品牌做 Databricks 时有一条 Join 作业稳定跑 40 分钟，其中最后一个 task 占了 35 分钟——典型倾斜。诊断之后发现是一个内部测试设备的 device_id 占了事实表 30% 的数据，QA 团队每天拿它跑自动化测试，累积几年造成了极端倾斜。

-- Step 1: 诊断——查 key 分布
SELECT device_id, COUNT(*) AS cnt,
  ROUND(COUNT(*) * 100.0 / SUM(COUNT(*)) OVER (), 2) AS pct
FROM fact_device_event
WHERE dt = '2024-03-15'
GROUP BY device_id
ORDER BY cnt DESC LIMIT 10;
 
-- 结果：
-- TEST_DEVICE_XXX   3.2 亿    30.1%   ← 明显倾斜
-- AA11BB22CC33     150 万    0.14%
-- ...
 
-- Step 2: 解决——AQE 自动处理（首选）
SET spark.sql.adaptive.enabled = true;
SET spark.sql.adaptive.skewJoin.enabled = true;
SET spark.sql.adaptive.skewJoin.skewedPartitionFactor = 5;
SET spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes = 256MB;
 
-- 开启后 AQE 会自动把倾斜分区切成多个小分区并行处理
-- 效果：40 分钟 → 8 分钟

段落一：解法一——AQE（Adaptive Query Execution）

AQE 是 Spark 3.x 的核心特性，对于大多数场景的倾斜问题是『零代码改动』的第一选择。它的原理是：运行时根据 shuffle 之后的实际数据分布判断哪些分区倾斜了，把倾斜分区在逻辑上拆成多个小分区并行处理。

AQE 适用场景：Spark 3.0+ 环境、倾斜程度中等（倾斜分区是平均的 5-10 倍）、无需修改 SQL。

AQE 不适用场景：Hive 引擎（AQE 是 Spark 特性）、倾斜极端（比如 30% 数据集中在一个 key，AQE 拆分之后这个 key 的数据块本身还是巨大的）。

我的真实数字：某出海硬件品牌那次测试设备倾斜，开启 AQE 之后 40 分钟降到 8 分钟。但我知道 AQE 不是万能的——这个测试设备后来又长到占 45% 数据的时候，AQE 也搞不定了，改成了『倾斜 key 单独处理』。

段落二：解法二——两阶段聚合（Group By 类倾斜）

Group By 导致的倾斜（比如 count 某个热门分类），加盐解决不了——加了盐再 reduce 时还是得合并，只是延迟了问题。正解是两阶段聚合：先局部聚合（加盐），再全局聚合（去盐）。

-- ❌ 直接 GROUP BY 会倾斜
SELECT country, COUNT(*) FROM fact_event GROUP BY country;
-- 美国数据 60% 全进一个 reducer
 
-- ✅ 两阶段聚合
-- 阶段 1：加盐后局部聚合（每个 country + 随机后缀单独聚合）
WITH salted AS (
  SELECT country,
    CONCAT(country, '_', CAST(FLOOR(RAND() * 10) AS STRING)) AS salted_country
  FROM fact_event
),
partial AS (
  SELECT country, salted_country, COUNT(*) AS cnt
  FROM salted
  GROUP BY country, salted_country  -- 10 个 salt 分散压力
)
-- 阶段 2：去盐后全局聚合
SELECT country, SUM(cnt) AS total
FROM partial
GROUP BY country;

关键点：第一阶段 10 个 salt 就等于把每个倾斜 key 分到了 10 个 reducer。第二阶段数据量已经是第一阶段的 1/N，聚合就快了。

段落三：解法三——加盐 Join（Join 类倾斜）

Join 场景的倾斜加盐思路稍不一样：大表的倾斜 key 加随机前缀，小表的对应 key 扩容 N 倍（每个前缀都有一份）。

-- 场景：A 表是大事实表（某些 uid 是热门），B 表是小维度表
 
-- A 表：倾斜 uid 加随机前缀 0-9
WITH a_salted AS (
  SELECT uid, event_ts,
    CONCAT(CAST(FLOOR(RAND() * 10) AS STRING), '_', uid) AS join_key
  FROM fact_event
),
-- B 表：每个 uid 复制 10 份，配上 0-9 前缀
b_expanded AS (
  SELECT b.uid, b.country,
    CONCAT(CAST(salt AS STRING), '_', b.uid) AS join_key
  FROM dim_user b
  CROSS JOIN (SELECT explode(array(0,1,2,3,4,5,6,7,8,9)) AS salt) s
)
SELECT a.uid, a.event_ts, b.country
FROM a_salted a JOIN b_expanded b USING (join_key);

代价：B 表膨胀 N 倍。所以加盐更适合『B 是小表、A 是倾斜大表』的组合。如果 B 也是大表，膨胀之后存储和 IO 反而更差。

段落四：解法四——倾斜 key 单独处理（最暴力也最快）

如果倾斜 key 只有一两个、而且是已知的（比如那个测试设备 ID），最简单的做法是把这几个 key 单独 Join 一次，其他 key 正常 Join，最后 UNION 结果。

-- 先把大 key 单独拎出来
WITH hot AS (  -- 倾斜 key 单独处理
  SELECT /*+ BROADCAST(b) */ a.*, b.country
  FROM fact_event a
  JOIN dim_user b USING (uid)
  WHERE a.uid IN ('TEST_DEVICE_XXX', 'QA_BOT_001')
),
cold AS (  -- 非倾斜正常 Join
  SELECT a.*, b.country
  FROM fact_event a
  JOIN dim_user b USING (uid)
  WHERE a.uid NOT IN ('TEST_DEVICE_XXX', 'QA_BOT_001')
)
SELECT * FROM hot UNION ALL SELECT * FROM cold;

这个方案的精髓：倾斜 key 由于数据都是同一个 key，自然可以 Broadcast Join 秒出；非倾斜部分因为已经剔除了大 key，分布均匀不倾斜。我在某出海硬件品牌那个 45% 测试设备的场景用的就是这个。

段落五：四种方案对比

方案	适用场景	代码成本	性能提升
AQE	Spark 3.0+, 中等倾斜	零改动	3-5 倍
两阶段聚合	Group By 类倾斜	SQL 改动大	5-10 倍
加盐 Join	大表 Join 小表的倾斜	SQL 改动大	5-10 倍
倾斜 key 单独处理	已知且少量倾斜 key	SQL 改动中等	10 倍+

选型决策树：先看能不能开 AQE（零成本），不行再看是 Group By 还是 Join——Group By 两阶段聚合、Join 加盐或倾斜 key 单独处理。倾斜 key 少且已知就用单独处理（最快），多且不可枚举就用加盐。

面试高频追问

追问一：AQE 自动处理倾斜的底层是怎么实现的？

AQE 的倾斜处理依赖两个能力：（1）shuffle 写出时统计每个分区的字节数；（2）stage 之间暂停执行、等统计数据出来后重新决策。当 AQE 发现某个分区大于其他分区中位数的 N 倍（默认 5）且超过阈值（默认 256MB），就把这个分区在逻辑上切成多个小分区，对应 join 对侧的分区复制 N 份——思路和手工加盐一模一样，只是自动化了。

追问二：倾斜和小文件问题有什么区别？

正好相反。倾斜是『少数分区数据超大』，小文件是『大量分区数据超小』。倾斜看 task 的 max duration，小文件看 task 的数量和每个 task 的输入量。解决方法也不同——倾斜要拆分，小文件要合并。但两者都会被 AQE 处理——AQE 的『自动合并 shuffle 分区』就是专门治小文件的。

追问三：如果一个倾斜 Join 任务既不能改 SQL，也没法开 AQE，还有什么办法？

真实场景里有。比如是 Hive 引擎、或者 SQL 是产品封装生成的不能改。这时候只剩两招：（1）增加 reducer 数量（set mapred.reduce.tasks），分摊倾斜；（2）在上游就把倾斜 key 打散——比如在 DWD 写出时就给倾斜 key 加前缀存成 dwd_shard，下游直接用分片表。本质是把倾斜问题提前到写入时解决。

实战踩坑：Null 值倾斜是最隐蔽的

我在某出海硬件品牌接手过一个 Join 任务，倾斜调查了半天找不到 key。最后发现是 NULL——上游有 15% 的事件 user_id 是 NULL，Join 时所有 NULL 被归到同一个 reducer。这类倾斜特别隐蔽因为你 SELECT user_id, COUNT(*) ORDER BY cnt DESC 第一眼看不到，除非你把 IS NULL 单列出来。

解决办法：Join 之前把 NULL 过滤掉（如果业务允许），或者给 NULL 加随机后缀打散（如果 NULL 对应的数据要保留）。

-- ❌ 原始 Join（NULL 倾斜）
SELECT a.*, b.country
FROM fact a LEFT JOIN dim_user b ON a.uid = b.uid;
 
-- ✅ NULL 打散
SELECT a.*, b.country
FROM fact a LEFT JOIN dim_user b
  ON CASE WHEN a.uid IS NULL
          THEN CONCAT('null_', CAST(RAND() * 1000 AS INT))
          ELSE a.uid END = b.uid;
-- NULL 和 b.uid 永远 join 不上（预期），但至少打散到 1000 个 reducer

总结

数据倾斜这道题的上限是『能把四种方案的适用场景讲清楚』。能说出 AQE 是首选、Group By 用两阶段、Join 用加盐或单独处理，再结合一个真实案例（比如某测试设备 30% 数据 / 某 NULL 值 15% 数据），这题就到了资深水平。注意不要背配置值，要讲你踩过的坑和实际收益。

下一篇：『维度表设计——什么字段该冗余，什么字段该 JOIN 进来？』——进入建模深挖阶段，面试官接下来会连问 4 道建模题。

你遇到过最奇葩的倾斜 key 是什么？NULL、测试账号、默认值、还是别的？

关注「Wood聊大数据」| 9年大数据老兵，聊技术、聊职场、聊成长

刷题进度保存在当前浏览器