在字節(jié)跳動內(nèi)部，Spark 計(jì)算引擎被廣泛應(yīng)用于大規(guī)模數(shù)據(jù)處理，機(jī)器學(xué)習(xí)等場景，天任務(wù)數(shù)超過 150W。線上集群磁盤類型多樣，包括 SSD、HDD 及混合等。每天會產(chǎn)生超過 100PB 以上的 Shuffle 數(shù)據(jù)，同時單個任務(wù)的 Shuffle 數(shù)據(jù)量可能達(dá)到數(shù)百 TB。巨量的 Shuffle 數(shù)據(jù)和復(fù)雜的計(jì)算資源環(huán)境也給 Spark 運(yùn)行過程中的 Shuffle 性能帶來了很多挑戰(zhàn)。本文將從背景介紹、穩(wěn)定性資源場景和混部資源場景分享字節(jié)跳動在 Spark Shuffle 云原生化方面的大規(guī)模演進(jìn)實(shí)踐。

一、背景介紹

Spark 是字節(jié)跳動內(nèi)使用廣泛的計(jì)算引擎，已廣泛應(yīng)用于各種大規(guī)模數(shù)據(jù)處理、機(jī)器學(xué)習(xí)和大數(shù)據(jù)場景。目前中國區(qū)域內(nèi)每天的任務(wù)數(shù)已經(jīng)超過 150 萬，每天的 Shuffle 讀寫數(shù)據(jù)量超過 500 PB。同時某些單個任務(wù)的 Shuffle 數(shù)據(jù)能夠達(dá)到數(shù)百 TB 級別。

與此同時作業(yè)量與 Shuffle 的數(shù)據(jù)量還在增長，相比去年，今年的天任務(wù)數(shù)增加了 50 萬，總體數(shù)據(jù)量的增長超過了 200 PB，達(dá)到了 50% 的增長。Shuffle 是用戶作業(yè)中會經(jīng)常觸發(fā)的功能，各種 ReduceByKey、groupByKey、join、sortByKey 和 repartition 的操作都會使用到 Shuffle。所以在大規(guī)模的 Spark 集群內(nèi)，Spark Shuffle 經(jīng)常會成為性能及穩(wěn)定性的瓶頸。Shuffle 的計(jì)算涉及到頻繁的磁盤和網(wǎng)絡(luò) IO 操作，主要是需要把所有節(jié)點(diǎn)的數(shù)據(jù)進(jìn)行重新分區(qū)并組合。

1、原理

在社區(qū)版 ESS 模式下默認(rèn)使用的 Shuffle 模式的基本原理中，剛才提到 Shuffle 的計(jì)算會把數(shù)據(jù)進(jìn)行重新分區(qū)，這里就是把 Map 的數(shù)據(jù)重新組合到所有的 Reducers 上。如果有 M 個 Mappers，和 R 個 Reducers，就會把 M 個 Mappers 的 Partition 數(shù)據(jù)分區(qū)成后面 R 個 Reducers 的 Partition。

Shuffle 的過程可以分為兩個階段——Shuffle Write 和 Shuffle Read。

Shuffle Write 的時候，mapper 會把當(dāng)前的 Partition 按照 Reduce 的 Partition 分成 R 個新的 Partition，并排序后寫到本地磁盤上。生成的 map output 包含兩個文件：索引文件和按 Partition 排序后的數(shù)據(jù)文件。

當(dāng)所有的 Mappers 寫完 map output 后，就會開始第二個階段，Shuffle Read 階段。這個時候每個 Reducer 會向包含它的 Reducer Partition 的所有 ESS 訪問，并讀取對應(yīng) Reduce Partition 的數(shù)據(jù)。這里有可能會請求到所有 Partition 所在的 ESS，直到這個 Reducer 獲取到所有對應(yīng)的 Reduce Partition 的數(shù)據(jù)。

在 Shuffle Fetch 階段，每個 ESS 會收到所有 Reducer 的請求并返回相應(yīng)的數(shù)據(jù)。這將產(chǎn)生 M 乘 R 級別的網(wǎng)絡(luò)連接和隨機(jī)的磁盤讀寫 IO，涉及到大量的磁盤讀寫和網(wǎng)絡(luò)傳輸。這就是為什么 Shuffle 會對磁盤以及網(wǎng)絡(luò) IO 的請求都特別頻繁的原因。

