上个月排查一个跑批任务的时候我碰见了一条看起来再普通不过的SQL左侧是300万行的用户维表右侧是8亿行的订单事实表写法是dim_user u LEFT JOIN fact_order o ON u.user_id o.buyer_id。任务跑了两个多小时还拖着队列打开执行计划一看走的是 SortMergeJoin两个 Exchange 在疯狂 shuffle。按理说用户维表才几十MBSpark 完全可以用 MAPJOINBroadcastHashJoin把维表广播到每个节点为什么连资格都没捞到这个问题折腾了我一个下午才彻底想明白。它不涉及什么冷门配置也不是数据倾斜纯粹是“LEFT JOIN 的方向”和 Spark 广播连接机制的底层约束撞在一起了。今天就把整个排查过程、原理和几种解法完整写出来希望能让大家少走我走过的弯路。内容适合正在做数据开发、数仓建模以及日常要写 Spark SQL 调优的工程师尤其是对广播连接了解得比较模糊、只知道“小表要广播”的人。1. MapJoin为什么快以及这里为什么没走MapJoin1.1 一条SQL希望变成的执行计划MapJoin 这个叫法是从 Hive 时代留下来的Spark 里的正式名字是 BroadcastHashJoin也就是广播连接。它的核心思想很简单不把两张表都拖到 shuffle 里重新分区而是把里面比较小的那张表完整广播到每一个 executor 上另一张大表留在本地逐行去和这份“广播小表”做哈希匹配。你可以把它想象成考试时每个考场都发了一份公式手册考生不需要为了查一条公式专门跑到学校图书馆去。大表就是考生手里的试卷小表就是那本公式手册。如果走 SortMergeJoin相当于每道题都要先把试卷和图书馆里的书按页码重新分类汇总再回到考场上答题。谁快谁慢一眼就能看出来。所以理想情况下这条 SQL 的计划应该是BroadcastHashJoin :- BroadcastExchange把dim_user发到所有executor : - FileScan dim_user - FileScan fact_order大表没有 Exchange没有 shuffle只有一份几十MB的广播数据在网络里分发。这也是我们做调优时最想看到的物理计划。1.2 Build Side和Stream Side到底是谁很多初学者会问Spark 是怎么决定广播哪一张表的这里有两个概念必须先搞清楚。广播的那一侧叫 Build Side也就是用来构建哈希表的那一侧。逐行扫描的另一侧叫 Stream Side。对于等值连接来说Spark 并不是随便选一侧广播的它在源码里对不同的 join type 有硬性约束。简单归纳下来是这样Join类型BroadcastHashJoin 的 Build Side 限制Inner Join左右两侧都可以作为 Build Side优化器倾向于选统计值更小的一侧Left Outer Join只能把右表作为 Build SideRight Outer Join左右两侧都可以作为 Build Side优化器倾向于选更小的一侧Full Outer Join不能使用 BroadcastHashJoin这里最关键的是 Left Outer Join。为什么 LEFT JOIN 不能广播左表因为 LEFT JOIN 的语义要求左表的每一行都必须保留在最终结果里包括那些在右表中找不到匹配的行。如果 Spark 把左表广播出去、让右表逐行去探测那么每个分区只能看到右表的一部分谁来决定“左表这一行在全量右表里完全没匹配”这个判断在分布式环境下会变得非常麻烦。所以 Spark 干脆从规则层面不允许 LeftOuter 把左表作为 Build Side。回到我们那条SQLSELECT u.user_id, u.user_name, o.order_id, o.amount FROM dim_user u LEFT JOIN fact_order o ON u.user_id o.buyer_id;左表是小表 dim_user右表是大表 fact_order。LEFT JOIN 的硬性约束决定 Spark 只能把右表当作 Build Side。可是右表是 8 亿行的大表怎么可能广播于是 Spark 只好退回到 SortMergeJoin两张表全部 shuffle。小表虽然小但是因为它站在了“不能广播”的位置上整个 MapJoin 优化直接失效。1.3 阈值与统计信息Spark如何判断“可以广播”除了 join type 的方向限制Spark 在是否能广播这件事上还有一道门槛就是spark.sql.autoBroadcastJoinThreshold默认值是 10MB。也就是说那张表在逻辑计划里的估算大小如果小于 10MBSpark 才会考虑广播。注意这里说的“表大小”不是我们体感上的大小而是 Spark 从三方面拿到的估算值Hive Metastore 里的统计信息、文件系统的文件大小、以及ANALYZE TABLE手动收集的统计信息。如果这些信息缺失或者过期Spark 对表大小的判断就会走样进而影响它是否选择广播。所以“小表 LEFT JOIN 大表”失效的本质不是小表不够小而是两个约束叠在一起LeftOuter 不允许广播左表而右表又太大没有资格广播。明白了这个根因后面的解决方案就有了方向。2. 最优解把LEFT JOIN改写成RIGHT JOIN让广播侧落在小表2.1 改写原理既然 LEFT JOIN 只能把右表当作 Build Side那我们能不能把“小表”挪到右表去答案是可以但要通过改写 join 的方向来实现。A LEFT JOIN B在语义上等价于B RIGHT JOIN A。前者保留 A 的全部行后者保留 A 的全部行两者只是把表的位置换了一下。原SQL里小表 dim_user 在左、大表 fact_order 在右改成fact_order RIGHT JOIN dim_user之后小表跑到了右表位置而 RightOuter 是允许右表作为 Build Side 的。这样一来Spark 会审视右表 dim_user 的大小。如果它满足广播阈值优化器就会选择 BuildRight生成 BroadcastHashJoin大表不再 shuffle。这是我在实际调优中用得最多、效果也最直接的方案。2.2 具体SQL改造示例改造后的 SQL 长这样SELECT u.user_id, u.user_name, o.order_id, o.amount FROM fact_order o RIGHT JOIN dim_user u ON u.user_id o.buyer_id;列顺序和原SQL保持一致业务代码基本不用改。如果 dim_user 的实际大小超过了默认的 10MB 阈值就再用广播提示给它加一道保险SELECT /* BROADCAST(u) */ u.user_id, u.user_name, o.order_id, o.amount FROM fact_order o RIGHT JOIN dim_user u ON u.user_id o.buyer_id;执行计划会变成类似这样BroadcastHashJoin [buyer_id#12], [user_id#5], RightOuter, BuildRight :- BroadcastExchange HashedRelation : - FileScan dim_user - FileScan fact_order看到RightOuter, BuildRight基本就放心了。这次调优的效果很明显同一个任务从 2 小时 40 分降到了 18 分钟而且集群的 shuffle 量从几百GB骤降到几十MB。不是玄学就是少做了一次大表全量 shuffle。2.3 改写时容易踩的坑改写 SQL 看起来只是换了表的位置但实际操作里有几个坑我一个个列出来。第一个坑是列顺序。如果上游消费结果集的时候是按下标取值改了 join 方向后必须保证 SELECT 字段顺序和原来完全一致。我在测试环境见过有人把u.user_id和o.order_id的顺序写反结果下游数据对不上排查了半天。第二个坑是过滤条件的位置。LEFT JOIN 改写为 RIGHT JOIN 后ON 子句里的条件要保持等价。如果在原SQL里写的是dim_user u LEFT JOIN fact_order o ON u.user_id o.buyer_id AND o.amount 100那改写后同样要放在 ON 子句里。如果把它移到 WHERE 子句语义就变了WHERE 里过滤大表字段会把 NULL 行也过滤掉结果集少掉一批本来该保留的用户。第三个坑是主表的唯一性。dim_user 作为主表如果它里面有重复的 user_idLEFT JOIN 之后这些重复行会各自匹配一次订单改写后 RIGHT JOIN 也一样行数不会变化。但如果右表有重复 keyLEFT JOIN 会产生笛卡尔式放大这种放大和 join 方向无关问题可能出在数据质量上别误判成改写造成的。第四个坑是大表列里的 NULL。LEFT JOIN 之后未匹配的用户对应的订单字段本来就是 NULL这是业务上要接受的现实。有些同事看到 NULL 就以为是 bug实际上只要大表字段本身也有 NULL 值你根本无法只凭最终结果判断到底是“未匹配”还是“本来就是 NULL”。这个在改写前就应和业务方确认清楚。3. 兜底方案Hint、阈值调参与统计信息修正3.1 Broadcast Hint的正确打开方式如果因为某些原因不能改 SQL 方向还可以用 Hint 去影响优化器。Spark 支持BROADCAST和MAPJOIN两种写法SELECT /* MAPJOIN(dim_user) */ ... SELECT /* BROADCAST(dim_user) */ ...这里必须强调一个容易误导人的点在小表 LEFT JOIN 大表的原始 SQL 上加/* BROADCAST(dim_user) */是没用的。因为 LeftOuter 这个 join type 从源码规则上就不允许左表作 Build SideHint 只能告诉优化器“这张表适合广播”不能突破 join type 的硬约束。Spark 看到这个 Hint 后如果发现放不下就会安静地忽略它继续走 SortMergeJoin。所以在 RIGHT JOIN 改写版本的 SQL 上加 Hint效果才是稳定可期的。正确姿势是把 Hint 加在已经在右表位置的维表名上让 BuildRight 的意图更加明确。3.2 阈值怎么调spark.sql.autoBroadcastJoinThreshold默认 10MB实际生产环境里很多“小表”远不止 10MB。比如我们这条SQL里的 dim_user 是 60MB那就得手动调大阈值spark.sql.autoBroadcastJoinThreshold100m spark.sql.broadcastTimeout600这里要特别提醒调阈值不是调得越大越好。广播表是要完整塞进每个 executor 内存的一张 60MB 的表序列化之后加上构建哈希表、GC 开销实际占用可能是 300MB 甚至更多。如果集群上每个 executor 只有 2GB 内存而你把阈值调到 500MB一旦广播表真的超过预期轻则拖慢整个任务重则直接 OOM。我自己的经验是三个原则广播表大小别超过单个 executor 可用内存的 20%广播数据的总量要低于 DM 网卡和队列带宽的承受范围优先用 Hint 而不是全局阈值因为 Hint 是局部影响全局阈值会无差别地作用于所有 join风险面太大。另外如果你们用的是 Spark 3.2 以上版本还有一个自适应参数spark.sql.adaptive.autoBroadcastJoinThreshold值得关注。AQE 会在 shuffle 执行过程中根据实际读到的数据量重新判断是否有机会转换为广播连接。有时候一开始没进 MapJoinAQE 结束后反而能救回来。但它和spark.sql.adaptive.coalescePartitions.enabled的联动比较微妙建议在测试环境先压一轮再上线。3.3 用EXPLAIN验证是否真的MapJoin调优最怕的是一顿操作猛如虎计划还是原样。所以每次改完 SQL我都会先把执行计划打出来看一眼EXPLAIN EXTENDED SELECT /* BROADCAST(u) */ u.user_id, u.user_name, o.order_id, o.amount FROM fact_order o RIGHT JOIN dim_user u ON u.user_id o.buyer_id;如果走上 MapJoin物理计划里会有BroadcastHashJoin和BroadcastExchange节点而且大表一侧没有Exchange。如果走 SortMergeJoin你会看到左右两侧各有一个Exchange那就说明广播优化没生效。我还习惯在 Spark UI 的 SQL 页签里看 Shuffle 字节数。MapJoin 生效时Shuffle Read/Write 的数据量会断崖式下降这个数字比执行计划更直观也更容易说服团队里其他同事。4. 当小表不适合广播时分步JOIN与其它手段4.1 先裁剪大表再做外连接不是所有“小表”都适合广播。如果维表有 2GB强行广播会导致后来所有并行 task 都背着这个大包袱反而比 shuffle 更慢。这种情况下第一条策略是对大表做裁剪。对大事实表来说最常见的裁剪方式是时间分区。原SQL如果写成FROM fact_order o JOIN dim_user u ON u.user_id o.buyer_id WHERE o.dt 2024-01-01那么这个过滤后的右表可能只有几十GB依然很大但有时已经足够小。实际操作里我会先看过滤后的右表到底多大。如果能在几百MB以内就值得尝试广播如果过滤后仍然几个GB那还是别广播了。更进一步的裁剪方式是“用维表 key 去圈定大表的数据范围”。虽然纯 SQL 里IN (SELECT ... FROM dim_user)的执行效率通常不高但在某些数仓分层场景下可以提前把事实表的 buyer_id 字典集合落成一个中间表再做裁剪。这个思路的本质就是先把大表“瘦身”再让外连接的上游输入变小。4.2 内连接加补NULL的拆分思路如果 SQL 结构很复杂不得不维持原来的 LEFT JOIN还有一种拆方案的思路把整个匹配逻辑拆成“匹配部分 未匹配补NULL部分”。匹配部分可以用SELECT /* BROADCAST(u) */ u.user_id, u.user_name, o.order_id, o.amount FROM dim_user u JOIN fact_order o ON u.user_id o.buyer_id这里是 inner join优化器允许把小表当 Build Side即使它在左表位置也能广播。未匹配部分则用 anti join 找出维表里没有订单的用户再把大表字段置成 NULL最后两者 UNION ALL。这种写法确实能把广播用起来但它的代价也很明显要对大表扫描两次一次是内连接一次是 anti join。除非大表能提前被过滤得很小否则整体耗时不一定比重写为 RIGHT JOIN 更短。所以我的建议是它只适合那些“SQL 由上游工具自动生成、无法手动改表顺序”的僵化场景手工可改的情况下优先用第 2 节的重写方案。4.3 ShuffleHashJoin/SMB等替代方案广播连接不行的时候也别死磕 MapJoin。还有两个方向可以考虑。第一是 ShuffleHashJoin。通过 Hint 强制SELECT /* SHUFFLE_HASH(u) */ ...它会把两张表都 shuffle但选择小表作为构建哈希表的一侧避免 SortMergeJoin 中额外的排序和归并。在 join key 本身比较均匀、没有明显倾斜的时候ShuffleHashJoin 比 SortMergeJoin 快不少。第二是分桶表 SMB Join。如果事实表和维表都预先把 join key 分成同样数量的桶而且两边桶的哈希规则一致那么 join 时两个表可以按桶直接读取不需要 shuffle。这个方案对 ETL 建模要求比较高适合高频运行的固定 join一次性建模成本摊薄到多次任务后很划算。5. 常见问题与排查经验实录5.1 排查步骤清单如果你现在也在处理类似“小表 LEFT JOIN 大表”导致变慢的问题可以按下面这套顺序来排查。第一步用EXPLAIN先看物理计划里是BroadcastHashJoin还是SortMergeJoin。这是最直接的证据。第二步确认 join type。是 Inner、LeftOuter 还是 RightOuter。LeftOuter 的话先锁定右表是否能广播。第三步check 左右表的统计信息。ANALYZE TABLE dim_user COMPUTE STATISTICS能解决大部分“明明很小却不广播”的问题。第四步看广播表估算大小和 executor 内存。在 Spark UI 的 Executors 页签里看每个 executor 实际内存使用心里有数再调阈值。第五步改完 SQL 之后做数据比对。用 COUNT 和抽样字段的 MD5 校验改写前后结果是否一致尤其是 NULL 分布和重复行。5.2 高频问题速查表现象可能原因处理方式广播等待时间过长广播数据量太大超过spark.sql.broadcastTimeout默认 300 秒调大 timeout或放弃广播改用 ShuffleHashJoinExecutor OOM广播表太大或者 executors 内存不足降低广播阈值给小表瘦身增大 executor memory overhead明明很小却没走 BroadcastHashJoin统计信息缺失或过期执行ANALYZE TABLE刷新统计信息加BROADCAST(hint)没生效LEFT JOIN 中试图广播左表违反 join type 约束改成 RIGHT JOIN把目标表放到右表位置改写 RIGHT JOIN 后结果和原来不一致列顺序变了、ON/WHERE 条件位置不对、主表重复 key逐项对比 SELECT 列、ON 条件、重复行数广播后任务变慢广播传输消耗大或者广播表在多个 stage 反复序列化使用 ReusedExchange或评估是否该走 shuffle5.3 我踩过的几个坑第一次踩这个坑时我犯的错误就是只加 Hint 不改方向。在原 SQL 上写了/* MAPJOIN(u) */满怀期待地等执行计划结果 Spark 依然走 SortMergeJoin。当时一度怀疑是 Hint 拼写错了后来翻了源码才明白是 LeftOuter 的 Build Side 限制在作祟。第二个坑是阈值调太猛。有次我为了追求 MapJoin把一个公司内部自以为“不大”的宽表直接广播那张表实际 20GB但统计信息显示 1GB。结果广播数据在网络里横飞executor 内存爆掉整个调度队列都被拖垮。从那以后我对广播阈值一直很保守凡是超过 200MB 的表一定先压测试环境验证内存。第三个坑就是数据比对。有一回我把 LEFT JOIN 改写成 RIGHT JOIN 后测试环境跑出来的结果行数少了几万行。排查到最后发现是 ON 条件里有一个给大表的过滤我没有同步迁移导致部分原本会保留的维表行因为右表匹配不上而被当作 NULL 丢掉了。这个案例我记到现在。最后再补一句个人经验我现在的习惯是SQL Review 里只要看到LEFT JOIN右侧是超大表就会立刻警觉。如果左侧是小表不用犹豫直接反向写成RIGHT JOIN让广播侧落在右侧小表上。这是一条非常省力又稳定的优化路径。另一个小技巧是永远不要只依赖执行时间来判断优化效果。打开 Spark UI看一眼 Shuffle Read 字节数和 Physical Plan 里的BuildRight/BuildLeft字样这才是真正能证明 MapJoin 生效的硬证据。调优虽然是门体力活但只要方向对了几十行 SQL 也能带来小时级的收益提升。