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