由于 Shuffle 對資源的需求和消耗都非常高，所以 CPU、磁盤和網(wǎng)絡(luò)開銷都很有可能是造成 Fetch failure 的原因或 Shuffle 速度較慢的瓶頸。在字節(jié)跳動大規(guī)模的 Shuffle 場景中，同一個 ESS 節(jié)點(diǎn)可能需要同時服務(wù)多個商戶，而這些集群沒有進(jìn)行 IO 的隔離，就可能會導(dǎo)致 Shuffle 成為用戶作業(yè)失敗的主要原因和痛點(diǎn)問題。

字節(jié)跳動從 2021 年初開始了 Spark Shuffle 的云原生化相關(guān)工作，Spark 作業(yè)與其他大數(shù)據(jù)生態(tài)開始了從 Yarn Godel 的遷移。Godel 是字節(jié)跳動基于 Kubernetes 自研的調(diào)度器，遷移時也提供了 Hadoop 上云的遷移方案——Yodel（Yarn on Godel），是一個完全兼容 Hadoop Yarn 的協(xié)議，目標(biāo)是將所有大數(shù)據(jù)應(yīng)用平滑地遷移到 Kubernetes 體系上。

在這套遷移工作中，ESS 也做了定制化的相關(guān)工作，完成了從之前 Yarn Node Manager 模式下的 Yarn Auxiliary Service 遷移至 Kubernetes DaemonSet 部署模方式的適配工作，并開始 Shuffle 作業(yè)的遷移工作。歷時兩年，在 2023 年，順利將所有大數(shù)據(jù)應(yīng)用包括 Spark 應(yīng)用都遷移到了如今的云原生生態(tài)上。

2、挑戰(zhàn)

在云原生化的遷移過程中，也遇到了很多挑戰(zhàn)：

• 首先，從 NM 遷移到 DaemonSet 的過程中，DaemonSet 上 ESS 的 CPU 有非常嚴(yán)格的限制，而在之前的 NM 模式下，ESS 基本上可以使用所有的 CPU 資源。所以在這個遷移實(shí)踐中，往往最開始設(shè)置的 ESS 的 CPU 資源是不夠的，需要經(jīng)過持續(xù)不斷的調(diào)整。后續(xù)，某些高優(yōu)集群甚至直接放開對 ESS 的 CPU 使用。
• 同時，DaemonSet 和 Pod 對 Spark 作業(yè)的 CPU 有更嚴(yán)格的限制。這也導(dǎo)致不少用戶的作業(yè)遷移到了新的架構(gòu)后變得更加緩慢了。這是因?yàn)樵谥暗哪Ｊ较?，CPU 是有一定的超發(fā)的，因此需要對這個情況進(jìn)行調(diào)整。我們在 Kubernetes 和 Godel 架構(gòu)下開啟了 CPU Shares 模式，使用戶在遷移過程中感知不到性能上的差異。
• 另外，Pod 對內(nèi)存的限制也非常嚴(yán)格，這導(dǎo)致 Shuffle Read 時無法使用空閑的 page cache 資源，從而導(dǎo)致 Shuffle Read 時 page cache 的命中率非常低。這個過程會帶來更多的磁盤 IO 開銷，導(dǎo)致整體性能變差。對此我們采取了相應(yīng)的措施，通過適當(dāng)開放 Pod 對 page cache 的使用，降低 Shuffle 在遷移后對性能的影響。

3、收益

完成遷移工作之后，我們成功地將所有的離線資源池完成統(tǒng)一，在調(diào)度層面能夠更友好地實(shí)施一些優(yōu)化和調(diào)度策略，從而提高整體的資源使用率。ESS Daemonset 相比于 Yarn Auxilary Service 也獲得了不少的收益。首先，ESS DaemonSet 被獨(dú)立出來成為一個服務(wù)，脫離與 NM 的緊耦合，減少了運(yùn)維成本。另外，Kubernetes 和 Pod 對 ESS 資源的隔離也增加了 ESS 的穩(wěn)定性，這意味著 ESS 不會再受到其他作業(yè)或者節(jié)點(diǎn)上其它服務(wù)的影響。

