題幹與適用場景
一個 Spark SQL 每日作業把 4.8 TB 事件事實表與 180 GB 商品維度表依 product_id 做左外連接。 上游發布後，32% 的事件被統一映射為 product_id='UNKNOWN'。作業設定了 2,000 個 shuffle 分割區；Spark UI 顯示該連接 Stage 的 task shuffle read 中位數為 1.1 GiB，卻有一個 task 讀取 720 GiB，持續發生磁碟 spill，並在重試後 OOM。作業時間從 24 分鐘增加到 96 分鐘，業務要求 45 分鐘內完成，同時不能丟棄未知商品事件或改變左外連接結果。
資料表大小、key 占比、分割區指標、時間與 SLA 都是面試假設。題目的核心是用分割區層級證據區分 資料傾斜、資源不足與連接結果膨脹，再依資料語意選擇修復。它歸入 data，因為主要考察 Spark 執行計畫、shuffle 分割區、資料分布與批次處理正確性。現有 Kafka 熱分割區題關注訊息分割鍵、 順序與消費位點；本題關注 SQL 執行期分割區、join 策略、AQE 與結果守恆，故障層和驗證方法都不同。
這類題也適用於 groupBy、distinct、視窗函式或其他寬相依：只要同一 key 的大量資料列在 shuffle 後集中到少數 task，就可能形成長尾、spill、GC 壓力或 OOM。答案不能從「加記憶體」開始， 因為必須先證明最慢 task 是否承擔了不成比例的資料與計算。
面試官考察點
先看證據粒度。強回答會從 Job 進入 SQL 查詢與具體 Stage，再比較每個 task 的 duration、 shuffle read records/bytes、spill、peak execution memory 與 GC time。一個 task 比中位數讀取幾百倍 資料，並且重試到其他 executor 後仍處理同一個巨量分割區，才支持確定性的資料傾斜。只看 executor 總記憶體或作業總時間無法完成歸因。
執行計畫理解同樣關鍵。EXPLAIN FORMATTED 用來確認 Exchange、連接類型與實體 join； EXPLAIN COST 與 SQL UI 的執行期統計用來檢查估算和實際資料量。Spark 的 AQE 會利用執行期統計 調整計畫。Spark 4.2.0 文件中，skew join 最佳化會拆分傾斜的 sort-merge join 分割區，並在需要時 複製較小一側。預設判定同時要求分割區大於中位數的 5 倍和 256 MiB；實際回答應檢查執行環境的有效 設定，不能把文件預設值當成所有叢集的固定事實。
資料語意是分水嶺。UNKNOWN 可能代表合法的「無法歸因」事件，也可能由上游錯誤填入。 刪除、隨機分散或改寫這些資料列都會影響結果。加鹽 join 必須只複製維度表中的熱點 key，並替事實表 熱點資料列分配確定性 salt；替所有維度表資料列做笛卡兒複製會放大資料量，兩側獨立隨機加鹽則會漏掉連接。
方案選擇也能呈現回答深度。增加 spark.sql.shuffle.partitions 會產生更多 hash bucket，但同一個熱點 key 仍進入同一個 bucket。廣播只適合經過欄位裁剪、過濾與統計後確實能安全放入每個 executor 的一側； 180 GB 維度表不應靠 hint 強制廣播。AQE 是低侵入首選，明確加鹽適用於熱點 key 穩定、自動最佳化 未生效或仍無法達到 SLA 的情況；聚合類作業則常用熱點 key 分片預先聚合後再合併。
最後要看驗證完整性。效能修復必須同時證明資料列數、業務金額、未知 key、未匹配率與重複率 沒有變化。只展示作業從 96 分鐘降到 40 分鐘，仍無法證明結果正確，也無法解釋下一次 key 分布變化時 方案是否繼續有效。
回答前需要釐清的問題
- 慢點位於 scan、shuffle write 還是 shuffle read 之後？ scan task 不均勻可能來自超大檔案或不可切分輸入；本題的異常位於 join 後的 shuffle read，才進入 key 傾斜路徑。
- 32% 指資料列數、壓縮位元組還是處理成本？ 寬資料列、複雜 UDF 或結果扇出會造成成本傾斜。需要同時比較資料列、位元組、時間與輸出資料列數。
UNKNOWN的業務語意是什麼？ 若未知商品不需要維度屬性，可以把它們從 join 主路徑分離後依原語意補齊空值；若必須匹配維度表中的 sentinel 資料列，則應保留連接並拆分該熱點。- 商品維度表在
product_id上是否唯一？ 若UNKNOWN在維度表中有多筆，事實表熱點與維度表重複會形成多對多結果膨脹。修復前必須區分分割區傾斜與錯誤基數。 - 目前 Spark 版本和 AQE 有效設定是什麼？ 需要從 Environment 頁面和最終 adaptive plan 核對開關、門檻、join 類型與是否真的執行 skew split。
- 180 GB 是原始大小還是裁剪後的連接輸入？ 如果只保留連接鍵和兩個屬性後變成可安全廣播的小表，廣播可能優於雙邊 shuffle；必須用執行期統計和 executor 記憶體預算證明。
- 結果的守恆條件有哪些？ 至少明確定義總資料列數、唯一事件數、金額等可加總指標、未知 key 資料列數、未匹配資料列數和允許的重複語意。
- 熱點 key 是否穩定且可列舉？ 穩定的少數熱點適合定向加鹽；每天變化的長尾分布更適合 AQE、動態熱點偵測或上游語意修復。
30 秒回答框架
「我先在 SQL UI 比較 join task 的 shuffle read、spill、GC 和重試位置。720 GiB 對 1.1 GiB 中位數， 加上 UNKNOWN 占 32%，支持 key 傾斜；我還會檢查維度表唯一性，排除結果膨脹。先驗證 AQE skew join 是否拆分大分割區；仍超過 45 分鐘時，用穩定 event_id 對 UNKNOWN 定向加鹽，只複製 sentinel 資料列。最後對照基線核對資料列數與金額，再比較最大/中位 task 輸入、Stage 時間和成本。」
分步深入解答
第一步：把 96 分鐘拆到具體 Stage 和 task
先保留 event log，使用 Spark History Server 查看同一資料日期的正常版和異常版。沿 SQL 查詢進入 Stage 詳情，記錄 task duration 的分位數和最大值，並對齊 shuffle read records/bytes、shuffle spill、peak execution memory、GC time、失敗原因和 executor。SQL 計畫中確認左外連接前是否有 Exchange hashpartitioning(product_id, 2000)，最終實體運算子是否為 sort-merge join，以及 adaptive plan 是否完成。
本題中，同一個 task 在不同 executor 上重試仍讀取約 720 GiB，而大多數 task 約 1.1 GiB。這項證據 指向輸入分割區本身。若慢 task 讀取量普通、只在一台 executor 上出現高 GC 或磁碟等待，應先排查 節點；若所有 task 都均勻 spill，重點是整體分割區大小與資源預算；若輸出資料列數突然倍增，先檢查 維度表重複與連接條件。
第二步：用 key 分布與連接基數證明原因
在與正式環境相同的過濾和 key 正規化之後統計熱點。不能只查原始欄位，因為 trim、大小寫轉換、 coalesce 或 UDF 可能把多個值合併成同一個 key。大表上應利用既有統計、抽樣或受控聚合，避免為 診斷再製造一個無邊界作業。下面的查詢假設事實表已有或預先計算 payload_bytes，用來同時檢查 資料列與位元組傾斜；沒有該欄位時，應從儲存統計或受控序列化估算位元組。查詢表達需要核對的口徑：
SELECT
COALESCE(product_id, '<NULL>') AS join_key,
COUNT(*) AS row_count,
SUM(payload_bytes) AS payload_bytes
FROM fact_events
WHERE event_date = DATE '2026-07-17'
GROUP BY COALESCE(product_id, '<NULL>')
ORDER BY row_count DESC
LIMIT 20;同時驗證維度表中每個 product_id 最多一筆，並比較 join 前後的事實事件數。對本題的左外連接而言， 維度表 key 唯一時，每個事實事件應恰好輸出一筆；未匹配事件仍會保留。若 UNKNOWN 有 32% 的事實 資料列且維度表只有一筆 sentinel 資料，輸出沒有倍增，720 GiB 的單一分割區就能由 hash 分割規則解釋。
第三步：先處理語意根因，再選執行層手段
檢查上游發布為何把 32% 映射成 UNKNOWN。若它是回歸，應回復或修正映射，並重跑受影響分割區； 這會同時恢復資料品質與效能。若它是合法業務值，執行層仍要承載該分布。
合法但不需要商品屬性的未知事件可以分流：冷 key 正常連接，未知 key 依原契約直接補齊維度空值， 最後 unionByName。這能消除沒有資訊增益的 shuffle。若業務要求 UNKNOWN 匹配 sentinel 屬性， 則保留 join，使用 AQE 或定向加鹽拆分工作。選擇由結果語意決定，不能為了均勻分割區隨意刪除資料列。
第四步：依成本由低到高選擇修復
先核對 AQE。Spark 4.2.0 中 spark.sql.adaptive.enabled 和 spark.sql.adaptive.skewJoin.enabled 預設開啟，但叢集、作業或平台可能覆寫。skew join 判定的倍數 門檻與絕對位元組門檻必須同時滿足。查看最終 adaptive plan 是否出現 skew 處理，並確認 Stage 指標 確實拆開大分割區。AQE 對符合條件的 sort-merge join 可拆分大側、複製較小側；它能跟隨每天變動的 分布，但可能增加 shuffle 與複製成本，也不能修復錯誤的多對多連接。
如果裁剪後的維度表有可靠統計且足夠小，可以評估 broadcast hash join，以免事實表依 join key shuffle。原始 180 GB 明顯不符合一般廣播預算；強制 hint 可能讓每個 executor 記憶體不足。需要用 投影後的位元組數、並行 task、executor heap 與廣播逾時共同判斷。
AQE 未生效或仍未達標時，對穩定熱點做手動加鹽。下面的程式碼假設 event_id 穩定且唯一、維度表 product_id 唯一，並且只有 UNKNOWN 需要拆分。事實表熱點資料列被確定性分到 32 個 salt；維度表 只把對應的一筆熱點資料複製 32 份。冷 key 保持 salt 0，因此沒有全表複製。
from pyspark.sql import functions as F
SALT_BUCKETS = 32
HOT_KEYS = ["UNKNOWN"]
events_salted = events.withColumn(
"salt",
F.when(
F.col("product_id").isin(*HOT_KEYS),
F.pmod(F.xxhash64("event_id"), F.lit(SALT_BUCKETS)).cast("int"),
).otherwise(F.lit(0)),
)
salt_values = spark.range(SALT_BUCKETS).select(
F.col("id").cast("int").alias("salt")
)
products_hot = (
products.filter(F.col("product_id").isin(*HOT_KEYS))
.crossJoin(salt_values)
)
products_cold = (
products.filter(~F.col("product_id").isin(*HOT_KEYS))
.withColumn("salt", F.lit(0))
)
products_salted = products_cold.unionByName(products_hot)
result = (
events_salted.join(products_salted, ["product_id", "salt"], "left")
.drop("salt")
)32 只是面試假設下的起始候選。bucket 數應由熱點分割區位元組、目標 task 大小、可用平行度與小側 複製成本推導，再用代表性資料測試。bucket 過少仍有長尾，過多則會增加排程、檔案與複製開銷。 單純執行 repartition(4000, "product_id") 仍把 UNKNOWN 放入一個分割區，無法取得相同效果。
對於 groupBy(productid)，通常不需要複製維度表：先依 (productid, salt) 做區域聚合，再依 product_id 合併部分結果。可結合交換律與結合律安全拆分的 sum、count、min、max 較適合； 精確 median、有順序要求的聚合或不可結合 UDF 需要不同演算法，不能機械套用。
第五步：把正確性、效能與成本放進同一套驗收
在固定輸入快照上平行執行基線與候選方案。正確性優先：比較總輸出資料列數、唯一 event_id、 UNKNOWN 資料列數、未匹配資料列數、依業務維度聚合的金額與計數，並抽取熱點、冷 key、空值和 重複維度 key 樣本做逐列 diff。左外連接每個事實事件輸出一筆的前提是維度表 key 唯一，這項約束 要單獨告警。
效能方面比較 join Stage 的 p50、p95 與最大 task duration，最大/中位 shuffle read，再比對 spill、 GC、OOM、task retry、Stage 時間與總作業時間。成本方面記錄 executor-hours、shuffle bytes 與輸出 檔案數。達到 45 分鐘只是一項門檻；若時間達標但 shuffle 加倍、結果不守恆或換一天資料又出現新熱點， 方案仍不合格。
發布時先對一個歷史日期重播，再對新資料日期做 shadow 執行並比對結果。熱點 key 集合、最大/中位 task 輸入比和未知 key 占比應進入監控。上游發布後 UNKNOWN 從基線上升到 32% 的變化需要資料品質 告警，讓語意回歸在拖慢作業前被發現。
高品質示範回答
「我會先把問題定位到 SQL 查詢中的具體 join Stage。Spark UI 已提供很強的傾斜訊號：2,000 個 task 的 shuffle read 中位數為 1.1 GiB，單一 task 是 720 GiB，而且重試到其他 executor 後仍處理同一個 大分割區。接下來我會查看最終 adaptive plan、task spill 和 GC，再依連接前的正規化邏輯統計 key 分布。維度表也要做唯一性檢查；如果 UNKNOWN 有多筆維度資料，目前症狀還包含 join 結果膨脹。
假設維度表 key 唯一、UNKNOWN 占事實表 32%，那麼同 key 經 hash shuffle 集中到一個分割區， 足以解釋長尾。增加分割區數只會增加其他 hash bucket，無法拆開這個 key；提高 executor 記憶體最多 延後 OOM。我的第一個動作是檢查上游映射是否回歸。如果 UNKNOWN 不需要維度屬性，我會把它從 join 主路徑分離，並依原左外連接語意補空值。若它必須匹配 sentinel 資料，我會先確認 AQE 的 skew join 在最終計畫中實際生效，因為它可以依執行期統計拆分傾斜 sort-merge join 分割區並複製小側。
如果 AQE 後仍超過 45 分鐘，我會做定向加鹽。用穩定 event_id 把 UNKNOWN 事實資料列確定性分到 例如 32 個 salt，只複製維度表的 UNKNOWN 資料列 32 份；其他 key 的 salt 都是 0。這樣能維持每個 事件只匹配一筆維度資料，同時把熱點工作拆給多個 task。bucket 數由熱點位元組與目標 task 大小推導， 不會固定寫死。
驗證時我會在同一輸入上比較基線：總資料列數、唯一事件數、未知與未匹配資料、金額聚合都必須一致； 接著看最大/中位 shuffle read、task 長尾、spill、OOM、Stage 時間、executor-hours 和輸出檔案數。 最後用一個歷史日期重播和一個新日期 shadow 執行，並對 UNKNOWN 占比和新熱點告警。這樣能同時證明 作業在 45 分鐘內完成、結果未變，而且上游分布再次改變時可以及時發現。」
常見錯誤
- 直接把 shuffle 分割區從 2,000 調到 8,000 → 同一個熱點 key 仍雜湊到一個分割區，其他 task 反而更碎 → 先確認 key 分布，再用 AQE、語意分流或熱點加鹽拆開該 key。
- 只提高 executor memory → 它擴大單一 task 的容忍量，卻沒有減少 720 GiB 輸入和長尾 → 先降低最大分割區工作量，再依測得的 task 大小設定資源。
- 看到一個慢 task 就宣布資料傾斜 → 節點故障、GC、遠端 fetch 或慢 UDF 也會造成長尾 → 比較 task 輸入、重試位置、spill、GC 與執行計畫。
- 替事實表和維度表各自隨機加鹽 → 兩側 salt 對不上會漏連接，重複嘗試還可能造成不確定結果 → 事實側使用穩定資料列 ID，維度表側列舉同一組 salt。
- 把整張維度表複製 32 份 → 180 GB 維度表會帶來巨量網路與記憶體成本 → 只複製已確認的熱點 key 資料列，冷 key 保持 salt 0。
- 強制廣播 180 GB 維度表 → 每個 executor 都要承擔廣播資料，可能直接 OOM → 先裁剪與統計，只有在記憶體和並行預算證明安全後才廣播。
- 過濾
UNKNOWN讓作業變快 → 輸出語意與下游指標會改變 → 先確認未知事件的業務契約，分流時也維持左外連接結果。 - 只比較總作業時間 → 效能改善可能來自少算、重複或遺失資料 → 先做資料列和業務守恆，再比較 task 分布、成本與 SLA。
追問及應對
AQE 已開啟，為什麼傾斜 join 仍沒有被拆分？
先看最終 adaptive plan 與有效設定，確認 spark.sql.adaptive.enabled、skew join 開關、倍數門檻與 絕對位元組門檻。兩個門檻需要同時命中。再確認實體 join 是否屬於 AQE 支援的路徑、執行期統計是否 可用，以及計畫是否被 hint 或平台設定限制。可以在代表性資料上調整門檻或評估強制 skew 最佳化， 同時量化新增 shuffle；若計畫結構無法受益，再使用定向加鹽。只看到設定檔中的 true 不能證明 執行計畫已經拆分。
如果裁剪後的維度表只有 6 GiB，可以直接廣播嗎？
6 GiB 仍需結合 executor heap、並行 task、序列化大小、廣播逾時和叢集穩定性判斷。廣播可以消除 大側依 join key 的 shuffle，因此可能自然避開該熱點；它也會把維度表複製到 executor。先用統計證明 實際廣播位元組，在小規模與正式環境形態的壓力測試中觀察 peak memory 和 GC，再決定是否使用 hint。 不能只根據「比事實表小很多」得出可廣播結論。
如果熱點每天變化，手動維護 HOT_KEYS 怎麼辦？
優先讓 AQE 依執行期統計處理。若仍需明確加鹽，可以在主作業前從受控統計中產生達到資料列數、位元組 或成本門檻的熱點 key 小表，並廣播該小表決定是否加鹽；熱點清單要有資料日期、門檻、數量上限和 回復策略。動態方案增加一個規劃階段和維運狀態，因此要用穩定的 SLA 收益證明它值得存在。
若慢操作是 groupBy，還需要複製維度表嗎？
不需要。對可結合聚合，先替熱點資料列加 salt，依 (key, salt) 做第一階段部分聚合，再依 key 合併。這會把一個熱點 key 的輸入分給多個 task，第二階段只處理少量部分結果。回答時應說明聚合函式 是否滿足結合要求；依賴全域順序或無法合併的狀態不能直接採用該方法。
如何選擇 32 個 salt bucket？
先用熱點分割區位元組除以目標 task 輸入量得到下限，再受可用核心數、小側複製成本、排程開銷與 輸出檔案約束。假設 720 GiB 希望降到約 32 GiB，理論下限約為 23，可以從 32 開始測試；這只是 本題推導示例。最終應比較多個候選的最大 task 輸入、Stage 時間與 executor-hours，並替持續成長的 熱點保留有限餘量。
開啟 speculative execution 能解決這個長尾嗎？
確定性傾斜 task 的副本仍讀取同一個 720 GiB 分割區，通常只會讓兩台 executor 重複昂貴工作。推測 執行較適合偶發慢節點或不可重現的抖動。先看同一 task 重試到其他 executor 後是否仍慢；若輸入量 始終異常，就應拆分資料工作，不能只複製該工作。