Spark 大表 Join 调优：策略选择与加盐打散

大表 Join 变慢，通常不是"数据量太大"这一个原因，而是三类问题叠加：Join 策略选错（该走 Broadcast 的大小表走了 Sort-Merge）、个别 Key 倾斜导致单 Task 长尾甚至 OOM、以及 Shuffle 数据量过大。调优的顺序应当是：先用执行计划确认当前 Join 策略，再按表的大小关系选择策略、调整阈值，最后针对倾斜 Key 做加盐打散或两阶段聚合。下面按这个顺序展开。

Spark Join 三类策略的原理与触发条件

Spark SQL 的等值连接（Equi-Join）在物理计划上主要有三类实现：

• Broadcast Hash Join（BHJ）：将小表收集到 Driver，广播到所有 Executor，在 Executor 本地用小表构建哈希表与大表分区做 Join。全程没有 Shuffle，是最快的 Join 方式，前提是小表能放进内存。
• Shuffle Hash Join（SHJ）：两侧都按 Join Key 做 Shuffle 后，在同一个 Task 内用较小的一侧构建哈希表。它仍然有 Shuffle，但省去了排序。由于默认配置 spark.sql.join.preferSortMergeJoin=true，且构建侧需要满足"预估大小在可构建本地哈希表的范围内"，它很少被自动选中，通常要通过 Hint 强制使用。
• Sort Merge Join（SMJ）：两侧都按 Join Key 做 Shuffle 并按 Key 排序，然后两个有序流归并。它不要求一侧能装进内存，是大表对大表的默认策略，但排序本身有额外开销。

触发条件可以简单记为：能广播就广播，不能广播时默认走 Sort-Merge，只有显式 Hint 或关闭 preferSortMergeJoin 时才走 Shuffle Hash。另外，如果两侧表都做了 Bucket 且分桶数与桶内 Key 一致，Spark 可以用 Bucket Join 直接省掉 Shuffle 和排序。

先判断当前到底用了哪种 Join

调优的第一步不是改参数，而是看执行计划。

EXPLAIN
SELECT /*+ BROADCAST(d) */ o.order_id, d.user_name
FROM orders o JOIN dim_user d ON o.user_id = d.user_id;

执行计划中会出现三类节点名，对应三种策略：

• BroadcastHashJoin：已经走了广播，无需再调阈值。
• SortMergeJoin：两侧都大，或一侧其实很小但被错误地估算成大数据。
• ShuffleHashJoin：走的是哈希连接。

如果看到的是 SortMergeJoin，但其中一个输入是明显的小表（比如几千行的维表），说明统计信息不准或阈值太低，优先检查 spark.sql.autoBroadcastJoinThreshold。Spark UI 的 SQL 页签也能看到每个 Stage 的 Shuffle 读写量、Task 耗时分布，这些信息比猜测更能定位问题。

Broadcast Join 阈值调优

spark.sql.autoBroadcastJoinThreshold 的默认值是 10MB（10485760 字节），当一侧表（或经过过滤后）的预估大小低于该阈值时，Spark 自动选择广播。

spark.sql.autoBroadcastJoinThreshold=20971520   # 20MB，按需调大

实践中常见做法：

• 维表、配置表这类"真小表"用 Hint 显式指定：/*+ BROADCAST(t) */，不依赖估算。
• 阈值调大时要谨慎：广播表会被复制到每个 Executor，阈值翻倍意味着每台机器上的内存占用翻倍；在 Executor 内存紧张或大集群上，盲目调大容易引发 Driver/Executor OOM。
• 广播的数据也会被 Driver 持有，多个大广播叠加时 Driver 内存风险最高，建议同时关注 Driver 内存配置。

更完整的阈值与策略说明可以参考 Spark 官方 SQL 性能调优文档，结合版本差异核对具体参数。

Skew Join 症状识别

数据倾斜在 Join 场景的典型表现：

• 某个 Stage 里个别 Task 运行时间远长于其他 Task（长尾）。
• 长尾 Task 的 Shuffle Read 数据量远大于中位数（例如一个 Task 读了几 GB，其他 Task 只有几十 MB）。
• 长尾 Task 频繁 GC、Spill 到磁盘，甚至直接 OOM。

注意区分两类倾斜：一类是源数据本身在某个 Key 上大量重复（比如空值、默认值、热点用户），另一类是 Shuffle 之后分区大小不均。前者在读取阶段就能看到某个 Task 处理量异常，后者要在 Shuffle Read 阶段才能看到。判断方法很简单：看 Spark UI 里 Stage 的 Task 耗时直方图，或者对比各 Task 的 Shuffle Read 字节数。

先用 AQE 的自动倾斜优化

从 Spark 3.2 起 AQE（Adaptive Query Execution）默认开启，其中 spark.sql.adaptive.skewJoin.enabled 默认也是 true。它的判定规则是：一个分区同时满足"大于 spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes（默认 256MB）"且"大于所有分区大小的中位数乘以 spark.sql.adaptive.skewJoin.skewedPartitionFactor（默认 5）"，才被认定为倾斜分区，然后自动把倾斜分区拆分，并把另一侧对应的分区复制多份来对齐。

