字節跳動在Spark SQL上的核心優化實踐

本文作者是字節跳動數據倉庫架構負責人，數據倉庫架構團隊負責數據倉庫領域架構設計，支持字節跳動幾乎所有產品線（包含但不限於

概括

今天的分享分為三個部分，第一個部分是 Spark SQL 的架構簡介，第二部分介紹字節跳動在 Spark SQL 引擎上的優化實踐，第三部分是字節跳動在 Spark Shuffle 穩定性提升和性能優化上的實踐與探索。

Spark SQL架構簡介

我們先簡單聊一下Spark SQL 的架構。下面這張圖描述了一條 SQL 提交之後需要經歷的幾個階段，結合這些階段就可以看到在哪些環節可以做優化。

很多時候，做數據倉庫建模的同學更傾向於直接寫 SQL 而非使用 Spark 的 DSL。一條 SQL 提交之後會被 Parser 解析並轉化為 Unresolved Logical Plan。它的重點是 Logical Plan 也即邏輯計劃，它描述了希望做什麼樣的查詢。Unresolved 是指該查詢相關的一些信息未知，比如不知道查詢的目標表的 Schema 以及數據位置。

上述信息存於 Catalog 內。在生產環境中，一般由 Hive Metastore 提供 Catalog 服務。Analyzer 會結合 Catalog 將 Unresolved Logical Plan 轉換為 Resolved Logical Plan。

到這裡還不夠。不同的人寫出來的SQL 不一樣，生成的 Resolved Logical Plan 也就不一樣，執行效率也不一樣。為了保證無論用戶如何寫 SQL 都可以高效的執行，Spark SQL 需要對 Resolved Logical Plan 進行優化，這個優化由 Optimizer 完成。Optimizer 包含了一系列規則，對 Resolved Logical Plan 進行等價轉換，最終生成 Optimized Logical Plan。該 Optimized Logical Plan 不能保證是全局最優的，但至少是接近最優的。

上述過程只與 SQL 有關，與查詢有關，但是與 Spark 無關，因此無法直接提交給 Spark 執行。Query Planner 負責將 Optimized Logical Plan 轉換為 Physical Plan，進而可以直接由 Spark 執行。

由於同一種邏輯算子可以有多種物理實現。如 Join 有多種實現，ShuffledHashJoin、BroadcastHashJoin、BroadcastNestedLoopJoin、SortMergeJoin 等。因此 Optimized Logical Plan 可被 Query Planner 轉換為多個 Physical Plan。如何選擇最優的 Physical Plan 成為一件非常影響最終執行性能的事情。一種比較好的方式是，構建一個 Cost Model，並對所有候選的 Physical Plan 應用該 Model 並挑選 Cost 最小的 Physical Plan 作為最終的 Selected Physical Plan。

Physical Plan 可直接轉換成 RDD 由 Spark 執行。我們經常說“計劃趕不上變化”，在執行過程中，可能發現原計劃不是最優的，後續執行計劃如果能根據運行時的統計信息進行調整可能提升整體執行效率。這部分動態調整由 Adaptive Execution 完成。

後面介紹字節跳動在 Spark SQL 上做的一些優化，主要圍繞這一節介紹的邏輯計劃優化與物理計劃優化展開。

Spark SQL引擎優化

Bucket Join改進

在 Spark 裡，實際並沒有 Bucket Join 算子。這裡說的 Bucket Join 泛指不需要 Shuffle 的 SortMergeJoin。

下圖展示了 SortMergeJoin 的基本原理。用虛線框代表的 Table 1 和 Table 2 是兩張需要按某字段進行 Join 的表。虛線框內的 partition 0 到 partition m 是該錶轉換成 RDD 後的 Partition，而非表的分區。假設 Table 1 與 Table 2 轉換為 RDD 後分別包含 m 和 k 個 Partition。為了進行 Join，需要通過 Shuffle 保證相同 Join Key 的數據在同一個 Partition 內且 Partition 內按 Key 排序，同時保證 Table 1 與 Table 2 經過 Shuffle 後的 RDD 的 Partition 數相同。

