8 月 25 日，字節(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é)一起參與共建！
項(xiàng)目地址：
https://github.com/bytedance/CloudShuffleService
開(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ì)額外消耗讀寫(xiě) 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)磁盤(pán)壞了也會(huì)導(dǎo)致數(shù)據(jù)丟失，同樣引起 FetchFailed 問(wèn)題；
-
Shuffle Data File 寫(xiě)到本地磁盤(pán)的方式，依賴計(jì)算節(jié)點(diǎn)上的磁盤(pá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)，在在離線混部 (在線資源磁盤(pán)不足)/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 磁盤(pán) IO 能力 / 磁盤(pán)壞等情況，導(dǎo)致大量的 Shuffle FetchFailed 引起的作業(yè)慢 / 失敗 /Stage 重算等問(wèn)題，影響穩(wěn)定性 & 資源利用率
-
External Shuffle Service (以下簡(jiǎn)稱 ESS) 存算無(wú)法分離，遇到磁盤(pán)容量低的機(jī)器經(jīng)常出現(xiàn)磁盤(pá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ù)寫(xiě)到同一個(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 類(lèi)似，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)題 / 磁盤(pá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ī)器 (如磁盤(pán)能力不同)/ 負(fù)載 等維度的 CSS Worker 分配策略。