spark.sql.adaptive.enabled=true
spark.sql.adaptive.skewJoin.enabled=true
spark.sql.adaptive.skewJoin.skewedPartitionFactor=5
spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes=268435456

先开 AQE 再考虑手工加盐：如果倾斜不严重，AQE 的自动拆分通常就够了；只有 AQE 拆不动（比如某个 Key 的数据量占到单侧八成以上，拆分后复制放大倍数过大）或版本较老时才需要手工加盐。

加盐打散与两阶段聚合实操

手工加盐的思路是：给倾斜的 Join Key 追加一个随机后缀，把一个大 Key 拆成多个小 Key，让数据分散到更多 Task；对应的另一侧（通常是维表/小表）把该 Key 复制成同样数量的副本。示例如下。

第一步，给大表加上盐列（盐值取 0 到 N-1 的随机数），小表侧把热点 Key 展开成 N 份：

SELECT o.order_id, d.user_name
FROM (
  SELECT order_id, user_id, floor(rand() * 100) AS salt
  FROM orders
) o
JOIN (
  SELECT user_id, user_name, explode(sequence(0, 99)) AS salt
  FROM dim_user
) d
ON o.user_id = d.user_id AND o.salt = d.salt;

这里把热点用户在小表里复制成 100 份，大表侧每个订单随机落到其中一个桶，热点 Key 的负载就被摊到 100 个 Task 上。注意加盐只对等值 Join 有效，且必须两侧同时加同样的盐规则。

如果倾斜发生在 GroupBy 而不是 Join，可以用两阶段聚合：先按 (key, salt) 分组做局部聚合，再去掉盐按 key 聚合一次：

SELECT key, sum(cnt) AS total
FROM (
  SELECT key, salt, count(*) AS cnt
  FROM t
  GROUP BY key, salt
)
GROUP BY key;

两阶段聚合的要点是第一阶段必须保留 salt，否则局部聚合会把不同盐的同一 Key 混在一起，第二阶段就失去意义。

常见参数与加盐示例汇总

把高频参数集中管理，避免散落在各种作业里：

spark.sql.autoBroadcastJoinThreshold=10485760    # 广播阈值，默认 10MB
spark.sql.join.preferSortMergeJoin=true          # 默认倾向 Sort-Merge
spark.sql.shuffle.partitions=200                 # Shuffle 默认分区数，按数据量调整
spark.sql.adaptive.enabled=true                  # AQE，3.2 起默认开启
spark.sql.adaptive.coalescePartitions.enabled=true   # 合并过小分区
spark.sql.adaptive.skewJoin.enabled=true         # 自动倾斜处理

一个完整的大表-大表加盐示例。假设订单表与用户行为表都很大、无法广播，且 user_id 存在热点：

SELECT o.order_id, b.behavior
FROM (
  SELECT order_id, user_id, floor(rand() * 50) AS salt
  FROM orders
) o
JOIN (
  SELECT user_id, behavior, explode(sequence(0, 49)) AS salt
  FROM user_behavior
) b
ON o.user_id = b.user_id AND o.salt = b.salt;

这里两表都要做 Shuffle：热点 user_id 的订单被随机摊到 50 个分区，行为表侧把热点用户复制成 50 份，保证每个盐桶都能匹配上。注意"广播 + 加盐"不要叠加：小表已经广播时，Join 本身不会产生 Shuffle 倾斜，此时加盐只对后续聚合有意义，反而徒增副本开销。实际操作中把"确认广播是否生效、量化倾斜比例、决定盐值数量"三步分开做，避免多个优化叠加后互相掩盖问题。

权衡与失败模式

加盐不是免费的，常见的坑包括：

• 加盐后无法利用原 Key 的过滤下推，某些谓词（如 WHERE user_id = 123）会失效，反而更慢。
• 盐值数量 N 的选择：太小打散效果不明显，太大则小表副本膨胀、Shuffle 数据量成倍增加，热点 Key 不多时反而浪费。
• 加盐后有后续聚合时，必须走两阶段聚合，否则结果被拆散后无法合并。
• 空值与默认值造成的倾斜要先在业务层处理，不要在加盐时把空值随机化，否则语义会变。
• AQE 的自动倾斜优化只作用于 Shuffle 之后的倾斜，如果倾斜发生在 Map 端（如单个源文件过大），要靠调整分区数或拆分文件解决，AQE 帮不上忙。
• 用 Hint 强行广播大表（/*+ BROADCAST(大表) */）会把大表复制到每个 Executor，直接引发 Driver/Executor OOM，且症状容易误判为数据倾斜；Hint 只能用来确认"确定的小表"，不要拿来压大表。

下一步

按这个顺序动手：先用 EXPLAIN 确认几个关键 Join 分别走了什么策略；把确定的小表用 BROADCAST Hint 固定下来；打开 AQE 并保留默认倾斜参数跑一轮，用 Spark UI 对比 Task 耗时分布；如果仍有长尾，用 SQL 统计每个 Key 的行数分布找出真正热点，再针对性地对那一个 Key 加盐，而不是全局加盐；最后用"调优前后同一 SQL 的 Stage 耗时与 Shuffle 量"对比验证效果。优先解决"该广播没广播"和"AQE 能自动处理的倾斜"，手工加盐留到最后。