如下圖所示，經過 Shuffle 後只需要啟動 n 個 Task，每個 Task 處理 Table 1 與 Table 2 中對應 Partition 的數據進行 Join 即可。如 Task 0 只需要順序掃描 Shuffle 後的左右兩邊的 partition 0 即可完成 Join。

該方法的優勢是適用場景廣，幾乎可用於任意大小的數據集。劣勢是每次 Join 都需要對全量數據進行 Shuffle，而 Shuffle 是最影響 Spark SQL 性能的環節。如果能避免 Shuffle 往往能大幅提升 Spark SQL 性能。

對於大數據的場景來講，數據一般是一次寫入多次查詢。如果經常對兩張表按相同或類似的方式進行 Join，每次都需要付出 Shuffle 的代價。與其這樣，不如讓數據在寫的時候，就讓數據按照利於 Join 的方式分佈，從而使得 Join 時無需進行 Shuffle。如下圖所示，Table 1 與 Table 2 內的數據按照相同的 Key 進行分桶且桶數都為 n，同時桶內按該 Key 排序。對這兩張表進行 Join 時，可以避免 Shuffle，直接啟動 n 個 Task 進行 Join。

字節跳動對 Spark SQL 的 BucketJoin 做了四項比較大的改進。

改進一：支持與Hive兼容

在過去一段時間，字節跳動把大量的 Hive 作業遷移到了 SparkSQL。而 Hive 與 Spark SQL 的 Bucket 表不兼容。對於使用 Bucket 表的場景，如果直接更新計算引擎，會造成 Spark SQL 寫入 Hive Bucket 表的數據無法被下游的 Hive 作業當成 Bucket 表進行 Bucket Join，從而造成作業執行時間變長，可能影響 SLA。

為了解決這個問題，我們讓 Spark SQL 支持 Hive 兼容模式，從而保證 Spark SQL 寫入的 Bucket 表與 Hive 寫入的 Bucket 表效果一致，並且這種表可以被 Hive 和 Spark SQL 當成 Bucket 表進行 Bucket Join 而不需要 Shuffle。通過這種方式保證 Hive 向 Spark SQL 的透明遷移。

第一個需要解決的問題是，Hive 的一個 Bucket 一般只包含一個文件，而 Spark SQL 的一個 Bucket 可能包含多個文件。解決辦法是動態增加一次以 Bucket Key 為 Key 並且並行度與 Bucket 個數相同的 Shuffle。

第二個需要解決的問題是，Hive 1.x 的哈希方式與 Spark SQL 2.x 的哈希方式（Murmur3Hash）不同，使得相同的數據在 Hive 中的 Bucket ID 與 Spark SQL 中的 Bucket ID 不同而無法直接 Join。在 Hive 兼容模式下，我們讓上述動態增加的 Shuffle 使用 Hive 相同的哈希方式，從而解決該問題。

改進二：支持倍數關係Bucket Join

Spark SQL 要求只有 Bucket 相同的表才能（必要非充分條件）進行 Bucket Join。對於兩張大小相差很大的表，比如幾百 GB 的維度表與幾十 TB （單分區）的事實表，它們的 Bucket 個數往往不同，並且個數相差很多，默認無法進行 Bucket Join。因此我們通過兩種方式支持了倍數關係的 Bucket Join，即當兩張 Bucket 表的 Bucket 數是倍數關係時支持 Bucket Join。

第一種方式，Task 個數與小表 Bucket 個數相同。如下圖所示，Table A 包含 3 個 Bucket，Table B 包含 6 個 Bucket。此時 Table B 的 bucket 0 與 bucket 3 的數據合集應該與 Table A 的 bucket 0 進行 Join。這種情況下，可以啟動 3 個 Task。其中 Task 0 對 Table A 的 bucket 0 與 Table B 的 bucket 0 + bucket 3 進行 Join。在這裡，需要對 Table B 的 bucket 0 與 bucket 3 的數據再做一次 merge sort 從而保證合集有序。

