两张大表JOIN，某个key有500万条——JOIN倾斜的三种解法怎么选？

-- 典型的JOIN倾斜场景
SELECT o.order_id, o.amount, u.user_name
FROM orders o
JOIN users u ON o.user_id = u.user_id;
 
-- 如果某个user_id（比如平台账号或测试账号）有500万条订单
-- 那对应的Task要处理：500万条orders + users表中这1行 = 数据量极大
 
-- 先排查哪些key是倾斜的
SELECT
  user_id,
  COUNT(*) AS cnt,
  COUNT(*) * 100.0 / SUM(COUNT(*)) OVER() AS pct
FROM orders
GROUP BY user_id
ORDER BY cnt DESC
LIMIT 10;

-- Spark SQL里强制Broadcast Join
SELECT /*+ BROADCAST(u) */
  o.order_id, o.amount, u.user_name
FROM orders o
JOIN users u ON o.user_id = u.user_id;
 
-- 或者设置自动广播阈值
SET spark.sql.autoBroadcastJoinThreshold = 209715200;  -- 200MB
-- 小于200MB的表会被自动广播
 
-- 我在出海硬件公司做调优时，把这个阈值从默认10MB调到200MB
-- 有多个维度表（设备型号表、国家码表等）自动变成Broadcast Join
-- 这些作业平均执行时间降低了40%

-- 原始：orders表某些user_id极度倾斜
-- 方案：大表加随机盐，小表按盐扩展
 
WITH salted_orders AS (
  -- 大表orders：每行user_id加随机前缀0-9，打散成10份
  SELECT
    CONCAT(CAST(FLOOR(RAND() * 10) AS STRING), '_', user_id) AS salted_user_id,
    order_id,
    amount
  FROM orders
),
expanded_users AS (
  -- 小表users：每行扩展10倍，对应0-9每个前缀都有一行
  SELECT
    CONCAT(CAST(salt.n AS STRING), '_', user_id) AS salted_user_id,
    user_name
  FROM users
  CROSS JOIN (
    SELECT explode(array(0,1,2,3,4,5,6,7,8,9)) AS n
  ) salt
)
SELECT o.order_id, o.amount, u.user_name
FROM salted_orders o
JOIN expanded_users u ON o.salted_user_id = u.salted_user_id;

代价与限制：小表被扩展了N倍（上例是10倍）。如果users表有1000万行，扩展后就是1亿行，内存压力很大。所以这个方案适合「大表 JOIN 中等大小表」，不适合「大表 JOIN 大表」。如果两张都是超大表，扩展后内存放不下。

方案三：倾斜key单独处理（最稳但最复杂）

把倾斜的key先识别出来，拆成两条路：倾斜key单独用一个方案（比如Broadcast），剩余正常key走标准Shuffle Join，最后UNION。这个方案对数据准确性最有保障，但代码复杂度高。

-- 前提：已知倾斜的user_id列表（通过上面的排查SQL得到）
-- 假设倾斜的user_id有：'user001', 'user002', 'user003'
 
-- 路径A：倾斜key，用Broadcast Join处理
SELECT /*+ BROADCAST(u) */
  o.order_id, o.amount, u.user_name
FROM (
  SELECT * FROM orders WHERE user_id IN ('user001','user002','user003')
) o
JOIN users u ON o.user_id = u.user_id
 
UNION ALL
 
-- 路径B：非倾斜key，走正常Shuffle Join
SELECT o.order_id, o.amount, u.user_name
FROM (
  SELECT * FROM orders WHERE user_id NOT IN ('user001','user002','user003')
) o
JOIN users u ON o.user_id = u.user_id;
 
-- 倾斜key少时，路径A的数据量不大，Broadcast完全没压力

优点：对大表没有额外的内存压力，不需要扩展任何表。

缺点：需要提前知道哪些key是倾斜的，适合倾斜key比较固定（如测试账号、系统账号）的场景。如果倾斜key是动态变化的，就不适合。

三种方案的选择框架

方案	前提条件	内存代价	代码复杂度	适合场景
Broadcast Join	一张表能放内存（<200MB通常）	小（每Executor一份）	低	大表JOIN小维度表
加盐打散(Salting)	两张都是大表，但其中一张可以扩展N倍	中（小表扩N倍）	中	大表JOIN中等表，倾斜key多
倾斜key单独处理	倾斜key已知且固定	低	高	倾斜key少且固定（如系统账号）
AQE自动处理	Spark 3.0+，开启AQE	无额外	低（自动）	作为兜底，配合以上方案使用