云原生化后的 Spark 作業(yè)目前有兩個主要的運(yùn)行環(huán)境：

• 穩(wěn)定資源集群環(huán)境。這些穩(wěn)定資源的集群主要以服務(wù)高優(yōu)和 SLA 的任務(wù)為主。部署的磁盤是性能比較好的 SSD 磁盤。對于這些穩(wěn)定資源集群，主要使用基于社區(qū)、深度定制化后的 ESS 服務(wù)。使用 SSD 磁盤，ESS 讀寫，也可以使用到本地的高性能 SSD 磁盤。部署在 Daemonset 模式，Godel 架構(gòu)下。
• 混部資源集群環(huán)境。這些集群主要服務(wù)于中低游的作業(yè)，以一些臨時查詢、調(diào)試或者測試任務(wù)為主。這些集群的資源主要都部署在 HDD 磁盤上，有些是通過線上資源出讓或與其他服務(wù)共用的或者其他線上的服務(wù)共同部署的一些資源。這就導(dǎo)致集群的資源都不是獨(dú)占的，整體的磁盤性能以及儲存環(huán)境也都不是特別優(yōu)異。

二、穩(wěn)定資源場景

在穩(wěn)定集群環(huán)境中，存在較多的高優(yōu)作業(yè)，首要任務(wù)是提高這些作業(yè) Shuffle 的穩(wěn)定性，以及運(yùn)行時的作業(yè)時長，以確保這些作業(yè)的 SLA。為了解決 Shuffle 的問題，對 ESS 深度定制了以下三方面能力：增強(qiáng) ESS 的監(jiān)控/治理能力、增加 ESS Shuffle 的限流功能、增加 Shuffle 溢寫分裂功能。

1、ESS 深度定制

（1）增強(qiáng) ESS 的監(jiān)控及治理能力

在監(jiān)控方面，我們使用開源版本的過程中發(fā)現(xiàn)現(xiàn)有的監(jiān)控不足以深度排查遇到的 Shuffle 問題和當(dāng)前的 ESS 狀況。這就導(dǎo)致沒有辦法快速定位是哪些節(jié)點(diǎn)造成的 Shuffle 問題，也沒有辦法感知到有問題的節(jié)點(diǎn)，因此，我們對監(jiān)控能力進(jìn)行了一些增強(qiáng)。

首先，我們增加了監(jiān)控 Shuffle 慢和 Fetch Rate 能力的一些關(guān)鍵指標(biāo)，包括 Queued Chunks 和 Chunk Fetch Rate。Queued Chunks 用于監(jiān)控當(dāng)前請求 ESS 節(jié)點(diǎn)上請求的堆積，而 Chunk Fetch Rate 用于監(jiān)控這些節(jié)點(diǎn)上請求的流量。同時，我們還將 ESS 的 Metrics 指標(biāo)接入了字節(jié)跳動的 Metrics 系統(tǒng)，使我們能夠通過系統(tǒng)提供的 Application 維度的指標(biāo)快速定位 ESS 節(jié)點(diǎn)的堆積情況。在用戶界面 (UI) 方面，我們在 Stage 詳情頁加入了兩個新功能，用于展示當(dāng)前 Stage 里每個 Task Shuffle 遇到最慢的幾個節(jié)點(diǎn)，以及經(jīng)過 Stage 統(tǒng)計(jì)后所有 Task 遇到 Shuffle 次數(shù)最多的 top 節(jié)點(diǎn)。這不僅方便用戶查詢，也可以利用這些指標(biāo)進(jìn)行相關(guān)大盤的搭建。

• 收益

有了這些監(jiān)控與 UI 改善后，當(dāng)用戶在 UI 上看到 Shuffle 慢的時候可以通過 UI 打開對應(yīng)的 Shuffle 監(jiān)控。方便用戶和我們團(tuán)隊(duì)快速定位到導(dǎo)致 Shuffle 問題的 ESS 節(jié)點(diǎn)，看到這些節(jié)點(diǎn)上的實(shí)際情況，并快速定位這些堆積請求量是來自于哪些 Application。