如果 Table A 與 Table B 的 Bucket 個數相差不大，可以使用上述方式。如果 Table B 的 Bucket 個數是 Bucket A Bucket 個數的 10 倍，那上述方式雖然避免了 Shuffle，但可能因為並行度不夠反而比包含 Shuffle 的 SortMergeJoin 速度慢。此時可以使用另外一種方式，即 Task 個數與大表 Bucket 個數相等，如下圖所示。

在該方案下，可將 Table A 的 3 個 Bucket 讀多次。在上圖中，直接將 Table A 與 Table A 進行 Bucket Union （新的算子，與 Union 類似，但保留了 Bucket 特性），結果相當於 6 個 Bucket，與 Table B 的 Bucket 個數相同，從而可以進行 Bucket Join。

改進三：支持BucketJoin降級

公司內部過去使用 Bucket 的表較少，在我們對 Bucket 做了一系列改進後，大量用戶希望將錶轉換為 Bucket 表。轉換後，表的元信息顯示該表為 Bucket 表，而歷史分區內的數據並未按 Bucket 表要求分佈，在查詢歷史數據時會出現無法識別 Bucket 的問題。

同時，由於數據量上漲快，平均 Bucket 大小也快速增長。這會造成單 Task 需要處理的數據量過大進而引起使用 Bucket 後的效果可能不如直接使用基於 Shuffle 的 Join。

為了解決上述問題，我們實現了支持降級的 Bucket 表。基本原理是，每次修改 Bucket 信息（包含上述兩種情況——將非 Bucket 錶轉為 Bucket 表，以及修改 Bucket 個數）時，記錄修改日期。並且在決定使用哪種 Join 方式時，對於 Bucket 表先檢查所查詢的數據是否只包含該日期之後的分區。如果是，則當成 Bucket 表處理，支持 Bucket Join；否則當成普通無 Bucket 的表。

改進四：支持超集

對於一張常用表，可能會與另外一張表按 User 字段做 Join，也可能會與另外一張表按 User 和 App 字段做 Join，與其它表按 User 與 Item 字段進行 Join。而 Spark SQL 原生的 Bucket Join 要求 Join Key Set 與表的 Bucket Key Set 完全相同才能進行 Bucket Join。在該場景中，不同 Join 的 Key Set 不同，因此無法同時使用 Bucket Join。這極大的限制了 Bucket Join 的適用場景。

針對此問題，我們支持了超集場景下的 Bucket Join。只要 Join Key Set 包含了 Bucket Key Set，即可進行 Bucket Join。

如下圖所示，Table X 與 Table Y，都按字段 A 分 Bucket。而查詢需要對 Table X 與 Table Y 進行 Join，且 Join Key Set 為 A 與 B。此時，由於 A 相等的數據，在兩表中的 Bucket ID 相同，那 A 與 B 各自相等的數據在兩表中的 Bucket ID 肯定也相同，所以數據分佈是滿足 Join 要求的，不需要 Shuffle。同時，Bucket Join 還需要保證兩表按 Join Key Set 即 A 和 B 排序，此時只需要對 Table X 與 Table Y 進行分區內排序即可。由於兩邊已經按字段 A 排序了，此時再按 A 與 B 排序，代價相對較低。

物化列

Spark SQL 處理嵌套類型數據時，存在以下問題：

• 讀取大量不必要的數據：對於 Parquet / ORC 等列式存儲格式，可只讀取需要的字段，而直接跳過其它字段，從而極大節省 IO。而對於嵌套數據類型的字段，如下圖中的 Map 類型的 people 字段，往往只需要讀取其中的子字段，如 people.age。卻需要將整個 Map 類型的 people 字段全部讀取出來然後抽取出 people.age 字段。這會引入大量的無意義的 IO 開銷。在我們的場景中，存在不少 Map 類型的字段，而且很多包含幾十至幾百個 Key，這也就意味著 IO 被放大了幾十至幾百倍。

