本文整理自字節跳動基礎架構的大數據開發工程師魏中佳在 ApacheCon Aisa 2022 「大數據」議題下的演講，主要介紹 Cloud Shuffle Service（css） 在字節跳動 Spark 場景下的設計與實現。
作者｜字節跳動基礎架構 大數據研發 工程師-魏中佳
01
背景介紹
在大數據場景下，數據 Shuffle 表示了不同分區數據交換的過程，Shuffle 的性能往往會成為作業甚至整個集群的性能瓶頸。特別是在字節跳動每日上百 PB Shuffle 數據的場景下，Shuffle 過程暴露出來了很多問題，本文會逐個展開此類問題并介紹在字節跳動的優化實踐。
External Shuffle Service
首先來看，在 Spark 3.0 及最新的 Spark 3.3 中，External Shuffle Service（以下簡稱 ESS）是如何完成 Shuffle 任務的？
如下圖，每一個 Map Task，從 MApper 1 到 Mapper M 都會在本地生成屬于自己的 Shuffle 文件。這個 Shuffle 文件內部由 R 個連續的數據片段組成。每一個 Reduce Task 運行時都會分別連接所有的 Task，從 Mapper 1 一直到 Mapper M 。連接成功后，Reduce Task 會讀取每個文件中屬于自己的數據片段。
上述方式帶來的問題是顯而易見的：
-
由于每次讀取的都是這個 Shuffle 文件的 1/R，通常情況下這個數據量是非常非常小的，大概是 KB 級別（從幾百 KB 到幾 KB 不等），這樣會給磁盤（尤其是 HDD ）帶來大量隨機的讀請求。
-
同時，大家可以看到，Reduce 進行的 Shuffle Fetch 請求整體看是一個網狀結構，也就是說會存在大量的網絡請求，量級大概是 M 乘以 R，這個請求的數量級也是非常大的。
這兩個問題隨著作業規模的擴大，會帶來越來越嚴重的 Shuffle Failure 問題。Shuffle Failure 意味著超時，Shuffle Failure 本身還有可能導致 Stage 重算，甚至導致作業失敗，嚴重影響批式作業的穩定性，同時還會浪費大量的計算資源（因為 Fetch 等待超時的時候，CPU 是空閑的）。
Spark 在字節跳動的應用
在字節跳動內部，Spark 作業規模較大：
-
日均 100 萬左右個作業
-
日均 300 PB Shuffle 數據
-
大量作業簽署 SLA，對穩定性要求非常高，超時嚴重還會嚴重影響下游
-
大量 HDD 機器和少量 SSD 機器
-
大量在線業務低峰出讓的資源，可用磁盤空間非常小，需要把存儲拉遠
下圖是字節跳動內部一個 Spark 作業的 Shuffle Chunk Size 情況。這個作業只有 400 兆的 Shuffle 數據，但是它的 M 乘以 R 量級是 4 萬乘 4 萬。簡單算一下，在這個例子中，平均的 Fetch Chunk 大小甚至遠遠小于 1K ，量級是非常非常小的。
再看一個混部集群中 Spark 作業的 Shuffle Fetch-Failure 的實時監控。下圖監控中每個點的含義是——在這個時刻處于 Running 狀態的 Application 的 Fetch-Failure 次數的總和。
上文提到，每一個 Fetch-Failure 都可能意味著一定時間的超時等待和計算資源空跑，同時還可能意味著觸發 Stage 重算，甚至作業的失敗。
所以，解決這個問題對于提升 Spark 的資源利用率和穩定性都具有重要意義。
問題總結
綜上所述，ESS 在字節跳動業務場景下面臨如下問題：
-
Chunk Size 過小導致磁盤產生大量隨機 IO，降低磁盤的吞吐，引發 Chunk Fetch 請求的堆積、超時甚至引發 Stage Retry；
-
磁盤 IOPS 無法在操作系統層面進行隔離，Shuffle 過程中不同 Application 作業會互相影響；
-
在離線混部場景下，我們希望利用在線服務業務低峰期的 CPU，但缺少對應的磁盤資源。
02
External Shuffle Service 的優化
針對上述問題和需求，我們先對 ESS 進行了優化。
參數調優
首先是參數調優。為了實現參數調優，我們研發了一個旁路系統，如下圖。
-
首先，采集 Spark、Yarn 運行時的 Event Log 作為數據源；
-
其次，使用 Flink 對原始數據進行 Join 和計算，得到作業某個 Stage 的 Shuffle 量、Task 數量等指標；
-
針對上述指標，
- 一方面，在計算過程使用可插拔的啟發式規則對單個作業進行診斷；
-
另一方面，同時存在著大量的周期作業重復運行生成該作業的歷史畫像；
-
-
最終，結合歷史畫像與特征診斷信息對特定作業進行自動調參。
下面是一個自動調參的例子。經過若干次調參的迭代后，最終調整了兩個參數并達到穩定狀態：
-
spark.sql.adaptive.shuffle.targetPostShuffleInputSize：64M->512M
-
spark.sql.files.maxPartitionBytes：1G->40G
最終效果如下圖，
因為我們增大了單個 Task 處理的數據量，恰好這個作業又使用了 Combine 算子，所以它整體的 Shuffle 量有所降低，從 300G 降低到了 68G。
因為增大了這個 Chunk Size，也就是降低了這個作業的并發度，從而減小了整個 Shuffle 過程中的 IOPS，避免了長時間的 Blocked Time。如截圖所示，大家可以看到就是在截圖的指標里邊， Shuffle Read Blocked Time 最大從 21 分鐘降到了 79 毫秒，整體這個作業的端到端時間也降低為原來的一半，從 40 多分鐘降到了 20 分鐘。
以上是參數調優對這一個作業的影響，實際上這一個作業的調優還會影響其他作業。在調參之前，21 分鐘的 Shuffle Read Blocked Time 意味著磁盤是忙碌狀態，在這個磁盤上的其他作業都會受到影響。當前在線上，我們針對 Shuffle 進行自動調參的作業大概有 2 萬個，大量數據的驗證表明，調參優化的效果是很不錯的。
Shuffle 限流
Shuffle 限流主要解決的是磁盤的 IOPS 不易隔離的問題。我們通過對低優但高負載的作業進行限流，來減輕對同節點上高優作業的影響。整體的思路是當我們發現 ESS 響應請求的 Letency (延遲）升高到一定程度時，比如 10 秒或 15 秒，我們就認為這個節點當前處于異常狀態，這時 ESS 就會針對內部正在排隊的 Fetch 請求，按照 Application 分類進行分析，綜合當前堆積的排隊長度和作業的優先級，給每個作業劃定一個合適的長度范圍，超過范圍的作業會被 ESS 告知對應的 Shuffle Client 進行休眠，暫停數據請求，通常暫停1～2分鐘，這時該作業的客戶端就進入休眠狀態，進行等待，同時原本分配給它的 ESS 的服務能力提供給更高優或其他不受影響的作業。
通過 Shuffle 限流，我們實現了以下目標：
-
正常任務打開限流沒有影響，不會觸發流量限制；
-
異常任務開啟限流，不會讓任務變慢或失敗，大概率會使得任務變快 （限流減少重試，減輕 Server 壓力）；
此處有必要解釋一下，為什么任務會變得更快呢？原因在于當 Latency 升高時，Chunkr Fetch 開始堆積，大量排隊，此時往往容易形成惡性循環，請求過來-開始排隊-超時-超時后重試-重試后繼續排隊-繼續超時，Fetch 請求可能永遠都得不到正常響應。 但當我們開啟限流之后，我們主動地讓客戶端等待，而非發一個請求過來在服務端排隊，由此就可以避免大量無效的 Fetch 請求。也正因如此，大概率即便是被限流的作業也會變得更快。
-
不同優先級的任務，在限流情況下，高優先級任務允許更高的流量；
上文提到，我們是根據排隊的數量，及作業的優先級綜合地劃定一個合適的范圍。在劃定這個范圍的時候，更高優的作業大概率是不會被限流的。
-
異常節點快速恢復，2min~5min 能恢復正常。
結合第二點，因為我們讓一部分發送大量 Fetch 請求的作業的客戶端進行了等待休眠，所以異常節點會得到一個非?？焖俚幕謴?，大概 2～5 分鐘就能恢復正常，恢復正常后，就可以給所有的 Fetch 繼續提供服務。
03
Cloud Shuffle Service 的設計與實現
我們針對 ESS 存在的問題進行了上述優化，但是 ESS 的 Fetch Based 的整體思路決定了其存在不可避免的性能瓶頸（隨機讀、寫放大）。針對這些問題，我們自研了 Cloud Shuffle Service（以下簡稱CSS），接下來從基本思路、整體架構、讀寫過程、性能分析四個方面闡述 CSS 的設計與實現。
基本思路
Cloud Shuffle Service 的整體思路是 Push Based Shuffle，在 Shuffle Write 階段，直接把相同 Partition 的數據通過網絡寫入到遠端的一個 Buffer 并最終 Dump 到文件中，在 Shuffle Read 階段，可以通過連續讀的方式直接讀取已經合并好的文件。對該思路進行拆解，我們可以概括為以下三個方面：
第一個問題是備份。為了解決我們在背景中提到的大量隨機讀請求的問題，我們需要在 Reduce 讀取前使用 Push Shuffle 的方式將數據聚合到一起。由于是遠程聚合，所以還可以順便解決本地磁盤空間不足的問題。
然而，聚合雖然可以解決隨機 Shuffle 的問題，但也會帶來一個新的問題——數據丟失的成本比原來更高。原因在于，以前每個 Task 生成自己的文件雖然沒有備份，但這個文件丟失的成本是非常低的，只需要單個 Task 重算即可。但當我們把所有 Map Task 同一個環節的數據都聚合到一起時，一旦發生數據丟失，就需要重算整個 Stage。
因此我們需要對這些數據進行備份。備份的時候，我們發現 HDFS 太重了，它的寫入速度滿足不了我們的需求，隨后我們就采用了雙磁盤副本的方式，通過自己管理兩個客戶端雙寫來解決這個問題。
第二個問題是 IO 聚合。IO 聚合對于讀提升是顯而易見的，因為它將大量的隨機 IO 變成了極少數的連續 IO，但是在寫入速度上就有可能會受影響。因為寫入的時候原來是直接寫本地盤，現在變成需要通過網絡請求來寫數據。
同時因為可能需要多個 Mapper 去寫一個 Buffer，這個時候就有可能在寫 Buffer 的時候會產生鎖的爭搶，這些都是寫入時的代價。這就需要我們去花更多的時間在寫入時去做優化。
所以面臨的第三個問題是寫入速度。在寫入速度的優化上，我們選擇了主從 InMemory 副本，全部都是異步刷盤。即在數據寫入到服務端的內存后就快速返回主從，寫入到內存中的數據通過異步的方式去刷到磁盤里面。這其中有一個風險，即如果主從同時刷盤失敗，就會造成數據丟失。主從只有一個刷完失敗的話，有一個磁盤的文件數據丟失，另外一個磁盤的文件是沒有丟失，但因為可能后續可能繼續運行一段時間，可能將來完整的文件都會丟失，雖然不是同時丟失，但可能會在不同的時間丟失數據，這樣的話就會造成整個 Stage 重算。但我們認為這個概率是非常非常低的,我們以極小的失敗幾率換取更高速的寫入速度是完全值得的。事實也證明，這個思路是正確的，在整個 CSS 的應用過程中，到目前我們還沒有在線上觀察到任何一起數據丟失的問題。
整體架構
CSS 整體架構
上圖右側是 CSS 的整體架構，主要分為4個部分：
-
Zookeeper WorkerList：我們使用 zookeeper 來提供服務發現的功能；
-
CSSWorker [Partitions / Disk |HDFS]：管理磁盤并提供 Shuffle Push 服務節點。每一個機器上都會啟動 Worker 進程，當收到啟動指令時，它就會向 Zookeeper 進行注冊，并定時更新上報信息；
-
SparkDriver：集成啟動 CSS Master 和 ClusterName + ZK
-
-
CSS Master 的作用是規劃和統計，Master 從 Zookeeper 中拉取所有 Worker 的信息，并對 Worker 進行分配，然后把 Worker 和 Shuffle 以及每個 Partition 的對應關系通知到 Executor
-
ClusterName + ZK：通過配置的 ClusterName 在 ZK 中尋找對應的 Workerlist
-
-
CSSShuffleClient：Writer 和 Read 的集合，負責跟 Worker通信，讀取數據或寫入數據。
讀寫過程
下面我們來看讀寫過程，下圖是完整的寫入過程。
寫入過程
首先 Mapper 從 Master 中獲取分配好的 Worker List 及它們與 Partition 的對應關系，也就是上圖中 P0 對應的 Worker 0 和 Worker 1。
隨后 Mapper 開始寫數據，正常的話它會把數據寫入到內存，然后返回，由 Worker 異步地把數據刷到磁盤中。
直到某一次 Worker 刷數據的時候發生異常，數據沒有寫到磁盤中，比如說此時磁盤突然壞了。此時，實際上這個請求已經返回給了 Mapper，Mapper 會認為它的兩次寫都是成功的，直到 Mapper 下一次寫的時候，因為 Worker 已經把異常記錄到了內存里，等 Mapper 下次寫的時候，Worker 就會向 Mapper 返回上次寫入失敗的信息。
這時 Mapper 意識到它上次寫入的數據是失敗，這時他就會向 Master 再申請新的一個 Worker 就是我們看到的 Worker 3，再繼續進行寫入請求。
大家可以注意到，在第一個文件也就是 P0-0 里，實際上它保存了失敗前所有的數據，因此這個過程中實際上并沒有數據丟失，最終生成的成功的完整文件就是 P0-0、P0-1 和 P0'-1 三個文件。
此處有必要提到，實際上 P0-0 里是包含了一份多余的信息，即 P0-1 的第一條數據。下面我們說讀取過程的時候也會提到。
讀取過程
上文提到，正確的文件有三個，P0-0 是唯一一個正確的文件，P0-1 和 P0'-1 可以任選其一。
這些 Mata 信息其實都記錄在 Driver 的 Master 里，然后 Reduce 會根據這些文件的 Banner 信息選擇合適的文件來讀取。
值得一提的是去重，除了寫失敗可能導致的數據重復之外，因為 Spark 支持推測執行，所以還可能存在其他的重復問題，所以我們最終使用了 Mapld、Attemptld 和 Batchld 來共同進行數據去重。
性能分析
1TB 級別 TPS-DS 測試結果
CSS 開發完成后，我們用 TPS-DS 進行了測試。上圖是 1TB 級別的 TPS-DS 的測試結果。
通過上圖可以看到，相比原生的 ESS，使用 CSS 在查詢時間上有整體30%的提升。在個別 Query，如 q38 和 q35，提升是非常明顯的，大概有 60% 到 70%。
上面是從線上作業中選取的一個具體案例。可以看到，在使用原生的 ESS 時，讀取時間是 20 分鐘左右。使用 CSS 后，因為 CSS 使用了更高壓縮比的壓縮算法，所以整體的 Shuffle 數據量減少了很多。同時因為 IO 聚合讀取的時間也非常快，降低到了秒級，三個 Stage 加一起可能都不到一分鐘，相比是原來讀取時間的 1/20。
04
Cloud Shuffle Service的應用實踐
上文分析了 Cloud Shuffle Service 的設計和實現，下面講一下 Cloud Shuffle Service 的應用實踐。
CSS 在字節內部已經推廣，最新的數據顯示：
-
CSS Worker 數量 1000+，對應1000多臺機器
-
部署模式靈活：Shell、Yarn、K8S
-
支持作業類型眾多：Spark、MR、Flink Batch
-
接入作業數 6w+
-
單日 Shuffle 量 9PB+
集群部署&作業接入
構建運維接入管理平臺（CSS-Coordinator）
為了降低接入門檻，我們構建了一個運維接入管理平臺，叫作 CSS-Coordinator，他提供了如下功能：
-
提供用戶作業無感知接入功能：直接幫用戶注入 CSS 相關的參數；
-
提供 Cluster｜Queue｜Job等維度的灰度模式：支持以各種緯度接入作業，用戶僅需配置對應的接入緯度，該維度下的所有作業都會接入到 CSS 中；
-
異常作業的監控告警：作業運行結果會上報到 Coordinator 平臺，對于運行失敗的作業會進行報警
-
歷史 Shuffle 作業的 HBO 優化：平臺在作業接入過程中會針對作業歷史的 Shuffle 數據量進行評估，當 CSS 集群資源不足時會拒絕大 Shuffle 的作業接入 CSS；
此外，我們設計了作業 Shuffle FallBack 機制：
-
設置 spark.yarn.maxAppAttempts=2
-
保留用戶原始配置
-
作業 CSS 失敗自動 FallBack 到原生 Shuffle
踩坑記錄
在實踐的過程中，我們也踩了很多坑：
CSS 服務相關
-
超大 Register Shuffle 啟動緩慢
-
在最初的設計中，Register Shuffle 會對所有 Worker 進行初始化工作。因此，在規模比較大的 Shuffle 的場景下，Register 就會非常慢，用戶啟動一個 Stage 可能需要 2-3 分鐘。
后來，我們對 Register Shuffle 進行了精簡，把 Worker 的初始化動作改成了 Lazy 模式，即只有第一次數據 Push 過來的時候，Worker 才針對這一個作業的 Partition 進行對應的初始化工作。在 Register Shuffle 的時候，只進行 Worker 和 Partition 之間的分配，大大緩解了超大 Register Shuffle 啟動緩慢的問題。
-
Client 發送速率過快
-
因為我們是一個有狀態的服務，無法把 QPS 通過負載均衡的方式降下來，只能通過一些負反饋的方式讓 Client 降速，即當 Server 的服務能力無法滿足請求時，就讓請求在客戶端等待。
后續我們嘗試了很多方法，包括 Spark 原生的 Max Inflight 等，但效果都不太好，最終我們選擇了.NETflix 的一個三方庫。
大致原理是，針對最近一段時間的 RTT 做一個 Smoth 處理，得到一個理論的 RTT，然后拿當前的 RTT 與理論 RTT 做比較，如果小于這個值的話，就在 QPS 上做爬坡。如果大于這個值的話，系統就認為現在的 Server 有排隊現象，然后就啟動限流。
-
服務熱上線，用戶如何不感知
-
在 CSS beta 的過程中，每天都會有新的 Commit 合到主分支，每天也會產生新的問題。但是公司內部的 Spark 發展周期是比較長，跟 CSS 的迭代周期無法 Match。
最終，我們在 Spark 里只集成了一個最簡單的接口，其他的實現都放到 HDFS 上，這樣就把公司內 Spark 版本的周期與 CSS 的版本周期做了解耦，CSS 就可以做到小步快跑。在小步快跑的過程中，那我們解決了大量的問題。
接下來看2個與 Spark 集成相關的問題：
-
AQE Skew-Join 讀放大問題
AQE Skew-Join 原理圖
上圖是 Spark 社區提供的 AQE Skew-Join 原理圖，根據這個原理，當 Spark 發現某一個 Partition 數據非常大，遠超其他 Partition 的時候，它會主動把該 Partition 的數據拆分成多份數據，然后分別去做 Join。這樣最終每個 Task 處理的數據量就會更平均，整體作業的運營時間也會變短。
設想一下，當我們把 Map 的數據全部聚合起來后會發生什么？一個文件會讀很多遍，每次讀的時候還會 Skip 很多無效數據。舉個例子，一個傾斜的 Partition 上有 1T 數據，Spark 想把它拆成十份去讀，這時會發現什么呢？就是這個被聚合后 1T 的文件要讀 10 遍，且每次有 1/9 讀到的數據都是 Skip 的。
面對這個問題，我們的解決辦法也非常樸素，就是不再盲目地追求生成一個非常大的連續文件。實際上我們要解決的就是隨機讀的問題，所以只要文件足夠大就可以。因此，我們把文件默認按照 512G 的大小進行切分，一個大的 Partition 數據最終會被切分成若干個小文件。
比如上文的例子，1T 的數據會被切分成很多份 512G 的文件，當 AQE Skew-Join 觸發時，就不必把一個超大文件讀很多遍，只需把這些 512G 的文件按需分配給不同的 Task 進行 Join 就可以。
-
Task Huge Partition 導致 Executor 內存占用過大
在最初的設計中，基于 Push 的特性，我們是不想做排序的。最初的思路類似于 By Pass 的實現思路，給每一個 Partition 準備一個 64k 的 Buffer，一旦這個 Partition 的 Buffer 寫滿，就發送出去。后來發現當 Partition 數量非常大的時候，Buffer 就會占非常大的空間。
假設一個極端的場景，當有 10 萬個Partition 時，如果一個 Partition 的 Buffer 是 64k，那占用的內存還是非常大的。所以最終我們還是回到了 Sort 的路線，即把數據整體在內存里寫滿之后，再進行 Source Build， 那么 Spill 也不會再寫到磁盤里，Spill 之后也不需要 Merge 把 Spill 的數據發送出去。
這樣做還可以降低 Push 的請求數，同一個 Worker 不同的 Partition push 數據的時候，就可以把它們放到一起放到 Push Request 里。
收益分析
下面是我們線上的一些實際收益。第一個例子是某業務某個小時級的任務，這個任務的規模很大，有1.2 萬 Cores，在混部隊列上平均需要 2.5 小時。使用 CSS 之后，平均速度提升到了 1.3 個小時，提升 50%。
第二個例子是某業務 3小時周期調度任務整體穩定性的提升。在使用 CSS 之前，因為它的 Shuffle 經常觸發 Fetch-Failure 異常，造成作業頻繁重試，有時可能需要重試 8 次才能最終成功。接入 CSS 后，所有作業都可以一次性跑完，整體的穩定性提升了 70%。
05
未來展望
下面是 CSS 未來的規劃和展望。
第一是服務分級，即如何滿足 Quota & Shuffle 優先級，對不同的業務承諾不同的 SLA，未來我們希望 CSS 能以更有力的方式保證高優業務。
第二是CSS作業構建 Shuffle 元倉，進行更好的 HBO 優化。當前 CSS Shuffle 數據元倉的 HBO 優化只有比較簡單的 Yes 和 No 的功能，即用戶根據歷史的 Shuffle 數據量來允許或者拒絕提供服務。后續我們希望可以根據元倉的數據加強調度，比如把數據量大的作業更廣泛地打散，讓大的 Shuffle 作業和小的 Shuffle 作業同時分配在一臺機器上或一塊磁盤上，避免把很多大的作業同時調度到一塊磁盤上，從而讓負載更加均勻。
第三是CSSWorker 支持異構機器，自動調節負載，降低運維成本。因為我們收到的節點的型號、磁盤數量、網卡數量都不一樣。目前我們分配的算法會考慮負載能力，但是對相對比較靜態的負載，負載能力的這種差異還無法完全地自適應異構機器，自動地調節負載。在缺失這些能力的情況下，如果我們一個集群里使用了異構機器，就會導致某些相對來說性能比較差的機器，影響整個作業的性能。但是如果我們把不同類型的機器拆分出來，做成不同的集群，又會提高運維成本。所以支持異構機器是我們將來一個非常重要的目標。
此前，Cloud Shuffle Service 已在 Github 上開源，基于字節跳動大規模實踐的火山引擎批式計算 Spark 版也已經上線火山引擎，支持公有云、混合云及多云部署，全面貼合企業上云策略，歡迎掃碼了解