新增的監(jiān)控也會在運(yùn)行排查 Shuffle 問題時感知到 ESS 節(jié)點(diǎn)上實(shí)際的 Chunk 堆積、latency 等關(guān)鍵指標(biāo)。這在遇到 Shuffle 慢的情況下有助于更有效地實(shí)時采取措施。一旦定位到 Shuffle 問題，我們可以分析情況并提供治理方向和優(yōu)化。

治理工作主要是通過 BatchBrain 系統(tǒng)來實(shí)施。BatchBrain 是專門為 Spark 作業(yè)設(shè)計(jì)的一套智能作業(yè)調(diào)優(yōu)系統(tǒng)，它主要對作業(yè)數(shù)據(jù)進(jìn)行采集，并進(jìn)行離線與實(shí)時分析。采集的數(shù)據(jù)包括 Spark 本身的 Event Log、內(nèi)部打入更詳細(xì)的 Timeline event 以及各種 Metrics 指標(biāo)，包括對 ESS 加上的定制化 Shuffle 指標(biāo)等。

在離線分析中主要需要治理周期性作業(yè)，根據(jù)每個作業(yè)的歷史特征，結(jié)合采集的數(shù)據(jù)，對這些作業(yè)的 Shuffle Stage 性能進(jìn)行分析，并經(jīng)過多次迭代調(diào)整，最終提供一套適合的Shuffle 參數(shù)，使這些作業(yè)在重新運(yùn)行時可以對優(yōu)化后的Shuffle 參數(shù)進(jìn)行運(yùn)行，從而獲得更好的性能和效果。

BatchBrain 在實(shí)時分析部分也可以利用之前添加的 Shuffle 指標(biāo)進(jìn)行自動掃描。用戶還可以通過 BatchBrain API 查詢他們集群內(nèi)作業(yè)的 Shuffle 狀況，以及有效定位遇到 Shuffle 堆積的節(jié)點(diǎn)和作業(yè)，并通過報警通知相關(guān)人員。如果發(fā)現(xiàn) Shuffle 慢是由于其他的作業(yè)或者異常作業(yè)導(dǎo)致的，用戶也可以直接采取治理動作，例如停止或者驅(qū)逐這些作業(yè)，以便為更高優(yōu)先級的作業(yè)騰出更多資源進(jìn)行 Shuffle。

（2）Shuffle 限流功能

通過 Shuffle 的監(jiān)控和治理，我們發(fā)現(xiàn)在 ESS 節(jié)點(diǎn)上遇到 Shuffle 慢的情況，通常是因?yàn)槟承┤蝿?wù)的數(shù)據(jù)量過于龐大或者設(shè)置了不妥的參數(shù)，導(dǎo)致這些 Shuffle Stage 的 Mapper 和 Reducer 數(shù)量都異常地大。異常大量的 Mapper 和 Reducer 數(shù)量可能會導(dǎo)致 ESS 節(jié)點(diǎn)上出現(xiàn)大量的請求堆積，而這些請求的 chunk size 也可能非常小。有些異常作業(yè)的平均 Chunk size 可能連 20 KB 都沒達(dá)到。這些作業(yè)對 ESS 發(fā)送很大的請求量，這種情況可能會導(dǎo)致 ESS 無法及時處理所有的請求，從而引發(fā)請求堆積，甚至導(dǎo)致作業(yè)的延遲或直接失敗。

針對這些現(xiàn)象，我們采取的解決方案是對 ESS 節(jié)點(diǎn)上每個 Application 的總請求量進(jìn)行限制。當(dāng)某個 Application 的 Fetch 請求達(dá)到了上限，ESS 將拒絕該 Application 發(fā)送的新 Fetch 請求，直到該 Application 等待現(xiàn)有請求的部分結(jié)束后才能繼續(xù)發(fā)送新的請求。這樣可以防止出現(xiàn)單個 Application 占用節(jié)點(diǎn)上過大的資源而導(dǎo)致 ESS 沒有辦法正常為其他作業(yè)請求提供服務(wù)的情況，也可以避免其他作業(yè)失敗或 Shuffle 速度變慢。這個方案可以緩解異常或大規(guī)模的 Shuffle 作業(yè)對集群 Shuffle 的負(fù)面影響。

