Spark SQL 3.0 中倾斜 JOIN 应优先启用 AQE 自动优化,即设置 spark.sql.adaptive.enabled=true 和 spark.sql.adaptive.forceOptimizeSkewedJoin=true,AQE 运行时自动识别并拆分热点 shuffle partition,无需预知热点 key 或改写 SQL。

Spark SQL 3.0 中倾斜 JOIN 不该靠手动加盐硬扛,优先开 AQE + spark.sql.adaptive.forceOptimizeSkewedJoin。 这个配置在 3.0+ 默认关闭,但只要打开,就能自动识别并拆分热点 key,比写 CASE WHEN 加盐更稳、更少出错。
为什么 Spark SQL 3.0 的 AQE 能自动处理倾斜 JOIN
AQE(Adaptive Query Execution)在运行时能观测到每个 shuffle partition 的实际数据量。当某个 partition 的 size 超过阈值(默认是其他 partition 中位数的 5 倍),AQE 就会触发 skew join 优化:把那个大 partition 拆成 N 个小 partition,分别跟另一张表中对应 key 的数据做局部 join,最后 union 结果。
它不依赖你提前知道哪些 key 是热点,也不要求你改 SQL 逻辑——只要执行计划里有 ShuffleExchangeExec,AQE 就可能介入。
- 必须开启 AQE:
spark.sql.adaptive.enabled=true - 必须显式启用倾斜优化:
spark.sql.adaptive.forceOptimizeSkewedJoin=true(注意不是spark.sql.adaptive.skewJoin.enabled,后者在 3.0 中已废弃) - 默认只对
Inner和LeftOuter生效;RightOuter和FullOuter需要额外设spark.sql.adaptive.skewJoin.enabled(仅限 3.2+) - 拆分粒度由
spark.sql.adaptive.skewJoin.skewPartitionThresholdInBytes(默认 256MB)和spark.sql.adaptive.skewJoin.skewPartitionFactor(默认 5)共同决定
手动加盐仍需用时:盐值必须可重现,别信 RAND()
当 AQE 无法覆盖(比如用了 /*+ BROADCAST */ hint 强制关闭 shuffle,或跨集群调度导致统计信息不准),就得回退到加盐。但生产环境用 RAND() 是高危操作——它每次执行结果不同,两张表 salt 后根本 join 不上。
正确做法是用确定性哈希:
- 大表加盐:
CASE WHEN id IN (5, 6) THEN CONCAT(id, '_', CAST(hash(id) % 10 AS STRING)) ELSE CAST(id AS STRING) END - 小表同步加盐:同样逻辑,
hash(id) % 10必须一致,且模数(这里是 10)建议设为预期热点 key 数量 × 2~3 倍 - 加盐后必须重分区:
/*+ REPARTITION(200) */或提前DISTRIBUTE BY,否则 Spark 可能沿用原 partitioner,盐白加 - 别只 salt 大表:小表不 salt,join 条件就断了;两张表 salt 逻辑不对齐,结果为空
用 SKEWED_JOIN hint 前先确认执行计划是否真走 skew 逻辑
自定义 hint 如 /*+ SKEWED_JOIN(join_key(t1.id, t2.id), skewed_values(5,6)) */ 看似方便,但它本质是把原 JOIN 拆成两路:一路过滤掉热点 key 单独处理,另一路跑正常 JOIN 再 union。这要求你 100% 确认哪些 key 是热点,且这些 key 在两张表中都存在。
容易翻车的点:
- hint 不生效:Spark 版本低于 3.0.0(该 hint 是社区补丁,未进官方主线),或 optimizer 规则被禁用
- 小表没热点但大表有:hint 仍会强制拆分,反而引入额外 shuffle 和 union 开销
- 热点 key 在小表里分布异常(比如小表中 id=5 的记录只有 1 条,但大表有 500 万条):拆分后小表那条记录要广播 10 次,内存压力陡增
- 验证方式:看
EXPLAIN EXTENDED输出里是否有Union节点和两个Filter子计划,而不是单纯多了一行 hint 注释
真正难的不是选哪种方案,而是判断“这个 JOIN 到底是不是真倾斜”——很多所谓倾斜,其实是小表没广播成功、空值没过滤、或统计信息过期导致优化器选错 join 策略。先跑 ANALYZE TABLE,再看 Spark UI 里 shuffle write/read 的分布直方图,比急着加盐靠谱得多。










