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

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

項(xiàng)目地址：

https://github.com/bytedance/CloudShuffleService

開源背景

在大數(shù)據(jù)計(jì)算引擎中，Pull-Based Sort Shuffle 是一種常見的 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 問題影響作業(yè)穩(wěn)定性。

以 Spark 的 Sort Shuffle 為例：

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

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

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

• 大量的網(wǎng)絡(luò)請求會導(dǎo)致 Shuffle Service 容易形成積壓；
• Shuffle Service 會產(chǎn)生大量的隨機(jī)讀取，容易導(dǎo)致 IO 瓶頸，特別是 HDD 集群；
• Shuffle Service 無法做到 Application 的資源隔離，當(dāng)有一個(gè)異常作業(yè)時(shí)，可能會影響同一個(gè) Shuffle Service 節(jié)點(diǎn)上其它所有作業(yè)，問題容易放大；
• MapTask 生成的 Shuffle Data File 只存儲一份到本地，當(dāng)磁盤壞了也會導(dǎo)致數(shù)據(jù)丟失，同樣引起 FetchFailed 問題；
• Shuffle Data File 寫到本地磁盤的方式，依賴計(jì)算節(jié)點(diǎn)上的磁盤，無法做到存算分離

這些都很容易導(dǎo)致 ShuffleRead 慢或者超時(shí)，引起 FetchFailed 相關(guān)錯(cuò)誤，嚴(yán)重影響線上作業(yè)的穩(wěn)定性，ShuffleRead 慢也會大大降低資源利用率(CPU&Memory)，同時(shí) FetchFailed 也會導(dǎo)致 Stage 中相關(guān) Task 重算，浪費(fèi)大量資源，拖慢整個(gè)集群作業(yè)運(yùn)行；無法存算分離的架構(gòu)，在在離線混部(在線資源磁盤不足)/Serverless 云原生等場景下，也很難滿足要求。字節(jié)跳動(dòng)使用 Spark 作為主要的離線大數(shù)據(jù)處理引擎，每天線上運(yùn)行作業(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ī)器經(jīng)常出現(xiàn)磁盤打滿影響作業(yè)運(yùn)行

在此背景下，字節(jié)跳動(dòng)自研了 CSS，用來解決 Spark 原生 ESS 方案的痛點(diǎn)問題。自 CSS 在內(nè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 通過 Push 的方式將同一個(gè) Partition 的 Shuffle 數(shù)據(jù)發(fā)送給同一個(gè) CSS Worker 節(jié)點(diǎn)進(jìn)行存儲，ReduceTask 直接從該節(jié)點(diǎn)通過 CSS Worker 順序讀取該 Partition 的數(shù)據(jù)，相對于 ESS 的隨機(jī)讀取，順序讀的 IO 效率大大提升。

CSS 架構(gòu)

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

CSS Worker

CSS Worker 啟動(dòng)后會向 ZooKeeper 節(jié)點(diǎn)注冊節(jié)點(diǎn)信息，它提供 Push/Fetch 兩種服務(wù)請求，Push 服務(wù)接受來自 MapTask 的 Push 數(shù)據(jù)請求，并將同一個(gè) Partition 的數(shù)據(jù)寫到同一個(gè)文件；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 時(shí)，或者 Application 結(jié)束刪除 ZooKeeper 中 ApplicationId 的 Znode 時(shí)，CSS Workers 會 Watch 相關(guān)事件對 Shuffle 數(shù)據(jù)進(jìn)行清理。

CSS Master

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

ZooKeeper

如前描述，用來存儲 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 效率比較低的問題，實(shí)際會將多個(gè)連續(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ù)推送；同時(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 會從新老節(jié)點(diǎn)讀取完整的 Partition 數(shù)據(jù)；多副本存儲：ReduceTask 從 CSS Worker 讀取某個(gè) Partition 數(shù)據(jù)是按照 Batch 粒度進(jìn)行拉取的，當(dāng) CSS Worker 異常（如網(wǎng)絡(luò)問題/磁盤壞等）導(dǎo)致無法獲取該 Batch 數(shù)據(jù)，可以繼續(xù)選擇另外一個(gè)副本節(jié)點(diǎn)繼續(xù)讀取該 Batch 以及后續(xù) Batch 的數(shù)據(jù)；數(shù)據(jù)去重：當(dāng)作業(yè)開啟 Speculative 推測執(zhí)行會有多個(gè) AttempTask 并發(fā)跑，需要在讀取的時(shí)候進(jìn)行去重。在 Push Batch 的時(shí)候，會給 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)化。對于SkewJoin，CSS做了更多的適配優(yōu)化工作，解決了 Skew Partition 數(shù)據(jù)被多個(gè) ReduceTask 重復(fù)讀取問題，大大提高了性能。

CSS 性能測試

我們將 CSS 與開源的 ESS 使用獨(dú)占 Label 計(jì)算資源進(jìn)行 1TB 的 TPC-DS Benchmark 測試對比，整體端到端的性能提升15%左右，部分 Query 有30%以上的性能提升。同時(shí)我們也使用線上混部資源隊(duì)列(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ù)載信息，同時(shí)將當(dāng)前 CSS Master 分配 CSS Worker 的功能提到 ClusterManager；
• 基于異構(gòu)機(jī)器(如磁盤能力不同)/負(fù)載 等維度的 CSS Worker 分配策略。