Shuffle 限流功能的特征

• 在作業(yè)運(yùn)行正常的時候，即使開啟了限流功能，也不會對作業(yè)有任何影響。節(jié)點(diǎn)如果可以正常服務(wù)，是不需要觸發(fā)任何限流的。
• 只有當(dāng)節(jié)點(diǎn)的負(fù)載超過可以承受的范圍，且 Shuffle IO 超過設(shè)置的閾值后，才會啟動限流機(jī)制，減少異常任務(wù)可以向 ESS 發(fā)送的請求數(shù)量，減低這個 ESS 服務(wù)當(dāng)前的壓力。由于這時候 ESS 服務(wù)的負(fù)載能力已經(jīng)超過了可承受的范圍，即使它收到這些請求，也無法正常返回這些請求，因此，限制異常任務(wù)過多的請求反而可能更好地提高這些任務(wù)本身的性能。
• 在限流的情況下，也會考慮作業(yè)的優(yōu)先級。對于高優(yōu)的任務(wù)，會允許更大的流量。
• 當(dāng)限流生效后，如果發(fā)現(xiàn) ESS 的流量已經(jīng)恢復(fù)正常了將迅速解除限流。受限流的 Application 很快就可以恢復(fù)到之前的流量水平。

限流的詳細(xì)流程

限流功能主要在 ESS 服務(wù)端進(jìn)行，每隔 5 秒在節(jié)點(diǎn)上進(jìn)行 latency 指標(biāo)的掃描，當(dāng)這個 latency 指標(biāo)超過設(shè)置的閾值時，會判定該節(jié)點(diǎn)的負(fù)載已經(jīng)超出能夠承受的負(fù)載了。接著會對 ESS 節(jié)點(diǎn)當(dāng)前所有正在進(jìn)行 Shuffle 的 Application 進(jìn)行評估，判斷是否要開啟限流。利用之前加上的指標(biāo)，可以統(tǒng)計(jì)近 5 分鐘這個節(jié)點(diǎn)上 Fetch 的總流量和 IO，根據(jù)總流量的上限，對每個 ESS 節(jié)點(diǎn)當(dāng)前正在運(yùn)行 Shuffle 的 Application 合理地分配每個 Application 的流量并進(jìn)行限制。流量分配也會根據(jù) Application 的優(yōu)先級進(jìn)行調(diào)整。如果有任何 Application 的 Shuffle 或者當(dāng)前堆積的 Chunk Fetch Rate 已經(jīng)超過了其分配的流量，它們將受到限流，新發(fā)送的請求也會被拒絕，直到堆積的請求已經(jīng)部分解除為止。

對于限流的分配，也有一個分級系統(tǒng)。首先，根據(jù)當(dāng)前節(jié)點(diǎn)上運(yùn)行 Shuffle 的 Application 的數(shù)量進(jìn)行分配，Application 的數(shù)量越多，每個 Application 可以分配到的流量就越少。當(dāng)節(jié)點(diǎn)上 Application 數(shù)量比較少的時候，每個 Application 可以分配更多的流量。限流級別也會根據(jù)節(jié)點(diǎn)上的實(shí)際情況每 30 秒進(jìn)行調(diào)整。

在限流的情況下，如果節(jié)點(diǎn)上的 latency 沒有改善，且 Shuffle 的總流量也沒有恢復(fù)，就會升級限流，對所有 Application 進(jìn)行更嚴(yán)格的流量限制。相反，如果 latency 有好轉(zhuǎn)或者節(jié)點(diǎn)流量已經(jīng)在恢復(fù)，就會降級限流甚至直接解除掉。最后，限流也會根據(jù)所有作業(yè)的優(yōu)先級進(jìn)行適當(dāng)調(diào)整。

