字節跳動在Spark SQL上的核心優化實踐 | 字節跳動技術沙龍

時間 2019-11-17

標籤字節跳動 spark sql 核心優化實踐技術沙龍欄目 Spark 简体版

原文原文鏈接

10月26日， 字節跳動技術沙龍 | 大數據架構專場 在上海字節跳動總部圓滿結束。咱們邀請到字節跳動數據倉庫架構負責人-郭俊，Kyligence 大數據研發工程師-陶加濤，字節跳動存儲工程師-徐明敏，阿里雲高級技術專家-白宸和你們進行分享交流。

如下是字節跳動數據倉庫架構負責人-郭俊的分享主題沉澱：《字節跳動在Spark SQL上的核心優化實踐》。前端

團隊介紹

數據倉庫架構團隊負責數據倉庫領域架構設計，支持字節跳動幾乎全部產品線（包含但不限於抖音、今日頭條、西瓜視頻、火山視頻）數據倉庫方向的需求，如 Spark SQL / Druid 的二次開發和優化。

歸納

今天的分享分爲三個部分，第一個部分是 SparkSQL 的架構簡介，第二部分介紹字節跳動在 SparkSQL 引擎上的優化實踐，第三部分是字節跳動在 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 數相同。app

以下圖所示，通過 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 的透明遷移。oop

第一個須要解決的問題是，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 獲取屬於本身的那部分數據。

這個架構存在兩個問題：

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

性能問題：每一個 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 穩定性問題。

個人分享就到這裏，謝謝你們。

QA集錦

- 提問：物化列新增一列，是否須要修改歷史數據？

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

- 提問：若是用戶的請求同時包含新數據和歷史數據，如何處理？

回答：通常而言，用戶修改數據都是以 Partition 爲單位。因此咱們在 Partition Parameter 上保存了物化列相關信息。若是用戶的查詢同時包含了新 Partition 與歷史 Partition，咱們會在新 Partition 上針對物化列進行 SQL Rewrite，歷史 Partition 不 Rewrite，而後將新老 Partition 進行 Union，從而在保證數據正確性的前提下儘量充分利用物化列的優點。

- 提問：你好，大家針對用戶的場景，作了不少挺有價值的優化。像物化列、物化視圖，都須要根據用戶的查詢 Pattern 進行設置。目前大家是人工分析這些查詢，仍是有某種機制自動去分析並優化？

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

- 提問：剛剛介紹的基於 HDFS 的 Spark Shuffle 穩定性提高方案，是否能夠異步上傳 Shuffle 數據至 HDFS？

回答：這個想法挺好，咱們以前也考慮過，但基於幾點考慮，最終沒有這樣作。第一，單 Mapper 的 Shuffle 輸出數據量通常很小，上傳到 HDFS 耗時在 2 秒之內，這個時間開銷能夠忽略；第二，咱們普遍使用 External Shuffle Service 和 Dynamic Allocation，Mapper 執行完成後可能 Executor 就回收了，若是要異步上傳，就必須依賴其它組件，這會提高複雜度，ROI 較低。