• 無法進行向量化讀取：而向量化讀能極大的提升性能。但截止到目前（2019年10月26日），Spark 不支持包含嵌套數據類型的向量化讀取。這極大的影響了包含嵌套數據類型的查詢性能

• 不支持 Filter 下推：目前（2019年10月26日）的 Spark 不支持嵌套類型字段上的 Filter 的下推

• 重複計算：JSON 字段，在 Spark SQL 中以 String 類型存在，嚴格來說不算嵌套數據類型。不過實踐中也常用於保存不固定的多個字段，在查詢時通過 JSON Path 抽取目標子字段，而大型 JSON 字符串的字段抽取非常消耗 CPU。對於熱點表，頻繁重複抽取相同子字段非常浪費資源。

對於這個問題，做數倉的同學也想了一些解決方案。如下圖所示，在名為 base_table 的表之外創建了一張名為 sub_table 的表，並且將高頻使用的子字段 people.age 設置為一個額外的 Integer 類型的字段。下游不再通過 base_table 查詢 people.age，而是使用 sub_table 上的 age 字段代替。通過這種方式，將嵌套類型字段上的查詢轉為了 Primitive 類型字段的查詢，同時解決了上述問題。

這種方案存在明顯缺陷：

• 額外維護了一張表，引入了大量的額外存儲/計算開銷。

• 無法在新表上查詢新增字段的歷史數據（如要支持對歷史數據的查詢，需要重跑歷史作業，開銷過大，無法接受）。

• 表的維護方需要在修改表結構後修改插入數據的作業。

• 需要下游查詢方修改查詢語句，推廣成本較大。

• 運營成本高：如果高頻子字段變化，需要刪除不再需要的獨立子字段，並添加新子字段為獨立字段。刪除前，需要確保下游無業務使用該字段。而新增字段需要通知並推進下游業務方使用新字段。

為解決上述所有問題，我們設計並實現了物化列。它的原理是：

• 新增一個 Primitive 類型字段，比如 Integer 類型的 age 字段，並且指定它是 people.age 的物化字段。

• 插入數據時，為物化字段自動生成數據，並在 Partition Parameter 內保存物化關係。因此對插入數據的作業完全透明，表的維護方不需要修改已有作業。

• 查詢時，檢查所需查詢的所有 Partition，如果都包含物化信息（people.age 到 age 的映射），直接將 select people.age 自動重寫為 select age，從而實現對下游查詢方的完全透明優化。同時兼容歷史數據。

下圖展示了在某張核心表上使用物化列的收益：

物化視圖

在 OLAP 領域，經常會對相同表的某些固定字段進行 Group By 和 Aggregate / Join 等耗時操作，造成大量重複性計算，浪費資源，且影響查詢性能，不利於提升用戶體驗。

我們實現了基於物化視圖的優化功能：

如上圖所示，查詢歷史顯示大量查詢根據 user 進行 group by，然後對 num 進行 sum 或 count 計算。此時可創建一張物化視圖，且對 user 進行 gorup by，對 num 進行 avg（avg 會自動轉換為 count 和 sum）。用戶對原始表進行 select user, sum(num) 查詢時，Spark SQL 自動將查詢重寫為對物化視圖的 select user, sum_num 查詢。

Spark SQL引擎上的其它優化

下圖展示了我們在 Spark SQL 上進行的其它部分優化工作：

Spark Shuffle穩定性提升與性能優化

Spark Shuffle存在的問題

Shuffle的原理，很多同學應該已經很熟悉了。鑑於時間關係，這裡不介紹過多細節，只簡單介紹下基本模型。

如上圖所示，我們將 Shuffle 上游 Stage 稱為 Mapper Stage，其中的 Task 稱為 Mapper。Shuffle 下游 Stage 稱為 Reducer Stage，其中的 Task 稱為 Reducer。

每個 Mapper 會將自己的數據分為最多 N 個部分，N 為 Reducer 個數。每個 Reducer 需要去最多 M （Mapper 個數）個 Mapper 獲取屬於自己的那部分數據。

這個架構存在兩個問題：

1）穩定性問題

