讀懂 Spark 執行計畫:.explain() 到底在說什麼

· tech

#spark#data-engineering#performance

這個系列一路講下來,有兩句話反覆出現:「相信 Catalyst 最佳化器」、「打開 Spark UI 找瓶頸」。但我一直沒回答一個問題:你到底要怎麼看 Spark 做了什麼? 相信最佳化器不該是盲信——你得有辦法驗證「我以為會 broadcast 的 join,它真的 broadcast 了嗎?我的 filter 真的下推了嗎?」這篇就補上這塊技能:讀懂執行計畫。

一行 code 到執行:中間發生什麼

你寫的 DataFrame 或 SQL,不會直接跑。它會先變成一份「要什麼」的邏輯計畫,經 Catalyst 改寫,再落成一份「怎麼做」的實體計畫,最後才切成 task 去跑:

你的 code DataFrame API 或 Spark SQL —— 兩者等價 Logical Plan ·「要什麼」 解析出你引用的表與欄位,還沒最佳化 Catalyst 最佳化器 filter 下推 · 剪掉沒用的欄位 · 挑 join 策略 Physical Plan ·「怎麼做」 Exchange(= shuffle)、join 策略都定案 執行 切成 stage / task,丟到 executor 上跑
你寫的每段 Spark code 都走這條路。DataFrame 和 SQL 在第一步就殊途同歸,所以它們效能一樣——這也是 DataFrame 那篇 那張「編成同一個計畫」的圖的下半場。.explain() 就是把這條路印出來給你看

.explain():把計畫印出來

在任何 DataFrame 後面接 .explain(),就能看到它的實體計畫——不用真的跑,是靜態分析:

df = (big.filter(F.col("date") >= "2026-01-01")
         .join(F.broadcast(dim), on="id")
         .groupBy("cat").count())

df.explain()             # 只看 physical plan(最常用)
df.explain(True)         # 連 logical → optimized → physical 全印
df.explain("formatted")  # 分段、附欄位細節,最好讀

關鍵讀法只有一句:由下往上讀。 最底下是資料來源(掃描),往上一層層是 filter、join、shuffle、聚合,最上面才是最終結果。資料是從下往上流的。

讀 physical plan 要盯的三個字

計畫裡的字很多,但九成的效能判斷,只看三個關鍵字:

HashAggregate ← groupBy 的結果 Exchange hashpartitioning(cat) BroadcastHashJoin [id] Filter date ≥ 2026-01-01 FileScan parquet · big PushedFilters:[date ≥ …] FileScan dim → Broadcast 小表 Exchange = shuffle! 整個 query 最貴的一步 廣播小表,大表原地 join ✓ 沒有搬大表,省掉一次 shuffle filter 下推到掃描層 ✓ Parquet 只讀符合的資料,少讀一堆 由下往上讀 ↑
同一個 query 的實體計畫。由下往上:掃描→過濾→join→shuffle→聚合。盯三個字就夠——Exchange 是 shuffle、BroadcastHashJoin 是好 join、PushedFilters 是 filter 有沒有下推

實際印出來大概長這樣(explain("formatted") 精簡版):

*(3) HashAggregate(keys=[cat], functions=[count(1)])
+- Exchange hashpartitioning(cat, 200)                    ← shuffle 在這
   +- *(2) HashAggregate(keys=[cat], functions=[partial_count(1)])
      +- *(2) BroadcastHashJoin [id], [id], Inner, BuildRight   ← 廣播 join,沒搬大表 ✓
         :- *(2) Filter (date >= 2026-01-01)
         :  +- *(1) FileScan parquet big[id,cat,date]
         :        PushedFilters: [GreaterThanOrEqual(date, 2026-01-01)]  ← filter 下推 ✓
         +- BroadcastExchange
            +- *(1) FileScan parquet dim[id]

三個字的意義,直接對應你前面學過的東西:

  • Exchange = 一次 shuffle。這是你在計畫裡最該找的字——數一數有幾個 Exchange,大概就知道這個 query 貴在哪。 groupBy、非廣播的 join、distinct、repartition 都會生出它。
  • BroadcastHashJoin vs SortMergeJoin:前者是廣播 join,小表複製到各節點、大表原地做,沒有 shuffle;後者兩邊都要依 key shuffle。看到你「應該要 broadcast」的 join 卻顯示 SortMergeJoin,就是小表超過門檻沒被廣播——這是最常見的效能漏抓。
  • PushedFilters:你的 filter 有沒有被下推到掃描層。有的話,Parquet 這種欄式格式在讀檔時就跳過不符的資料,根本不讀進來;沒有的話,就是全讀進來才過濾,白白多搬一堆。

一個陷阱:AQE 會在執行期改計畫

.explain() 印的是執行前的計畫。但 Spark 3 預設開的 AQE(Adaptive Query Execution)會在跑的時候依實際資料量再改一次——合併過小的分區、把原本的 SortMergeJoin 換成 broadcast、拆掉傾斜的分區。所以 .explain() 看到的不一定是最後真正跑的。要看實際執行的最終計畫,得去 Spark UI 的 SQL / Query 頁,那裡的計畫圖會標出 AQE 調整後的樣子,還附上每個節點實際處理了多少列、shuffle 讀寫多少。靜態的 .explain() 看結構,動態的 Spark UI 看事實——兩個一起用。

反思

「相信最佳化器」不等於「不看它做了什麼」

我在 DataFrame 那篇說「把最佳化交給比你聰明的 Catalyst」,這話沒錯,但很容易被誤讀成「反正它會處理,我不用管」。學會讀 .explain() 之後我的心態精準多了:我信任 Catalyst,但我會驗證。 我以為 filter 會下推——打開計畫看有沒有 PushedFilters;我以為那個 join 會廣播——看它是不是 BroadcastHashJoin。八成的時候它做對了,但真正拖慢的,往往就是那兩成「我以為它會、但它沒有」的地方。信任而不盲信,靠的就是有能力打開黑箱看一眼。

看到 Exchange 就該心頭一緊

整個 shuffle 那篇在講 shuffle 多貴,而在計畫裡,shuffle 就叫 Exchange。這個對應關係一旦建立,讀計畫就變得很有方向感:我不再逐行細看,而是先掃有幾個 Exchange、分別是哪個操作造成的,再問「這一次搬移,省得掉嗎?」——能不能先 filter 變小、能不能 broadcast 掉、能不能少一次 groupBy。把抽象的「效能問題」變成具體的「數 Exchange」,是我讀計畫最大的收穫。

explain 是「假設」,Spark UI 是「事實」

這是我踩過坑才學到的:光看 .explain() 會被靜態計畫騙。它不知道你的資料實際上有沒有傾斜、某個 key 是不是佔了九成的量;這些只有真的跑起來、在 Spark UI 裡看每個 task 的耗時分佈才看得到。所以我現在的順序是——先用 .explain() 確認結構對不對(join 策略、filter 下推、有幾個 shuffle),再跑一次、用 Spark UI 驗證實況(哪個 stage 卡住、task 是不是有一根特別長)。 這跟我對所有工具的態度一致:先看清痛點在哪,再動手,而讀懂計畫,就是讓「痛點」從模糊的感覺變成螢幕上一個看得到、指得出來的節點。