上圖中有個例子，在作業(yè)較少的情況下，對一個高優(yōu)作業(yè)進(jìn)行限流，作業(yè)分配的流量可能會更高，然而，如果節(jié)點(diǎn)的負(fù)載一直沒有緩解，限流也會升級。同等的情況下，一個中低優(yōu)的作業(yè)，會給它分配更少的流量。開通限流功能之后，線上許多高優(yōu)集群都觀察到了性能的顯著提升。

首先，Chunk 的堆積問題得到了明顯的減輕。由于受到限流的限制，異常任務(wù)引發(fā)的 Chunk 堆積情況有效的減少了，大大降低了集群中某些節(jié)點(diǎn)上出現(xiàn)大量請求堆積的情況。

另外，Latency 的狀況也得到了改善。在開啟限流前，我們經(jīng)常會看到集群中的節(jié)點(diǎn)出現(xiàn)高延遲的情況。而在啟用限流功能后，整體的 Latency 狀況得到了明顯緩解。通過減少無必要和無效的請求，以及對各種大型或異常任務(wù)對 ESS 節(jié)點(diǎn)發(fā)起的請求量進(jìn)行限制，我們避免了這些異常大型任務(wù)對 ESS 服務(wù)負(fù)載的負(fù)面影響，減少了對其他高優(yōu)任務(wù)運(yùn)行的影響。

（3）Shuffle 溢寫分裂的功能

在分析一些慢 Shuffle 的作業(yè)時，我們也發(fā)現(xiàn)了另一個現(xiàn)象，一個作業(yè)中每個 Executor 寫 Shuffle 數(shù)據(jù)的數(shù)量可能非常不均衡。由于 ESS 使用了 Dynamic Allocation 機(jī)制，每個 Executor 的運(yùn)行時長和分配的 Map Task 數(shù)量可能不同。這導(dǎo)致在作業(yè)運(yùn)行期間，大量的 Shuffle 數(shù)據(jù)可能集中在少數(shù)的 Executor 上，導(dǎo)致 Shuffle 數(shù)據(jù)實(shí)際上都集中在少數(shù)節(jié)點(diǎn)上。

例如下圖中，我們發(fā)現(xiàn)有 5 個 Executor 的 Shuffle 寫入量超過了其他 Executor 的 10 倍以上。在這種情況下，Shuffle 的請求可能會集中在這幾個節(jié)點(diǎn)上，導(dǎo)致這幾個 ESS 節(jié)點(diǎn)的負(fù)載非常高，這也間接增加了 Fetch Failure 的可能性。

針對這種情況，我們提供的解決方案是控制每個容器或每個節(jié)點(diǎn)寫入磁盤的 Shuffle 數(shù)據(jù)總量。這個功能可以從兩個角度實(shí)現(xiàn)。首先，通過 Spark 本身來控制 Executor 的 Shuffle Write Size，也就是每個 Executor 在執(zhí)行 Shuffle 時寫入的最大數(shù)據(jù)量。每個 Executor 會計(jì)算其當(dāng)前寫入的 Shuffle 數(shù)據(jù)量，并將這信息匯報給 Spark Driver。Spark Driver 可以使用 Exclude on Failure 機(jī)制主動將那些寫入數(shù)據(jù)已經(jīng)超出閾值的 Executor 排除在調(diào)度范圍之外，并回收這些 Executor。此外，我們還通過 Godel 調(diào)度器改善調(diào)度策略，盡量將新的 Executor 調(diào)度到其他節(jié)點(diǎn)，避免單個容器的 Shuffle 寫入數(shù)據(jù)過多，從而導(dǎo)致該節(jié)點(diǎn)的磁盤被填滿，或者在 Shuffle Fetch 階段數(shù)據(jù)集中在這幾個 ESS 節(jié)點(diǎn)上。

2、云原生優(yōu)化