Mapper 的 Shuffle Write 數據存於 Mapper 本地磁盤，只有一個副本。當該機器出現磁盤故障，或者 IO 滿載，CPU 滿載時，Reducer 無法讀取該數據，從而引起 FetchFailedException，進而導致 Stage Retry。Stage Retry 會造成作業執行時間增長，直接影響 SLA。同時，執行時間越長，出現 Shuffle 數據無法讀取的可能性越大，反過來又會造成更多 Stage Retry。如此循環，可能導致大型作業無法成功執行。

2）性能問題

每個 Mapper 的數據會被大量 Reducer 讀取，並且是隨機讀取不同部分。假設 Mapper 的 Shuffle 輸出為 512MB，Reducer 有 10 萬個，那平均每個 Reducer 讀取數據 512MB / 100000 = 5.24KB。並且，不同 Reducer 並行讀取數據。對於 Mapper 輸出文件而言，存在大量的隨機讀取。而 HDD 的隨機 IO 性能遠低於順序 IO。最終的現象是，Reducer 讀取 Shuffle 數據非常慢，反映到 Metrics 上就是 Reducer Shuffle Read Blocked Time 較長，甚至佔整個 Reducer 執行時間的一大半，如下圖所示。

基於HDFS的Shuffle穩定性提升

經觀察，引起 Shuffle 失敗的最大因素不是磁盤故障等硬件問題，而是 CPU 滿載和磁盤 IO 滿載。

如上圖所示，機器的 CPU 使用率接近 100%，使得 Mapper 側的 Node Manager 內的 Spark External Shuffle Service 無法及時提供 Shuffle 服務。

下圖中 Data Node 佔用了整臺機器 IO 資源的 84%，部分磁盤 IO 完全打滿，這使得讀取 Shuffle 數據非常慢，進而使得 Reducer 側無法在超時時間內讀取數據，造成 FetchFailedException。

無論是何種原因，問題的癥結都是 Mapper 側的 Shuffle Write 數據只保存在本地，一旦該節點出現問題，會造成該節點上所有 Shuffle Write 數據無法被 Reducer 讀取。解決這個問題的一個通用方法是，通過多副本保證可用性。

最初始的一個簡單方案是，Mapper 側最終數據文件與索引文件不寫在本地磁盤，而是直接寫到 HDFS。Reducer 不再通過 Mapper 側的 External Shuffle Service 讀取 Shuffle 數據，而是直接從 HDFS 上獲取數據，如下圖所示。

快速實現這個方案後，我們做了幾組簡單的測試。結果表明：

• Mapper 與 Reducer 不多時，Shuffle 讀寫性能與原始方案相比無差異。

• Mapper 與 Reducer 較多時，Shuffle 讀變得非常慢。

在上面的實驗過程中，HDFS 發出了報警信息。如下圖所示，HDFS Name Node Proxy 的 QPS 峰值達到 60 萬。（注：字節跳動自研了 Node Name Proxy，並在 Proxy 層實現了緩存，因此讀 QPS 可以支撐到這個量級）。

原因在於，總共 10000 Reducer，需要從 10000 個 Mapper 處讀取數據文件和索引文件，總共需要讀取 HDFS 10000 * 1000 * 2 = 2 億次。

如果只是 Name Node 的單點性能問題，還可以通過一些簡單的方法解決。例如在 Spark Driver 側保存所有 Mapper 的 Block Location，然後 Driver 將該信息廣播至所有 Executor，每個 Reducer 可以直接從 Executor 處獲取 Block Location，然後無須連接 Name Node，而是直接從 Data Node 讀取數據。但鑑於 Data Node 的線程模型，這種方案會對 Data Node 造成較大沖擊。

最後我們選擇了一種比較簡單可行的方案，如下圖所示。

Mapper 的 Shuffle 輸出數據仍然按原方案寫本地磁盤，寫完後上傳到 HDFS。Reducer 仍然按原始方案通過 Mapper 側的 External Shuffle Service 讀取 Shuffle 數據。如果失敗了，則從 HDFS 讀取。這種方案極大減少了對 HDFS 的訪問頻率。

