字節(jié)跳動開源自研 Shuffle 框架——Cloud Shuffle Service

　　今天，字節(jié)跳動宣布，正式開源 Cloud Shuffle Service。

　　Cloud Shuffle Service(以下簡稱CSS) 是字節(jié)自研的通用 Remote Shuffle Service 框架，支持 Spark/FlinkBatch/MapReduce 等計算引擎，提供了相比原生方案穩(wěn)定性更好、性能更高、更彈性的數(shù)據(jù) Shuffle 能力，同時也為存算分離/在離線混部等場景提供了 Remote Shuffle 解決方案。

　　目前，CSS 已在 Github 上開源，歡迎感興趣的同學(xué)一起參與共建!

　　開源背景

　　在大數(shù)據(jù)計算引擎中，Pull-Based Sort Shuffle 是一種常見的 Shuffle 方案，比如 Spark/MapReduce/FlinkBatch (高于1.15版本)等都將 Sort Shuffle 作為引擎默認(rèn)方案，但是 Sort Shuffle 實現(xiàn)機制有一定的缺陷，在大規(guī)模生產(chǎn)環(huán)境下經(jīng)常因為 Shuffle 問題影響作業(yè)穩(wěn)定性。

　　以 Spark 的 Sort Shuffle 為例：

　　如上圖所示鏈路，Sort Shuffle 會存在以下一些問題：

　　將多個 Spill 文件合并成一個文件，會額外消耗讀寫 IO;

　　假設(shè)有 m 個 MapTask & n 個 ReduceTask，會產(chǎn)生 m*n 個網(wǎng)絡(luò)鏈接，當(dāng)數(shù)量特別多時：

　　大量的網(wǎng)絡(luò)請求會導(dǎo)致 Shuffle Service 容易形成積壓;

　　Shuffle Service 會產(chǎn)生大量的隨機讀取，容易導(dǎo)致 IO 瓶頸，特別是 HDD 集群;

　　Shuffle Service 無法做到 Application 的資源隔離，當(dāng)有一個異常作業(yè)時，可能會影響同一個 Shuffle Service 節(jié)點上其它所有作業(yè)，問題容易放大;

　　MapTask 生成的 Shuffle Data File 只存儲一份到本地，當(dāng)磁盤壞了也會導(dǎo)致數(shù)據(jù)丟失，同樣引起 FetchFailed 問題;

　　Shuffle Data File 寫到本地磁盤的方式，依賴計算節(jié)點上的磁盤，無法做到存算分離

　　這些都很容易導(dǎo)致 ShuffleRead 慢或者超時，引起 FetchFailed 相關(guān)錯誤，嚴(yán)重影響線上作業(yè)的穩(wěn)定性，ShuffleRead 慢也會大大降低資源利用率(CPU&Memory)，同時 FetchFailed 也會導(dǎo)致 Stage 中相關(guān) Task 重算，浪費大量資源，拖慢整個集群作業(yè)運行;無法存算分離的架構(gòu)，在在離線混部(在線資源磁盤不足)/Serverless 云原生等場景下，也很難滿足要求。字節(jié)跳動使用 Spark 作為主要的離線大數(shù)據(jù)處理引擎，每天線上運行作業(yè)數(shù)過百萬，日均 Shuffle 量 300+PB。在 HDFS 混部&在離線混部等場景，Spark 作業(yè)的穩(wěn)定性經(jīng)常無法得到保障，影響業(yè)務(wù) SLA：

　　受限 HDD 磁盤 IO 能力/磁盤壞等情況，導(dǎo)致大量的 Shuffle FetchFailed 引起的作業(yè)慢/失敗/Stage 重算等問題，影響穩(wěn)定性&資源利用率

　　External Shuffle Service (以下簡稱ESS)  存算無法分離，遇到磁盤容量低的機器經(jīng)常出現(xiàn)磁盤打滿影響作業(yè)運行

　　在此背景下，字節(jié)跳動自研了 CSS，用來解決 Spark 原生 ESS 方案的痛點問題。自 CSS 在內(nèi)部上線一年半以來，當(dāng)前線上節(jié)點數(shù) 1500+，日均 Shuffle 量 20+PB，大大提高了 Spark 作業(yè)的 Shuffle 穩(wěn)定性，保障了業(yè)務(wù)的 SLA。

　　Cloud Shuffle Service 介紹

　　CSS 是字節(jié)自研的 Push-Based Shuffle Service，所有 MapTask 通過 Push 的方式將同一個 Partition 的 Shuffle 數(shù)據(jù)發(fā)送給同一個 CSS Worker 節(jié)點進(jìn)行存儲，ReduceTask 直接從該節(jié)點通過 CSS Worker 順序讀取該 Partition 的數(shù)據(jù)，相對于 ESS 的隨機讀取，順序讀的 IO 效率大大提升。

　　CSS 架構(gòu)

　　Cloud Shuffle Service(CSS) 架構(gòu)圖CSS Cluster 是獨立部署的 Shuffle Service 服務(wù)，主要涉及的組件為：

　　CSS Worker

　　CSS Worker 啟動后會向 ZooKeeper 節(jié)點注冊節(jié)點信息，它提供 Push/Fetch 兩種服務(wù)請求，Push 服務(wù)接受來自 MapTask 的 Push 數(shù)據(jù)請求，并將同一個 Partition 的數(shù)據(jù)寫到同一個文件;Fetch 服務(wù)接受來自 ReduceTask 的 Fetch 數(shù)據(jù)請求，讀取對應(yīng) Partition 數(shù)據(jù)文件返回;CSS Worker還負(fù)責(zé) Shuffle 數(shù)據(jù)清理的工作，當(dāng) Driver 進(jìn)行 UnregisterShuffle 請求刪除 ZooKeeper 對應(yīng) ShuffleId 的 Znode 時，或者 Application 結(jié)束刪除 ZooKeeper 中 ApplicationId 的 Znode 時，CSS Workers 會 Watch 相關(guān)事件對 Shuffle 數(shù)據(jù)進(jìn)行清理。