同時，在云原生優(yōu)化方面，我們也進(jìn)行了一些 Executor 的調(diào)度和功能優(yōu)化，通過 Godel 調(diào)度器的策略，提升 Shuffle 能力。Godel 調(diào)度器提供的調(diào)度策略，可以在調(diào)度 Executor 時盡量避免負(fù)載高的 Shuffle 節(jié)點(diǎn)，從而降低這些節(jié)點(diǎn)后續(xù)遇到 Shuffle 問題的可能性。此外，調(diào)度器還可以為 Executor 的 Shuffle Write 提供更多的功能以實(shí)現(xiàn)打散。例如，它可以在磁盤壓力特別大的節(jié)點(diǎn)上驅(qū)逐 Executor，或者在磁盤剩余空間不足時，驅(qū)逐那些已經(jīng)寫入大量 Shuffle 數(shù)據(jù)的容器。

Spark Driver 控制 Executor 的 Shuffle 與云原生調(diào)度功能結(jié)合可以將整體的 Shuffle 數(shù)據(jù)分散到更多的節(jié)點(diǎn)上，使 Shuffle Fetch 階段的數(shù)據(jù)和請求更加均衡分布。

• 效果

在線上開啟了上述深度定制的 Shuffle 優(yōu)化后，我們觀察到了顯著的效果。以下是來自三個高優(yōu)集群的一些運(yùn)行數(shù)據(jù)，每天在這三個高優(yōu)集群中的任務(wù)總數(shù)可能超過 30 萬，但平均每天因?yàn)?nbsp;Shuffle Fetch 失敗而最終失敗的作業(yè)總數(shù)平均在 20 到 30 左右，可以說達(dá)到了低于 1/10000 的失敗率。如下圖可以觀察到這三個高優(yōu)集群在優(yōu)化后的穩(wěn)定性都有了顯著的提升，也大幅度減少了用戶在 Shuffle 上遇到的問題。

三、混部資源場景

接下來介紹在混部場景中進(jìn)行的優(yōu)化。首先值得注意的是，在混部集群場景下，F(xiàn)etch Failure 的情況通常比在穩(wěn)定資源環(huán)境中嚴(yán)重得多。每天平均的 Fetch Failure 次數(shù)非常高，主要原因是這些資源大多來自于線上資源空閑的出讓，它們的磁盤 IO 能力和磁盤空間都比較有限。此外，由于磁盤 IOPS 和磁盤空間可能非常有限，與 HDFS 或其他服務(wù)混合部署的資源對集群的 Shuffle 性能影響較大，因此發(fā)生失敗的概率也較高?；觳抠Y源治理以降低作業(yè)的失敗率，確保作業(yè)的穩(wěn)定性為主要目標(biāo)，同時需要提高整個集群的 Shuffle 性能，減少資源浪費(fèi)。

對于混部資源的集群，主要的方案是自研的 Cloud Shuffle Service（CSS），通過提供一個遠(yuǎn)端的 Shuffle 服務(wù)來減少這些作業(yè)對本地磁盤的依賴。

1、CSS 功能介紹

首先，CSS 提供了一個 Push Based Shuffle 模式，與剛才介紹的 ESS 模式不同，在 Push Based Shuffle 模式下，不同 Mapper 的同一個 Reducer Partition 數(shù)據(jù)都會發(fā)送到一個共同的遠(yuǎn)程服務(wù)上，在這個服務(wù)上進(jìn)行合并，最后在某個 Worker 上寫上一個或者多個文件，使得 Reduce 階段可以通過 Sequential Read 模式讀取這些 Partition 數(shù)據(jù)，減少隨機(jī) IO 的開銷。

CSS 也支持 Partition Group 功能，它的作用是將多個分區(qū)數(shù)據(jù)分配到一個 Reducer Partition Group。這樣，在 Map 階段 Mapper 可以通過 Batch Push 方式傳送數(shù)據(jù)，將批量數(shù)據(jù)直接傳輸?shù)綄?yīng)分區(qū)組的工作節(jié)點(diǎn)上，從而降低了批量模式下 IO 的開銷，提高了批量模式的性能。

