企业官网建设流程全解析

很多人一听到 Spark join，脑子里冒出来的是 inner join、left join 这些 SQL 语法。真正让运行时间差出十倍的地方，不是这些逻辑写法，而是 Spark SQL 在物理计划里最终选用了哪一种 join 实现：Broadcast Hash Join、Shuffle Hash Join、Sort Merge Join，以及在特殊情况下会冒出来的 Broadcast Nested Loop Join。一共也没几个，但每个的触发条件、内存开销、适用数据量完全不同。这篇文章把这几种方式掰开揉碎讲一遍，顺便附上我在排查慢 join 时实际用到的 explain 方法和调参思路，适合每一个被 join 慢查询折磨过的开发者。

1. 先分清逻辑 Join 和物理 Join 的区别

1.1 为什么同一个 SQL 会跑出不同计划

“join 有几种方式”这个问题，最容易踩的误区是把 SQL 写法当成执行方式。INNER JOIN、LEFT JOIN、RIGHT JOIN、FULL JOIN属于逻辑层，表示业务上要怎么关联；而 Spark 真正执行时会根据数据量、关联键、统计信息、当前配置，从物理层选一个实现算法。

我举个例子，同样是：

SELECT * FROM orders o JOIN users u ON o.user_id = u.id

如果users表只有几千行，Spark 大概率会选择 Broadcast Hash Join，直接把users广播给每个 executor，orders表根本不需要 shuffle。如果users表变成几亿行，Spark 就会老老实实走 Sort Merge Join，两边都 shuffle，再排序合并。

Spark Catalyst 优化器做的事情，很像导航软件根据当前路况选路线。你写 SQL 只表示“我要从 A 到 B”，导航选高速还是选省道，取决于路况、里程、实时拥堵。Spark 选哪种 join，则取决于表大小、等值条件、分区方式、内存配置和统计数据。

所以，排查 join 慢的第一步不是盲目加 executor，而是先看明白当前这个 SQL 最终跑在哪个物理算子上。

1.2 三种核心实现方式一览

先把最常见的情况列出来。Spark 里真正干活的三种 join 实现是：

其中前三个是开发时主要打交道的东西。Broadcast Nested Loop Join和Cartesian Product属于“会出现但最好别让它出现”的异常情况，后面单独说。

2. Broadcast Hash Join：小表广播救急指南

2.1 执行原理与 Build Side 选择

Broadcast Hash Join 的思路说穿了就一句话：把其中一张小表复制到所有 executor 上，让每个 executor 在本地内存里建一张哈希表，然后大表每个分片挨个去查这张哈希表。

这里有两个关键词：Build 侧和 Probe 侧。

Build 侧是要建哈希表的那张表，通常选小表；Probe 侧是拿着每条记录去查哈希表的那张表，通常是大表。Spark 默认喜欢把 Build 侧放在右侧，也就是BuildRight，因为很多 join 写法里右侧更容易是小表。如果优化器判断左侧更小，也会出现BuildLeft。

因为小表已经被完整广播到每个节点，大表不需要因为关联键做全局 shuffle。你可以把广播理解成“人手发一份通讯录”，然后每个人在自己的电脑上直接查通讯录，不用把所有数据都汇到一块再查。

这个方案最大的优点是快，因为省掉了一次最昂贵的 shuffle。最大的缺点是内存，如果广播的表太大，每个 executor 都要在内存里承载一份完整哈希表，内存不够就直接 OOM。

2.2 触发条件、参数与一个完整示例

触发 Broadcast Hash Join 的几个条件：

• 必须存在等值关联条件，比如a.id = b.id。
• 其中一张表的预估大小要小于等于spark.sql.autoBroadcastJoinThreshold。
• 表统计信息准确，或者你主动通过 hint 告诉优化器要广播哪张表。

