字節(jié)跳動(dòng)開(kāi)源自研 Shuffle 框架——Cloud Shuffle Service

　　今天，字節(jié)跳動(dòng)宣布，正式開(kāi)源 Cloud Shuffle Service。

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

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

　　開(kāi)源背景

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

　　以 Spark 的 Sort Shuffle 為例：

　　如上圖所示鏈路，Sort Shuffle 會(huì)存在以下一些問(wèn)題：

　　將多個(gè) Spill 文件合并成一個(gè)文件，會(huì)額外消耗讀寫 IO;

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

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

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

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

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

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

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

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

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

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

　　Cloud Shuffle Service 介紹

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

　　CSS 架構(gòu)

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

　　CSS Worker

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

　　CSS Master

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

　　ZooKeeper

　　如前描述，用來(lái)存儲(chǔ) CSS Worker 節(jié)點(diǎn)信息以及 ShuffleId 等信息。

　　CSS 特性

　　多引擎支持

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

　　PartitionGroup 支持

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

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

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

　　容錯(cuò)處理

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

　　Adaptive Query Execution(AQE) 適配

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

　　CSS 性能測(cè)試

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

　　CSS 1TB 測(cè)試提升 30% 以上的 Query

　　未來(lái)規(guī)劃

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

　　支持 MapReduce/FlinkBatch 引擎;

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

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