CSS 也提供了一個快速雙寫備份的功能。由于使用的是 push based Shuffle 和聚合模式，所有的數(shù)據(jù)其實(shí)都聚集在一個 Worker 上，如果這個 Worker 數(shù)據(jù)丟失的話，等于所有的 Mapper 都要重新計(jì)算所對應(yīng)的數(shù)據(jù)，因此對于 push 聚合的功能，使用一個雙寫備份是比較重要的。CSS 提高寫入的速度的方式是采用雙寫 In-memory 副本模式并進(jìn)行異步刷盤，這樣 Mapper 無需等待刷盤結(jié)束就可以繼續(xù)推送后續(xù)的數(shù)據(jù)。

CSS 本身也具有一個負(fù)載均衡功能。CSS 通過一個 Cluster Manager 去管理所有服務(wù)上的節(jié)點(diǎn)。Cluster Manager 會定期去采集和收取 CSS Worker 節(jié)點(diǎn)匯報的負(fù)載信息，當(dāng)有新的 Application 提交的時候，它會進(jìn)行資源的均衡分配，以確保 Shuffle Write 和 Shuffle Read 會優(yōu)先分配到集群上使用率較低的節(jié)點(diǎn)，從而實(shí)現(xiàn)集群中更好的 Shuffle 負(fù)載均衡。

2、CSS 整體架構(gòu)

• Cluster Manager 負(fù)責(zé)集群的資源分配，并維護(hù)集群 Worker 和 Application 狀態(tài)，它可以通過 Zookeeper 或者本地磁盤保存這些信息，達(dá)到具有 High Availability 的服務(wù)。
• Worker 支持兩種寫入模式，分別是磁盤模式和 HDFS 模式。目前常用的是磁盤模式，每個分區(qū)的數(shù)據(jù)會寫入兩個不同的 Worker 節(jié)點(diǎn)，以實(shí)現(xiàn)數(shù)據(jù)冗余。
• CSS Master 位于 Spark driver 端，主要負(fù)責(zé)與 Cluster Manager 的心跳聯(lián)系以及 Application Lifecycle。作業(yè)啟動時，也會向 Cluster Manager 申請 Worker。Shuffle Stage 的過程也會統(tǒng)計(jì) Shuffle Stage 的元數(shù)據(jù)以及的進(jìn)展。
• Shuffle Client 是一個接入了 Spark Shuffle API 的組件，允許任何 Spark 作業(yè)直接使用 CSS 而無需額外配置。每個 Executor 會使用 ShuffleClient 進(jìn)行讀寫。Shuffle Client 在寫入時進(jìn)行雙寫，在讀的時候，它可以向任何一個存有數(shù)據(jù)的 Worker 讀取這些數(shù)據(jù)，如果其中一個 Worker 讀取失敗的話，也會自動切換到另一個 Worker 上，并對多讀的數(shù)據(jù)進(jìn)行去重。

CSS 在寫入時 Worker 會直接發(fā)送數(shù)據(jù)，Mapper 會同時將數(shù)據(jù)發(fā)送到兩個 Worker，Worker 不會等到刷磁盤之后返回給 Mapper，而是異步返回給 Mapper 結(jié)果，如果遇到失敗，會在下一個請求再通知 Mapper。這時 Mapper 會重新跟節(jié)點(diǎn)申請兩個新的 Worker，重新推送傳送失敗的數(shù)據(jù)。讀的時候可以從任何一個節(jié)點(diǎn)讀取數(shù)據(jù)，通過 Map ID，Attempt ID 和 Batch ID 進(jìn)行去重。

3、CSS 性能與未來演進(jìn)

在 1TB 的 TPC-DS Benchmark 性能測試下，CSS 在 30% 以上的 Query 中得到了提升。

CSS 作為一個遠(yuǎn)端 Shuffle 服務(wù)，特別適合云原生化，支持彈性部署和更多的遠(yuǎn)程儲蓄服務(wù)。目前 CSS 已經(jīng)完成了開源，有興趣的朋友可以去 CSS 開源網(wǎng)站了解更多信息，也希望把后面的一些迭代和優(yōu)化同步到社區(qū)上。在未來云原生化的演進(jìn)中需要支持彈性部署、支持遠(yuǎn)程存儲服務(wù)等相關(guān)能力。

以上就是本次分享的內(nèi)容，謝謝大家。