实际生产中，我的优先级是：Broadcast Join（优先，最简单）→ 倾斜key单独处理（倾斜key固定时）→ 加盐打散（前两个不行再用）→ AQE兜底。Spark 3.0之后一定要开AQE，很多轻度倾斜它能自动处理。

面试高频追问

追问一：加盐打散后，如果我要LEFT JOIN，结果有没有问题？

有问题，需要特殊处理。LEFT JOIN要保留左表的所有行，即使右表没有对应数据。加盐后，左表的一行user_id被打散成10行（salted_user_id），每行都要跟扩展后的右表JOIN。如果某个user_id在右表不存在，正确结果是返回NULL；但扩展后10行都JOIN不上，结果是10行NULL，重复了10倍。解决方法：加盐前先判断，或者JOIN后做一次按原始key的去重。实际上LEFT JOIN场景加盐比较麻烦，优先考虑其他方案。

追问二：Spark AQE里的Skew Join是怎么工作的？

Spark 3.0的AQE Skew Join会在Shuffle之后、Join执行之前，检测各分区的数据量。如果某个分区数据量远超中位数（默认是中位数的5倍且大于256MB），AQE会自动把这个大分区拆成若干小分区，分给多个Task并行处理，相应地把Join另一侧的数据也复制多份。这和加盐方案的思路类似，但是自动完成的，不需要修改SQL。开启方式：spark.sql.adaptive.skewJoin.enabled=true（AQE开启后默认也开启）。

追问三：你们生产环境Broadcast Join的阈值设多少合适？

这取决于Executor的内存配置。经验公式：Broadcast阈值 ≤ Executor内存 × 0.1 ÷ Executor核数。比如Executor内存16GB，每个Executor跑4个Task，那每个Task能用的内存约是16GB × 0.6（Spark用60%做执行内存）÷ 4 ≈ 2.4GB。广播表每个Executor一份，所以阈值可以设到Executor内存的5-10%比较安全，即约800MB-1.6GB。我在出海硬件公司设的是200MB，因为当时Executor内存是8GB，数量50个，200MB × 50 = 10GB，在集群内存预算内。

实战踩坑

我在某元宇宙社交应用做实时数仓时，有个Flink作业要把用户行为流（Kafka消费）和用户画像表（MySQL同步过来的维度数据）做关联，给行为数据打标签。上线后发现有个VIP用户账号产生了大量行为数据，对应的Flink SubTask持续高负载，背压严重，checkpoint开始超时。

-- Flink里的倾斜：keyBy(user_id)后，某个user_id数据量暴增
-- 这是Flink SQL的做法，和Spark思路类似但有些差异
 
-- 问题：直接按user_id keyBy做维度关联
-- SELECT e.*, u.user_level, u.vip_type
-- FROM user_events e
-- JOIN user_dim FOR SYSTEM_TIME AS OF e.proctime AS u
-- ON e.user_id = u.user_id;
 
-- 解决：对该VIP账号做特殊过滤，走独立链路
-- 正常用户走维表JOIN，VIP用户直接打固定标签（因为VIP标签固定）
-- 在Flink SQL里用CASE WHEN实现分流：
SELECT
  e.event_id,
  e.user_id,
  e.event_type,
  CASE
    WHEN e.user_id IN ('vip_account_001') THEN 'VIP'  -- 固定标签
    ELSE u.user_level  -- 正常走维表查询
  END AS user_level
FROM user_events e
LEFT JOIN user_dim FOR SYSTEM_TIME AS OF e.proctime AS u
  ON e.user_id = u.user_id;

教训：Flink里的倾斜比Spark更危险，因为是实时流，倾斜一旦出现会持续累积，最终导致整个作业崩溃。上线前一定要对数据的key分布做压测，特别是要模拟极端场景（超大用户行为量）。

总结

JOIN倾斜的三种解法各有适用场景：有小表用Broadcast最省事；倾斜key固定就单独处理；两张都是大表才考虑加盐打散。Spark 3.0之后AQE可以自动处理部分场景，但手动优化仍然是基本功。面试时能结合实际场景说出「我会先看数据分布，判断能不能Broadcast，不能再看倾斜key的规律选方案」，就展示了工程思维而不是死记方案。

下一篇聊一个很多人没想到的倾斜场景——NULL值导致的倾斜：user_id有15%是NULL，JOIN时会发生什么？怎么处理？

你们生产环境加盐打散时，salt的数量（N）是怎么决定的？有没有什么公式？评论区聊聊～

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