上個(gè)月排查一個(gè)跑批任務(wù)的時(shí)候我碰見了一條看起來再普通不過的SQL左側(cè)是300萬行的用戶維表右側(cè)是8億行的訂單事實(shí)表寫法是dim_user u LEFT JOIN fact_order o ON u.user_id o.buyer_id。任務(wù)跑了兩個(gè)多小時(shí)還拖著隊(duì)列打開執(zhí)行計(jì)劃一看走的是 SortMergeJoin兩個(gè) Exchange 在瘋狂 shuffle。按理說用戶維表才幾十MBSpark 完全可以用 MAPJOINBroadcastHashJoin把維表廣播到每個(gè)節(jié)點(diǎn)為什么連資格都沒撈到這個(gè)問題折騰了我一個(gè)下午才徹底想明白。它不涉及什么冷門配置也不是數(shù)據(jù)傾斜純粹是“LEFT JOIN 的方向”和 Spark 廣播連接機(jī)制的底層約束撞在一起了。今天就把整個(gè)排查過程、原理和幾種解法完整寫出來希望能讓大家少走我走過的彎路。內(nèi)容適合正在做數(shù)據(jù)開發(fā)、數(shù)倉建模以及日常要寫 Spark SQL 調(diào)優(yōu)的工程師尤其是對(duì)廣播連接了解得比較模糊、只知道“小表要廣播”的人。1. MapJoin為什么快以及這里為什么沒走M(jìn)apJoin1.1 一條SQL希望變成的執(zhí)行計(jì)劃MapJoin 這個(gè)叫法是從 Hive 時(shí)代留下來的Spark 里的正式名字是 BroadcastHashJoin也就是廣播連接。它的核心思想很簡(jiǎn)單不把兩張表都拖到 shuffle 里重新分區(qū)而是把里面比較小的那張表完整廣播到每一個(gè) executor 上另一張大表留在本地逐行去和這份“廣播小表”做哈希匹配。你可以把它想象成考試時(shí)每個(gè)考場(chǎng)都發(fā)了一份公式手冊(cè)考生不需要為了查一條公式專門跑到學(xué)校圖書館去。大表就是考生手里的試卷小表就是那本公式手冊(cè)。如果走 SortMergeJoin相當(dāng)于每道題都要先把試卷和圖書館里的書按頁碼重新分類匯總再回到考場(chǎng)上答題。誰快誰慢一眼就能看出來。所以理想情況下這條 SQL 的計(jì)劃應(yīng)該是BroadcastHashJoin :- BroadcastExchange把dim_user發(fā)到所有executor : - FileScan dim_user - FileScan fact_order大表沒有 Exchange沒有 shuffle只有一份幾十MB的廣播數(shù)據(jù)在網(wǎng)絡(luò)里分發(fā)。這也是我們做調(diào)優(yōu)時(shí)最想看到的物理計(jì)劃。1.2 Build Side和Stream Side到底是誰很多初學(xué)者會(huì)問Spark 是怎么決定廣播哪一張表的這里有兩個(gè)概念必須先搞清楚。廣播的那一側(cè)叫 Build Side也就是用來構(gòu)建哈希表的那一側(cè)。逐行掃描的另一側(cè)叫 Stream Side。對(duì)于等值連接來說Spark 并不是隨便選一側(cè)廣播的它在源碼里對(duì)不同的 join type 有硬性約束。簡(jiǎn)單歸納下來是這樣Join類型BroadcastHashJoin 的 Build Side 限制Inner Join左右兩側(cè)都可以作為 Build Side優(yōu)化器傾向于選統(tǒng)計(jì)值更小的一側(cè)Left Outer Join只能把右表作為 Build SideRight Outer Join左右兩側(cè)都可以作為 Build Side優(yōu)化器傾向于選更小的一側(cè)Full Outer Join不能使用 BroadcastHashJoin這里最關(guān)鍵的是 Left Outer Join。為什么 LEFT JOIN 不能廣播左表因?yàn)?LEFT JOIN 的語義要求左表的每一行都必須保留在最終結(jié)果里包括那些在右表中找不到匹配的行。如果 Spark 把左表廣播出去、讓右表逐行去探測(cè)那么每個(gè)分區(qū)只能看到右表的一部分誰來決定“左表這一行在全量右表里完全沒匹配”這個(gè)判斷在分布式環(huán)境下會(huì)變得非常麻煩。所以 Spark 干脆從規(guī)則層面不允許 LeftOuter 把左表作為 Build Side?；氐轿覀兡菞lSQLSELECT 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 只能把右表當(dāng)作 Build Side?？墒怯冶硎?8 億行的大表怎么可能廣播于是 Spark 只好退回到 SortMergeJoin兩張表全部 shuffle。小表雖然小但是因?yàn)樗驹诹恕安荒軓V播”的位置上整個(gè) MapJoin 優(yōu)化直接失效。1.3 閾值與統(tǒng)計(jì)信息Spark如何判斷“可以廣播”除了 join type 的方向限制Spark 在是否能廣播這件事上還有一道門檻就是spark.sql.autoBroadcastJoinThreshold默認(rèn)值是 10MB。也就是說那張表在邏輯計(jì)劃里的估算大小如果小于 10MBSpark 才會(huì)考慮廣播。注意這里說的“表大小”不是我們體感上的大小而是 Spark 從三方面拿到的估算值Hive Metastore 里的統(tǒng)計(jì)信息、文件系統(tǒng)的文件大小、以及ANALYZE TABLE手動(dòng)收集的統(tǒng)計(jì)信息。如果這些信息缺失或者過期Spark 對(duì)表大小的判斷就會(huì)走樣進(jìn)而影響它是否選擇廣播。所以“小表 LEFT JOIN 大表”失效的本質(zhì)不是小表不夠小而是兩個(gè)約束疊在一起LeftOuter 不允許廣播左表而右表又太大沒有資格廣播。明白了這個(gè)根因后面的解決方案就有了方向。2. 最優(yōu)解把LEFT JOIN改寫成RIGHT JOIN讓廣播側(cè)落在小表2.1 改寫原理既然 LEFT JOIN 只能把右表當(dāng)作 Build Side那我們能不能把“小表”挪到右表去答案是可以但要通過改寫 join 的方向來實(shí)現(xiàn)。A LEFT JOIN B在語義上等價(jià)于B RIGHT JOIN A。前者保留 A 的全部行后者保留 A 的全部行兩者只是把表的位置換了一下。原SQL里小表 dim_user 在左、大表 fact_order 在右改成fact_order RIGHT JOIN dim_user之后小表跑到了右表位置而 RightOuter 是允許右表作為 Build Side 的。這樣一來Spark 會(huì)審視右表 dim_user 的大小。如果它滿足廣播閾值優(yōu)化器就會(huì)選擇 BuildRight生成 BroadcastHashJoin大表不再 shuffle。這是我在實(shí)際調(diào)優(yōu)中用得最多、效果也最直接的方案。2.2 具體SQL改造示例改造后的 SQL 長(zhǎng)這樣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保持一致業(yè)務(wù)代碼基本不用改。如果 dim_user 的實(shí)際大小超過了默認(rèn)的 10MB 閾值就再用廣播提示給它加一道保險(xiǎn)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;執(zhí)行計(jì)劃會(huì)變成類似這樣BroadcastHashJoin [buyer_id#12], [user_id#5], RightOuter, BuildRight :- BroadcastExchange HashedRelation : - FileScan dim_user - FileScan fact_order看到RightOuter, BuildRight基本就放心了。這次調(diào)優(yōu)的效果很明顯同一個(gè)任務(wù)從 2 小時(shí) 40 分降到了 18 分鐘而且集群的 shuffle 量從幾百GB驟降到幾十MB。不是玄學(xué)就是少做了一次大表全量 shuffle。2.3 改寫時(shí)容易踩的坑改寫 SQL 看起來只是換了表的位置但實(shí)際操作里有幾個(gè)坑我一個(gè)個(gè)列出來。第一個(gè)坑是列順序。如果上游消費(fèi)結(jié)果集的時(shí)候是按下標(biāo)取值改了 join 方向后必須保證 SELECT 字段順序和原來完全一致。我在測(cè)試環(huán)境見過有人把u.user_id和o.order_id的順序?qū)懛唇Y(jié)果下游數(shù)據(jù)對(duì)不上排查了半天。第二個(gè)坑是過濾條件的位置。LEFT JOIN 改寫為 RIGHT JOIN 后ON 子句里的條件要保持等價(jià)。如果在原SQL里寫的是dim_user u LEFT JOIN fact_order o ON u.user_id o.buyer_id AND o.amount 100那改寫后同樣要放在 ON 子句里。如果把它移到 WHERE 子句語義就變了WHERE 里過濾大表字段會(huì)把 NULL 行也過濾掉結(jié)果集少掉一批本來該保留的用戶。第三個(gè)坑是主表的唯一性。dim_user 作為主表如果它里面有重復(fù)的 user_idLEFT JOIN 之后這些重復(fù)行會(huì)各自匹配一次訂單改寫后 RIGHT JOIN 也一樣行數(shù)不會(huì)變化。但如果右表有重復(fù) keyLEFT JOIN 會(huì)產(chǎn)生笛卡爾式放大這種放大和 join 方向無關(guān)問題可能出在數(shù)據(jù)質(zhì)量上別誤判成改寫造成的。第四個(gè)坑是大表列里的 NULL。LEFT JOIN 之后未匹配的用戶對(duì)應(yīng)的訂單字段本來就是 NULL這是業(yè)務(wù)上要接受的現(xiàn)實(shí)。有些同事看到 NULL 就以為是 bug實(shí)際上只要大表字段本身也有 NULL 值你根本無法只憑最終結(jié)果判斷到底是“未匹配”還是“本來就是 NULL”。這個(gè)在改寫前就應(yīng)和業(yè)務(wù)方確認(rèn)清楚。3. 兜底方案Hint、閾值調(diào)參與統(tǒng)計(jì)信息修正3.1 Broadcast Hint的正確打開方式如果因?yàn)槟承┰虿荒芨?SQL 方向還可以用 Hint 去影響優(yōu)化器。Spark 支持BROADCAST和MAPJOIN兩種寫法SELECT /* MAPJOIN(dim_user) */ ... SELECT /* BROADCAST(dim_user) */ ...這里必須強(qiáng)調(diào)一個(gè)容易誤導(dǎo)人的點(diǎn)在小表 LEFT JOIN 大表的原始 SQL 上加/* BROADCAST(dim_user) */是沒用的。因?yàn)?LeftOuter 這個(gè) join type 從源碼規(guī)則上就不允許左表作 Build SideHint 只能告訴優(yōu)化器“這張表適合廣播”不能突破 join type 的硬約束。Spark 看到這個(gè) Hint 后如果發(fā)現(xiàn)放不下就會(huì)安靜地忽略它繼續(xù)走 SortMergeJoin。所以在 RIGHT JOIN 改寫版本的 SQL 上加 Hint效果才是穩(wěn)定可期的。正確姿勢(shì)是把 Hint 加在已經(jīng)在右表位置的維表名上讓 BuildRight 的意圖更加明確。3.2 閾值怎么調(diào)spark.sql.autoBroadcastJoinThreshold默認(rèn) 10MB實(shí)際生產(chǎn)環(huán)境里很多“小表”遠(yuǎn)不止 10MB。比如我們這條SQL里的 dim_user 是 60MB那就得手動(dòng)調(diào)大閾值spark.sql.autoBroadcastJoinThreshold100m spark.sql.broadcastTimeout600這里要特別提醒調(diào)閾值不是調(diào)得越大越好。廣播表是要完整塞進(jìn)每個(gè) executor 內(nèi)存的一張 60MB 的表序列化之后加上構(gòu)建哈希表、GC 開銷實(shí)際占用可能是 300MB 甚至更多。如果集群上每個(gè) executor 只有 2GB 內(nèi)存而你把閾值調(diào)到 500MB一旦廣播表真的超過預(yù)期輕則拖慢整個(gè)任務(wù)重則直接 OOM。我自己的經(jīng)驗(yàn)是三個(gè)原則廣播表大小別超過單個(gè) executor 可用內(nèi)存的 20%廣播數(shù)據(jù)的總量要低于 DM 網(wǎng)卡和隊(duì)列帶寬的承受范圍優(yōu)先用 Hint 而不是全局閾值因?yàn)?Hint 是局部影響全局閾值會(huì)無差別地作用于所有 join風(fēng)險(xiǎn)面太大。另外如果你們用的是 Spark 3.2 以上版本還有一個(gè)自適應(yīng)參數(shù)spark.sql.adaptive.autoBroadcastJoinThreshold值得關(guān)注。AQE 會(huì)在 shuffle 執(zhí)行過程中根據(jù)實(shí)際讀到的數(shù)據(jù)量重新判斷是否有機(jī)會(huì)轉(zhuǎn)換為廣播連接。有時(shí)候一開始沒進(jìn) MapJoinAQE 結(jié)束后反而能救回來。但它和spark.sql.adaptive.coalescePartitions.enabled的聯(lián)動(dòng)比較微妙建議在測(cè)試環(huán)境先壓一輪再上線。3.3 用EXPLAIN驗(yàn)證是否真的MapJoin調(diào)優(yōu)最怕的是一頓操作猛如虎計(jì)劃還是原樣。所以每次改完 SQL我都會(huì)先把執(zhí)行計(jì)劃打出來看一眼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物理計(jì)劃里會(huì)有BroadcastHashJoin和BroadcastExchange節(jié)點(diǎn)而且大表一側(cè)沒有Exchange。如果走 SortMergeJoin你會(huì)看到左右兩側(cè)各有一個(gè)Exchange那就說明廣播優(yōu)化沒生效。我還習(xí)慣在 Spark UI 的 SQL 頁簽里看 Shuffle 字節(jié)數(shù)。MapJoin 生效時(shí)Shuffle Read/Write 的數(shù)據(jù)量會(huì)斷崖式下降這個(gè)數(shù)字比執(zhí)行計(jì)劃更直觀也更容易說服團(tuán)隊(duì)里其他同事。4. 當(dāng)小表不適合廣播時(shí)分步JOIN與其它手段4.1 先裁剪大表再做外連接不是所有“小表”都適合廣播。如果維表有 2GB強(qiáng)行廣播會(huì)導(dǎo)致后來所有并行 task 都背著這個(gè)大包袱反而比 shuffle 更慢。這種情況下第一條策略是對(duì)大表做裁剪。對(duì)大事實(shí)表來說最常見的裁剪方式是時(shí)間分區(qū)。原SQL如果寫成FROM fact_order o JOIN dim_user u ON u.user_id o.buyer_id WHERE o.dt 2024-01-01那么這個(gè)過濾后的右表可能只有幾十GB依然很大但有時(shí)已經(jīng)足夠小。實(shí)際操作里我會(huì)先看過濾后的右表到底多大。如果能在幾百M(fèi)B以內(nèi)就值得嘗試廣播如果過濾后仍然幾個(gè)GB那還是別廣播了。更進(jìn)一步的裁剪方式是“用維表 key 去圈定大表的數(shù)據(jù)范圍”。雖然純 SQL 里IN (SELECT ... FROM dim_user)的執(zhí)行效率通常不高但在某些數(shù)倉分層場(chǎng)景下可以提前把事實(shí)表的 buyer_id 字典集合落成一個(gè)中間表再做裁剪。這個(gè)思路的本質(zhì)就是先把大表“瘦身”再讓外連接的上游輸入變小。4.2 內(nèi)連接加補(bǔ)NULL的拆分思路如果 SQL 結(jié)構(gòu)很復(fù)雜不得不維持原來的 LEFT JOIN還有一種拆方案的思路把整個(gè)匹配邏輯拆成“匹配部分 未匹配補(bǔ)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優(yōu)化器允許把小表當(dāng) Build Side即使它在左表位置也能廣播。未匹配部分則用 anti join 找出維表里沒有訂單的用戶再把大表字段置成 NULL最后兩者 UNION ALL。這種寫法確實(shí)能把廣播用起來但它的代價(jià)也很明顯要對(duì)大表掃描兩次一次是內(nèi)連接一次是 anti join。除非大表能提前被過濾得很小否則整體耗時(shí)不一定比重寫為 RIGHT JOIN 更短。所以我的建議是它只適合那些“SQL 由上游工具自動(dòng)生成、無法手動(dòng)改表順序”的僵化場(chǎng)景手工可改的情況下優(yōu)先用第 2 節(jié)的重寫方案。4.3 ShuffleHashJoin/SMB等替代方案廣播連接不行的時(shí)候也別死磕 MapJoin。還有兩個(gè)方向可以考慮。第一是 ShuffleHashJoin。通過 Hint 強(qiáng)制SELECT /* SHUFFLE_HASH(u) */ ...它會(huì)把兩張表都 shuffle但選擇小表作為構(gòu)建哈希表的一側(cè)避免 SortMergeJoin 中額外的排序和歸并。在 join key 本身比較均勻、沒有明顯傾斜的時(shí)候ShuffleHashJoin 比 SortMergeJoin 快不少。第二是分桶表 SMB Join。如果事實(shí)表和維表都預(yù)先把 join key 分成同樣數(shù)量的桶而且兩邊桶的哈希規(guī)則一致那么 join 時(shí)兩個(gè)表可以按桶直接讀取不需要 shuffle。這個(gè)方案對(duì) ETL 建模要求比較高適合高頻運(yùn)行的固定 join一次性建模成本攤薄到多次任務(wù)后很劃算。5. 常見問題與排查經(jīng)驗(yàn)實(shí)錄5.1 排查步驟清單如果你現(xiàn)在也在處理類似“小表 LEFT JOIN 大表”導(dǎo)致變慢的問題可以按下面這套順序來排查。第一步用EXPLAIN先看物理計(jì)劃里是BroadcastHashJoin還是SortMergeJoin。這是最直接的證據(jù)。第二步確認(rèn) join type。是 Inner、LeftOuter 還是 RightOuter。LeftOuter 的話先鎖定右表是否能廣播。第三步check 左右表的統(tǒng)計(jì)信息。ANALYZE TABLE dim_user COMPUTE STATISTICS能解決大部分“明明很小卻不廣播”的問題。第四步看廣播表估算大小和 executor 內(nèi)存。在 Spark UI 的 Executors 頁簽里看每個(gè) executor 實(shí)際內(nèi)存使用心里有數(shù)再調(diào)閾值。第五步改完 SQL 之后做數(shù)據(jù)比對(duì)。用 COUNT 和抽樣字段的 MD5 校驗(yàn)改寫前后結(jié)果是否一致尤其是 NULL 分布和重復(fù)行。5.2 高頻問題速查表現(xiàn)象可能原因處理方式廣播等待時(shí)間過長(zhǎng)廣播數(shù)據(jù)量太大超過spark.sql.broadcastTimeout默認(rèn) 300 秒調(diào)大 timeout或放棄廣播改用 ShuffleHashJoinExecutor OOM廣播表太大或者 executors 內(nèi)存不足降低廣播閾值給小表瘦身增大 executor memory overhead明明很小卻沒走 BroadcastHashJoin統(tǒng)計(jì)信息缺失或過期執(zhí)行ANALYZE TABLE刷新統(tǒng)計(jì)信息加BROADCAST(hint)沒生效LEFT JOIN 中試圖廣播左表違反 join type 約束改成 RIGHT JOIN把目標(biāo)表放到右表位置改寫 RIGHT JOIN 后結(jié)果和原來不一致列順序變了、ON/WHERE 條件位置不對(duì)、主表重復(fù) key逐項(xiàng)對(duì)比 SELECT 列、ON 條件、重復(fù)行數(shù)廣播后任務(wù)變慢廣播傳輸消耗大或者廣播表在多個(gè) stage 反復(fù)序列化使用 ReusedExchange或評(píng)估是否該走 shuffle5.3 我踩過的幾個(gè)坑第一次踩這個(gè)坑時(shí)我犯的錯(cuò)誤就是只加 Hint 不改方向。在原 SQL 上寫了/* MAPJOIN(u) */滿懷期待地等執(zhí)行計(jì)劃結(jié)果 Spark 依然走 SortMergeJoin。當(dāng)時(shí)一度懷疑是 Hint 拼寫錯(cuò)了后來翻了源碼才明白是 LeftOuter 的 Build Side 限制在作祟。第二個(gè)坑是閾值調(diào)太猛。有次我為了追求 MapJoin把一個(gè)公司內(nèi)部自以為“不大”的寬表直接廣播那張表實(shí)際 20GB但統(tǒng)計(jì)信息顯示 1GB。結(jié)果廣播數(shù)據(jù)在網(wǎng)絡(luò)里橫飛executor 內(nèi)存爆掉整個(gè)調(diào)度隊(duì)列都被拖垮。從那以后我對(duì)廣播閾值一直很保守凡是超過 200MB 的表一定先壓測(cè)試環(huán)境驗(yàn)證內(nèi)存。第三個(gè)坑就是數(shù)據(jù)比對(duì)。有一回我把 LEFT JOIN 改寫成 RIGHT JOIN 后測(cè)試環(huán)境跑出來的結(jié)果行數(shù)少了幾萬行。排查到最后發(fā)現(xiàn)是 ON 條件里有一個(gè)給大表的過濾我沒有同步遷移導(dǎo)致部分原本會(huì)保留的維表行因?yàn)橛冶砥ヅ洳簧隙划?dāng)作 NULL 丟掉了。這個(gè)案例我記到現(xiàn)在。最后再補(bǔ)一句個(gè)人經(jīng)驗(yàn)我現(xiàn)在的習(xí)慣是SQL Review 里只要看到LEFT JOIN右側(cè)是超大表就會(huì)立刻警覺。如果左側(cè)是小表不用猶豫直接反向?qū)懗蒖IGHT JOIN讓廣播側(cè)落在右側(cè)小表上。這是一條非常省力又穩(wěn)定的優(yōu)化路徑。另一個(gè)小技巧是永遠(yuǎn)不要只依賴執(zhí)行時(shí)間來判斷優(yōu)化效果。打開 Spark UI看一眼 Shuffle Read 字節(jié)數(shù)和 Physical Plan 里的BuildRight/BuildLeft字樣這才是真正能證明 MapJoin 生效的硬證據(jù)。調(diào)優(yōu)雖然是門體力活但只要方向?qū)α藥资?SQL 也能帶來小時(shí)級(jí)的收益提升。