　　CSS Master

　　作業(yè)啟動后會在 Spark Driver 中啟動 CSS Master，CSS Master 會從 ZooKeeper 中獲取到 CSS Worker 的節(jié)點列表，然后為后續(xù) MapTask 產(chǎn)生的各個 Partition 分配 n 個副本(默認(rèn)為2)的 CSS Worker 節(jié)點，并對這些 Meta 信息進(jìn)行管理，供 ReduceTask 獲取 PartitionId 所在的 CSS Worker 節(jié)點進(jìn)行拉取，同時在 RegisterShuffle/UnregisterShuffle 過程中會在 ZooKeeper 中創(chuàng)建對應(yīng)的 ApplicationId/ShuffleId 的 Znode，CSS Worker 會 Watch Delete 事件對 Shuffle 數(shù)據(jù)進(jìn)行清理。

　　ZooKeeper

　　如前描述，用來存儲 CSS Worker 節(jié)點信息以及 ShuffleId 等信息。

　　CSS 特性

　　多引擎支持

　　CSS除了支持 Spark(2.x&3.x) 之外，也可以接入其他引擎，目前在字節(jié)跳動內(nèi)部，CSS 還接入了 MapReduce/FlinkBatch 引擎。

　　PartitionGroup 支持

　　為了解決單個 Partition 太小，Push 效率比較低的問題，實際會將多個連續(xù)的 Partition 組合成更大的 PartitionGroup進(jìn)行 Push。

　　高效統(tǒng)一的內(nèi)存管理

　　跟 ESS 類似，MapTask 中的 CSS Buffer 將所有 Partition 的數(shù)據(jù)都存儲在一起，在 Spill 之前會對數(shù)據(jù)按照 PartitionId 進(jìn)行排序，然后按照 PartitionGroup 維度進(jìn)行數(shù)據(jù)推送;同時 CSS Buffer 完全納入 Spark 的 UnifiedMemoryManager 內(nèi)存管理體系，內(nèi)存相關(guān)參數(shù)由 Spark 統(tǒng)一管理。

　　容錯處理

　　Push 失敗：當(dāng)觸發(fā) Spill 進(jìn)行 Push PartitionGroup 數(shù)據(jù)時，每次 Push 的數(shù)據(jù)大小為 4MB(一個Batch)，當(dāng)某次 Push batch 失敗時，并不影響之前已經(jīng) Push 成功的數(shù)據(jù)，只需要重新分配節(jié)點(Reallocate)繼續(xù) Push 當(dāng)前失敗的數(shù)據(jù)以及后續(xù)還未 Push 的數(shù)據(jù)，后續(xù) ReduceTask 會從新老節(jié)點讀取完整的 Partition 數(shù)據(jù);多副本存儲：ReduceTask 從 CSS Worker 讀取某個 Partition 數(shù)據(jù)是按照 Batch 粒度進(jìn)行拉取的，當(dāng) CSS Worker 異常(如網(wǎng)絡(luò)問題/磁盤壞等)導(dǎo)致無法獲取該 Batch 數(shù)據(jù)，可以繼續(xù)選擇另外一個副本節(jié)點繼續(xù)讀取該 Batch 以及后續(xù) Batch 的數(shù)據(jù);數(shù)據(jù)去重：當(dāng)作業(yè)開啟 Speculative 推測執(zhí)行會有多個 AttempTask 并發(fā)跑，需要在讀取的時候進(jìn)行去重。在 Push Batch 的時候，會給 Batch 數(shù)據(jù)加上 Header 信息，Header 信息中包含  MapId + AttempId + BatchId 等信息，ReduceTask 讀取時可以根據(jù)這些 ID 信息進(jìn)行去重。

　　Adaptive Query Execution(AQE) 適配

　　CSS 完整支持 AQE 相關(guān)的功能，包括動態(tài)調(diào)整 Reduce 個數(shù)/ SkewJoin 優(yōu)化/Join 策略優(yōu)化。對于SkewJoin，CSS做了更多的適配優(yōu)化工作，解決了 Skew Partition 數(shù)據(jù)被多個 ReduceTask 重復(fù)讀取問題，大大提高了性能。

　　CSS 性能測試

　　我們將 CSS 與開源的 ESS 使用獨占 Label 計算資源進(jìn)行 1TB 的 TPC-DS Benchmark 測試對比，整體端到端的性能提升15%左右，部分 Query 有30%以上的性能提升。同時我們也使用線上混部資源隊列(ESS 穩(wěn)定較差)進(jìn)行 1TB 的 TPC-DS Benchmark 測試對比，整體端到端性能提升4倍左右。

　　CSS 1TB 測試提升 30% 以上的 Query

　　未來規(guī)劃

　　CSS 目前開源了部分 Feature，還有一些 Feature & 優(yōu)化后續(xù)會陸續(xù)開放：

　　支持 MapReduce/FlinkBatch 引擎;

　　CSS 集群增加 ClusterManager 服務(wù)角色，管理 CSS Worker 的狀態(tài)&負(fù)載信息，同時將當(dāng)前 CSS Master 分配 CSS Worker 的功能提到 ClusterManager;

　　基于異構(gòu)機器(如磁盤能力不同)/負(fù)載 等維度的 CSS Worker 分配策略。