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