該方案上線近一年：

• 覆蓋 57% 以上的 Spark Shuffle 數據。

• 使得 Spark 作業整體性能提升 14%。

• 天級大作業性能提升 18%。

• 小時級作業性能提升 12%。

該方案旨在提升 Spark Shuffle 穩定性從而提升作業穩定性，但最終沒有使用方差等指標來衡量穩定性的提升。原因在於每天集群負載不一樣，整體方差較大。Shuffle 穩定性提升後，Stage Retry 大幅減少，整體作業執行時間減少，也即性能提升。最終通過對比使用該方案前後的總的作業執行時間來對比性能的提升，用於衡量該方案的效果。

Shuffle性能優化實踐與探索

如上文所分析，Shuffle 性能問題的原因在於，Shuffle Write 由 Mapper 完成，然後 Reducer 需要從所有 Mapper 處讀取數據。這種模型，我們稱之為以 Mapper 為中心的 Shuffle。它的問題在於：

• Mapper 側會有 M 次順序寫 IO。

• Mapper 側會有 M * N * 2 次隨機讀 IO（這是最大的性能瓶頸）。

• Mapper 側的 External Shuffle Service 必須與 Mapper 位於同一臺機器，無法做到有效的存儲計算分離，Shuffle 服務無法獨立擴展。

針對上述問題，我們提出了以 Reducer 為中心的，存儲計算分離的 Shuffle 方案，如下圖所示。

該方案的原理是，Mapper 直接將屬於不同 Reducer 的數據寫到不同的 Shuffle Service。在上圖中，總共 2 個 Mapper，5 個 Reducer，5 個 Shuffle Service。所有 Mapper 都將屬於 Reducer 0 的數據遠程流式發送給 Shuffle Service 0，並由它順序寫入磁盤。Reducer 0 只需要從 Shuffle Service 0 順序讀取所有數據即可，無需再從 M 個 Mapper 取數據。該方案的優勢在於：

• 將 M * N * 2 次隨機 IO 變為 N 次順序 IO。

• Shuffle Service 可以獨立於 Mapper 或者 Reducer 部署，從而做到獨立擴展，做到存儲計算分離。

• Shuffle Service 可將數據直接存於 HDFS 等高可用存儲，因此可同時解決 Shuffle 穩定性問題。

>>>>

Q&A

Q1：物化列新增一列，是否需要修改歷史數據？

A：歷史數據太多，不適合修改歷史數據。

Q2：如果用戶的請求同時包含新數據和歷史數據，如何處理？

A：一般而言，用戶修改數據都是以 Partition 為單位。所以我們在 Partition Parameter 上保存了物化列相關信息。如果用戶的查詢同時包含了新 Partition 與歷史 Partition，我們會在新 Partition 上針對物化列進行 SQL Rewrite，歷史 Partition 不 Rewrite，然後將新老 Partition 進行 Union，從而在保證數據正確性的前提下儘可能充分利用物化列的優勢。

Q3：你好，你們針對用戶的場景，做了很多挺有價值的優化。像物化列、物化視圖，都需要根據用戶的查詢 Pattern 進行設置。目前你們是人工分析這些查詢，還是有某種機制自動去分析並優化？

A：目前我們主要是通過一些審計信息輔助人工分析。同時我們也正在做物化列與物化視圖的推薦服務，最終做到智能建設物化列與物化視圖。

Q4：剛剛介紹的基於 HDFS 的 Spark Shuffle 穩定性提升方案，是否可以異步上傳 Shuffle 數據至 HDFS？

A：這個想法挺好，我們之前也考慮過，但基於幾點考慮，最終沒有這樣做。第一，單 M

作者丨郭俊

來源丨字節跳動技術團隊（ID：toutiaotechblog）

dbaplus社群歡迎廣大技術人員投稿，投稿郵箱：editor@dbaplus.cn

閱讀更多 dbaplus社群 的文章

關鍵字: Spark Hive 跳動