spark.sql.autoBroadcastJoinThreshold的默认值在很多版本里是10m，也就是 10MB。这个值可以改，但不能无脑调大。我曾经在模拟项目里把它调到 200MB，小表广播是不慢，可惜 executor 内存被广播数据占掉一大块，别的任务全都跟着遭殃。

实操时可以先设置参数，再用 SQL 加 hint：

spark.conf.set("spark.sql.autoBroadcastJoinThreshold", 20 * 1024 * 1024) spark.conf.set("spark.sql.broadcastTimeout", 600) SELECT /*+ BROADCAST(u) */ o.order_id, u.user_name FROM orders o JOIN users u ON o.user_id = u.id

然后看一眼执行计划：

df.explain("extended")

如果走了广播，物理计划里会出现类似：

== Physical Plan == BroadcastHashJoin [user_id#4L], [id#0L], Inner, BuildRight, false :- Project [order_id#2L, user_id#4L] : +- FileScan parquet orders +- BroadcastExchange HashedRelationBroadcastMode(List(input[0, bigint, false]), false) +- Project [id#0L, user_name#1L] +- FileScan parquet users

看到BroadcastHashJoin和BroadcastExchange，就说明广播生效了。BuildRight表示右侧的users被广播。

2.3 加 hint 要注意什么

hint 不是灵丹妙药。/*+ BROADCAST(u) */写上去，优化器大概率会优先考虑广播，但如果被广播的表实际非常大，executor 内存会被打爆，任务反而比原来死得更快。

我之前踩过一个坑：有一张“看起来很小”的维度表，文件压缩后只有 30MB，我直接加了 broadcast hint，结果每个 executor 把表解压后建哈希表，内存直接飙高，整个作业频繁 GC。后来一查，这张表虽然文件不大，但行数有上千万，而且同一个 key 下挂了大量重复明细，哈希表实际膨胀得厉害。

所以记住两条经验：

• 广播前不要只看文件大小，要看真实行数和 key 的重复度。
• 如果用了 broadcast 还慢，先看是不是表统计信息太旧，或者spark.sql.broadcastTimeout太短导致广播传输超时。

3. Shuffle Hash Join：适合中等表的中间路线

3.1 执行流程与触发条件

Shuffle Hash Join 是介于广播和大表排序之间的一种方案。它没有把整张表广播到所有节点，而是把两张表都按照 join key 做一次 shuffle，让相同 key 的数据落到同一个分区上。然后在每个分区内部，选择较小的那一侧建哈希表，另一侧去查。

可以这么理解：Broadcast 是“每个人手里都有一份完整小表”，Shuffle Hash 是“只把相同编号的文件放到同一张桌子上，然后在这张桌子上各自建索引”。

它的触发条件比广播严格：

• 必须等值关联。
• 两侧数据都不满足广播阈值，无法走 BHJ。
• 某个分区内的 build 侧数据要能放进内存。
• 通常需要关闭spark.sql.join.preferSortMergeJoin这个参数，或者显式使用 hint。

默认情况下spark.sql.join.preferSortMergeJoin是true，也就是当数据量不满足广播时，Spark 更倾向于走 Sort Merge Join，因为排序方案对外存溢出更友好。如果你想尝试 Shuffle Hash Join，可以这样配：

spark.conf.set("spark.sql.join.preferSortMergeJoin", false) spark.sql(""" SELECT /*+ SHUFFLE_HASH(u) */ o.order_id, u.user_name FROM orders o JOIN users u ON o.user_id = u.id """)

物理计划中如果出现ShuffleHashJoin，说明这条路径被选中了。

3.2 什么时候值得放弃默认的 Sort Merge Join

Shuffle Hash Join 省掉了 Sort Merge Join 最花时间的排序阶段。Sort Merge Join 需要先对两边分别排序，然后再归并；而 Shuffle Hash Join 在 shuffle 之后直接建哈希表，平均复杂度上更有优势。

但代价是内存。Sort Merge Join 可以依赖外排，数据量大了会 spill 到磁盘，虽然慢但不容易爆内存；Shuffle Hash Join 的哈希表一旦超过内存，要么 OOM，要么性能急剧下降。

我的经验是，不要轻易全局关掉spark.sql.join.preferSortMergeJoin。全局关了之后，所有大表 join 都可能尝试走 Shuffle Hash Join，一旦某个分区出现热点 key，整个 job 就会很难看。更稳妥的做法是单独对某条 SQL 用 hint，让优化器只在这一条 SQL 里尝试 SHJ。

如果两张表都是中等规模，比如单表几千万行，join key 的基数比较高、分布比较均匀，同时集群内存比较充裕，Shuffle Hash Join 在某些任务里确实能比 Sort Merge Join 快一些。但你要先做好分区级别内存评估，别只看全表大小。

4. Sort Merge Join：大表关联的默认主力军

4.1 排序再合并的执行流程

Sort Merge Join 是 Spark 在大多数大表等值关联场景下的默认选择。它的流程听名字就很好理解：

• 两张表都按 join key 进行 shuffle，让相同 key 的数据落到同一分区。
• 每个分区内的数据按 key 排序。
• 两边各拿一个指针，按顺序像拉链一样合并。

因为整个过程不要求把整张表放进内存，排序可以溢写到磁盘，所以它非常稳。哪怕两张表都特别大，Spark 也能通过外排把 join 跑完，只是慢。

物理计划长这样：

== Physical Plan == SortMergeJoin [user_id#4L], [id#0L], Inner :- Sort [user_id#4L ASC NULLS FIRST], false, 0 : +- Exchange hashpartitioning(user_id#4L, 200), ENSURE_REQUIREMENTS, [plan_id=...] : +- Project [order_id#2L, user_id#4L] : +- FileScan parquet orders +- Sort [id#0L ASC NULLS FIRST], false, 0 +- Exchange hashpartitioning(id#0L, 200), ENSURE_REQUIREMENTS, [plan_id=...] +- Project [id#0L, user_name#1L] +- FileScan parquet users

看到Exchange hashpartitioning和Sort同时出现，基本就是 Sort Merge Join。

这种 join 慢不一定是因为算法选错，而可能是 shuffle 分区数、数据倾斜、资源并行度不匹配。默认的 shuffle 分区数是 200，这个数字对一个 100 节点的集群可能太小，对本地开发环境又可能太大。不要盲调，要结合每个分区的数据量来定，通常让每个任务处理的数据量在 100MB 到 200MB 这个量级，然后再看监控里的 task 耗时分布。

4.2 让 Sort Merge Join 更快的两个期权

我不太建议对大表 join 轻易改用别的算法，更值钱的方向是让 Sort Merge Join 自己变快。

第一个是分桶表。如果两张表都按 join key 做相同数量的 hash 分桶，Spark 在 join 时可以直接跳过Exchange，因为相同 key 的数据已经天然落到了对应分区里。如果桶内数据还按 key 排好序，连Sort都可以省掉。

比如建表时这样设计：

CREATE TABLE fact_table ( user_id BIGINT, order_id BIGINT, amount DECIMAL(10,2) ) USING parquet CLUSTERED BY (user_id) INTO 64 BUCKETS; CREATE TABLE dim_table ( user_id BIGINT, user_name STRING ) USING parquet CLUSTERED BY (user_id) INTO 64 BUCKETS;

两张表 bucket 数一致，join key 也都是user_id，这时 Spark 就有可能做 BucketJoin。物理计划里少了Exchange，性能自然上一个台阶。这里要特别注意，分桶数和 join key 必须严格对得上，否则优化器还是会老老实实加 shuffle。

第二个是开启自适应查询执行。新版 Spark 的 AQE 能根据 shuffle 阶段的实际输出动态调整分区数，还能对倾斜的分区做拆分。如果你的版本支持，建议在任务里这样配置：

spark.conf.set("spark.sql.adaptive.enabled", true) spark.conf.set("spark.sql.adaptive.coalescePartitions.enabled", true) spark.conf.set("spark.sql.adaptive.skewJoin.enabled", true)

不同版本的默认值有差异，建议先看一下当前环境实际输出。AQE 很香，但别指望它能解决所有问题，它更多是让已经选择的路更好走，而不是帮你换一条路。

5. 其他场景：Broadcast Nested Loop Join 和笛卡尔积

5.1 不等值关联的兜底方案

如果 SQL 里没有等值条件，比如：

SELECT * FROM orders o JOIN users u ON o.user_id > u.id

这时候 Broadcast Hash Join、Shuffle Hash Join、Sort Merge Join 都无能为力，因为它们都依赖等值条件来定位数据。Spark 只能退回到 Broadcast Nested Loop Join：把其中一张表广播出去，然后每一行去另一张表里逐行匹配。

这是一个非常昂贵的操作，复杂度基本是两张表行数的乘积。如果表不大，还能勉强接受；如果两边都是大表，这个算子跑完基本等于灾难。

处理这类需求，正确思路不是调参数，而是改写 SQL。比如先通过等值条件把数据压缩到小范围，再做非等值过滤：

SELECT * FROM orders o JOIN users u ON o.user_id = u.id AND o.create_time > u.reg_time

这里o.user_id = u.id负责走正常的 hash 或 sort merge join，create_time > reg_time则在 join 完成后的过滤阶段执行。执行效率和直接写一个大于号完全不同。

5.2 什么情况会误踩 Cartesian Product

还有一种更隐蔽的情况：你明明写了 join，但忘了写 on 条件，或者 join 条件被写进了 where。Spark 可能把它当成笛卡尔积。例如：

SELECT * FROM orders o JOIN users u WHERE o.user_id = u.id

这在很多数据库里是合法的隐式 join，但 Spark SQL 的优化器可能不会自动推断出这个等值条件，最终生成 CartesianProduct 物理节点。

我见过最离谱的一次慢查询，就是线上代码里 join 条件漏了on，写成了where o.user_id = u.id，结果数据量稍微上来之后作业直接跑了几小时。后来改成标准的JOIN ... ON ...语法，物理计划立刻恢复正常。

所以写完 SQL 后，不要只看结果对不对，最好顺手explain一下，看到CartesianProduct就要警惕：要么查询本身有设计问题，要么需要显式调整关联条件。

6. 实操：用 explain 判断当前到底用的是哪种 Join

6.1 一条 SQL 的物理计划怎么看

在写 Spark 作业时，最直接的诊断方式就是打印物理计划。

val df = spark.sql(""" SELECT o.order_id, u.user_name FROM orders o JOIN users u ON o.user_id = u.id """) df.explain("extended")

如果是在 SQL 编辑器里，也可以写：

EXPLAIN SELECT o.order_id, u.user_name FROM orders o JOIN users u ON o.user_id = u.id

explain("extended")会输出 Parsed Logical Plan、Analyzed Logical Plan、Optimized Logical Plan、Physical Plan 几层。物理计划在最下面，找到最顶层的 Join 算子，基本就是最终执行方式。

有些时候同一段逻辑会出现多个 Join 节点，尤其是子查询里面也有 join。这时候要从上往下看，先找最外层那个 Join，再结合子查询里的算子一起判断。

6.2 物理计划特征速查表

我整理了一个平时排查用的对照表，看到关键词可以直接判断：

这个表我在本地和集群环境都验证过，基本能覆盖绝大多数场景。看到实际执行方式之后，再决定是改 SQL、调阈值、换 hint，还是重新设计分桶。

7. 调参实战：一次慢 Join 的定位与优化记录

7.1 模拟场景与基线

某次处理一个离线数仓任务时，需要把一张 2 亿行的用户行为事实表和一 张维度表做关联。事实表按user_id关联维度表拿用户名称。乍一看这是典型的“大表 join 小表”，可任务跑了 18 分钟都没完。

我先用 explain 看了一眼物理计划，发现走的是 SortMergeJoin。问题就出在默认广播阈值是 10MB，而维度表的预估大小超过了阈值，所以 Spark 没有广播它。

我当时的处理过程分成三步。

第一步，先给维度表更新统计信息。表数据早就更新了，但统计信息还是旧的，优化器看到的可能是一个被高估的体积：

ANALYZE TABLE dim_user COMPUTE STATISTICS;

第二步，确认维度表真实大小之后，把广播阈值调到 64MB，并且给关联 SQL 加广播 hint：

spark.conf.set("spark.sql.autoBroadcastJoinThreshold", 64 * 1024 * 1024) spark.conf.set("spark.sql.broadcastTimeout", 600) spark.sql(""" SELECT /*+ BROADCAST(d) */ f.order_id, d.user_name FROM fact_event f JOIN dim_user d ON f.user_id = d.user_id """)

第三步，再看物理计划，确认变成了BroadcastHashJoin ... BuildRight。同一个逻辑，最终运行时间从 18 分钟压到了 4 分钟以内，而且 shuffle 数据量几乎消失。

7.2 从一次慢 Join 到定位的排查路径

这次调优给我的最大感受是：遇到 join 慢，不要直接猜，按固定路径排查最省时间。

第一，先看执行计划，确认当前是哪种 join。如果执行计划都没看，直接调spark.sql.shuffle.partitions，很可能白调。

第二，看表统计信息和广播阈值。ANALYZE TABLE这一步很多人会忽略，但统计信息不准，Spark 对表大小的判断就是错的。表明明很小，却走了 SortMergeJoin，八成是这里出了问题。

第三，看 join key 的分布情况。如果某个 key 特别集中，比如user_id=0这种脏数据占了 70%，那么就算你换了 Broadcast 或调整分区数，依然会被这一个 key 拖死。判断方式很简单：

SELECT user_id, COUNT(*) AS cnt FROM fact_event GROUP BY user_id ORDER BY cnt DESC LIMIT 10;

如果头部 key 特别突出，就要考虑数据清洗或者加盐拆分。加盐的基本思路是：对倾斜 key 打散成多个随机后缀，让它们分散到不同分区，再结合广播小表或二次 join 把结果归并。具体实现有点绕，但遇到极端倾斜时很管用。

8. 常见问题速查和避坑心得

8.1 常见问题对照表

下面这些问题都是我实际踩过或者帮别人排查过的，整理成速查表放在这里。

8.2 几个值得记住的经验

最后分享几条通用经验。

别总想着换算法，先确认当前算法是不是本来就合理。两个大表关联走 SortMergeJoin 是再正常不过的事，直接去调小广播阈值反而可能让 executor OOM。真正要调的往往是分区数、分桶策略、统计信息和数据分布。

广播阈值不是越大越好。它只解决“小表被误判成大表”的问题，但如果一张表本身就有上百 MB，强行广播只会把内存压力转嫁到每个节点。我看到很多人把所有 join 都加上 broadcast hint，最后集群内存被打满，得不偿失。

spark.sql.shuffle.partitions也不要拍脑袋设。它决定的是 shuffle 后分区数量，设太小会导致单个任务处理数据量过大，设太大又会产生大量小任务，调度本身都要花时间。比较好的做法是结合集群并行度和数据量做几组对比，观察 task 的平均耗时和 GC 情况再定。

我自己踩过最冤的坑，是查了半天参数，最后发现统计信息过期，一张 200MB 的表被当成无限大，死活不走广播。重建统计信息之后，计划立刻变了。所以遇到 join 慢，先不要忙着加资源，打开 explain，先问一句：Spark 到底选了哪种 join，为什么选它。把这几个算子看熟，比背一堆配置更管用。