Apache Celeborn 社區(qū)的今天和明天

作者：周克勇 2023-12-04 07:42:13

本文將介紹 Apache Celeborn 的用途、核心設(shè)計及其社區(qū)現(xiàn)狀與發(fā)展規(guī)劃。數(shù)據(jù)的轉(zhuǎn)換指的是數(shù)據(jù)發(fā)生變化、產(chǎn)生新的數(shù)據(jù)或者被過濾掉等。具體的實現(xiàn)方式包括算子和 UDF，如常見的 Filter、Project、Aggregation、Join，內(nèi)置 function，自定義 UDF 等。

一、Celeborn 是什么

1、計算引擎中間數(shù)據(jù)帶來的挑戰(zhàn)

大數(shù)據(jù)計算就是數(shù)據(jù)的轉(zhuǎn)換和流轉(zhuǎn)。

數(shù)據(jù)的轉(zhuǎn)換指的是數(shù)據(jù)發(fā)生變化、產(chǎn)生新的數(shù)據(jù)或者被過濾掉等。具體的實現(xiàn)方式包括算子和 UDF，如常見的 Filter、Project、Aggregation、Join，內(nèi)置 function，自定義 UDF 等。

數(shù)據(jù)流轉(zhuǎn)包括兩種：

一種是持久化數(shù)據(jù)的流轉(zhuǎn)，通常是表數(shù)據(jù)。一個大數(shù)據(jù)計算作業(yè)從對象存儲或分布式存儲讀取結(jié)構(gòu)化、半結(jié)構(gòu)化或者非結(jié)構(gòu)化的數(shù)據(jù)，進(jìn)行各種分布式計算處理，最后寫回存儲系統(tǒng)，這就是持久化數(shù)據(jù)的流轉(zhuǎn)。這種數(shù)據(jù)流轉(zhuǎn)目前主要由湖格式來完成。

另一種是中間（臨時）數(shù)據(jù)流轉(zhuǎn)。這里的中間數(shù)據(jù)指的是大數(shù)據(jù)計算引擎在執(zhí)行某個作業(yè)的過程中所產(chǎn)生的臨時數(shù)據(jù)，一般是由于內(nèi)存中放不下溢出到外部存儲，在作業(yè)完成之后可以安全刪除。典型的中間數(shù)據(jù)有三種：Shuffle 數(shù)據(jù)、Spill 數(shù)據(jù)、Cache 數(shù)據(jù)。

Apache Celeborn 正是專注于中間數(shù)據(jù)的處理，以提高這些數(shù)據(jù)的流轉(zhuǎn)效率。

在 Celeborn 誕生之前，大數(shù)據(jù)計算引擎在中間數(shù)據(jù)的處理上存在著諸多挑戰(zhàn)。下面以 Shuffle 為例說明。Shuffle 的過程如下圖所示：

MapTask 將其產(chǎn)生的 shuffle 數(shù)據(jù)根據(jù) Partition id 做本地排序，當(dāng)內(nèi)存不夠用時會引入外排，而外排會產(chǎn)生對本地磁盤寫放大的問題。最終生成數(shù)據(jù)文件和 Index 文件。接下來，下游的每個 ReduceTask 會從每個 Shuffle 數(shù)據(jù)文件中讀取 Partition 數(shù)據(jù)。從單個文件看，假設(shè)下游有幾千上萬個 ReduceTask，就會被讀取幾千上萬次，且每次讀取的是隨機(jī)位置和小的數(shù)據(jù)量，因此磁盤的 IOPS 會非常大，導(dǎo)致穩(wěn)定性和性能問題。

另外，中間數(shù)據(jù)存儲在本地磁盤，這就要求計算節(jié)點(diǎn)有大容量的本地磁盤。現(xiàn)在的數(shù)據(jù)湖架構(gòu)或者云原生架構(gòu)一般采用存算分離，存算分離的好處是根據(jù)功能的不同最優(yōu)化節(jié)點(diǎn)（機(jī)器）配置，能更好地適應(yīng)不同的計算負(fù)載和存儲需求，提高資源利用率。如果計算過程中需要大容量磁盤存儲中間數(shù)據(jù)，就難以存算分離。第二，難以及時縮容，像 Spark 計算引擎，作業(yè)在運(yùn)行過程中可以動態(tài)資源伸縮，Executor 處于 idle 一段時間就可以釋放，但是如果本地磁盤存儲了下游所需的中間數(shù)據(jù)，那么就無法釋放節(jié)點(diǎn)，特別是 Spark on K8s 的場景。

2、 Celeborn 的前世今生

Celeborn 的定位是通過接管中間數(shù)據(jù)來解決上述問題。在具體介紹 Celeborn 之前，先來看一下其發(fā)展歷程。

Celeborn 最早于 2020 年在阿里云內(nèi)部開發(fā)。
2021 年 12 月正式對外開源，趣頭條和小米成為種子用戶。
2022 年 10 月發(fā)布 0.1 系列，實現(xiàn)生產(chǎn)可用，支持 Spark，使用本地盤存儲，采用 Standalone 的方式部署。同年，正式捐贈給了Apache 軟件基金會（ASF），并更名為 Celeborn，正式進(jìn)入 Apache 孵化。
2023 年 3 月發(fā)布 0.2 系列，是捐贈后的首個版本，支持 K8s 部署，提升了穩(wěn)定性和性能。
2023 年 7 月發(fā)布 0.3.0，支持 Flink batch，也支持 Native Spark，在存儲層支持把數(shù)據(jù)存儲在 HDFS，并支持優(yōu)雅升級。
2023 年 10 月 13 日，發(fā)布了 0.3.1。

截至目前，參與貢獻(xiàn)和使用 Celeborn 的企業(yè)來自各個行業(yè)和海內(nèi)外。

3、整體架構(gòu)

下圖是 Celeborn 的主要架構(gòu)，其中藍(lán)色部分是 Celeborn 的組件。

從上圖可以看出 Apache Celeborn 是一個 Server-Client 的架構(gòu)。

服務(wù)端包括 Master 和 Worker 節(jié)點(diǎn)：

Master 的主要職責(zé)是：管理整個集群的狀態(tài)；負(fù)責(zé)負(fù)載的分配；同時基于 Raft 實現(xiàn)高可用。

Worker 的主要職責(zé)是：接收、存儲和服務(wù) shuffle 數(shù)據(jù)。實現(xiàn)了多層存儲，目前已經(jīng)支持 Local Disks 和 DFS ，Memory 正在開發(fā)中。

Spark和Flink計算引擎分兩種角色：一是Driver/JobMaster，負(fù)責(zé)整個Application 生命周期的管理、作業(yè)的調(diào)度；另一個是 Executor/TaskManager，實現(xiàn)真正的分布式計算執(zhí)行。

相對應(yīng)的，Celeborn 也分兩部分，一個是 Lifecycle Manager，另一個是 Shuffle Client。Lifecycle Manager 的職責(zé)是管理當(dāng)前作業(yè)的 shuffle metadata，把 shuffle 元數(shù)據(jù)從 Master 轉(zhuǎn)移到 Celeborn Application，讓 Application 自己管理自己的 shuffle，從而大幅降低 Master 的負(fù)載。Shuffle Client 存在于每一個Executor 或 TaskManager 上，負(fù)責(zé)具體推送和讀取 shuffle 數(shù)據(jù)。

Celeborn 的 Master、Worker、Lifecycle Manager 和 Shuffle Client 是與引擎無關(guān)的，沒有引入任何引擎的依賴。各計算引擎通過 Shuffle Client 的 API 來集成 Celeborn。

目前，Celeborn 社區(qū)官方已集成 Spark、Flink 和 MapReduce 三種計算引擎。另外，有公司基于 Shuffle Client 的 API 實現(xiàn)了 MR3 引擎的集成。

從上圖中可以看到，shuffle 數(shù)據(jù)不再存儲于本地磁盤，而是存儲到 Celeborn cluster 上，這就解除了計算節(jié)點(diǎn)對本地磁盤的依賴。

而穩(wěn)定性和性能問題，Celeborn 是通過 Partition 的聚合來解決的。

4、核心設(shè)計

MapTask在推送數(shù)據(jù)時，把屬于同一個Partition的數(shù)據(jù)都推給同一個 CelebornWorker，Worker 把接收到的數(shù)據(jù)聚合之后寫入磁盤，這樣一個 Reducer 只需要從一個 Worker 上讀取相應(yīng)的 Partition 文件即可。經(jīng)過聚合后在shuffle read 階段，網(wǎng)絡(luò)連接數(shù)變成了 n，單個文件被讀取數(shù)就變成了 1，從而很好地解決了磁盤 IOPS 的問題。單個 Partition 文件過大時，可能會對磁盤造成過大的壓力，系統(tǒng)會檢測每個 Partition 文件的大小，如果超過預(yù)設(shè)的閾值，會對 Partition 做切分，Partiton 切分相關(guān)信息會存儲到 Lifecycle Manager 上，不會造成 Reduce 丟失數(shù)據(jù)。

下圖展示了使用 Celeborn 之后 shuffle 的生命周期。

首先，MapTask 在第一次需要寫 shuffle 數(shù)據(jù)時，向 Lifecycle Manager 發(fā)起 RegisterShuffle 請求，Lifecycle Manager 向 Celeborn Master 發(fā)起RequestSlots 請求，Master 選擇一部分 Worker 為當(dāng)前的 shuffle 請求服務(wù)。Lifecycle Manager 向這些 Worker 發(fā)起 ReserveSlots 請求，這些 Worker 會做相應(yīng)的準(zhǔn)備工作。接下來，Lifecycle Manager 會把 PartitionLocations 信息返回給 MapTask，MapTask 就可以持續(xù)往這些 Worker 推送 shuffle 數(shù)據(jù)。Worker 接收到 shuffle 數(shù)據(jù)后在內(nèi)存中緩存，如果開啟了多副本，會在兩個 Worker 做數(shù)據(jù)冗余。單個 Partition 數(shù)據(jù)在內(nèi)存中超過默認(rèn)的 256k 時，就會 Append 到 Partition 文件中。每個 MapTask 結(jié)束時會向 Lifecycle Manager 發(fā)送 MapperEnd，當(dāng)所有 MapTask 都結(jié)束之后，Lifecycle Manager 向所有服務(wù)本次shuffle 的 Worker 發(fā)送 Commit files 請求，Worker 會把尚未 flush 的內(nèi)存數(shù)據(jù) flush 到磁盤。至此，就完成了整個 shuffle write 的過程。

ReduceTask 在啟動時向 Lifecycle Manager 獲取屬于自己的 PartitionLocations 信息，然后向相應(yīng)的 Worker 讀取數(shù)據(jù)。

Celeborn 的另一個核心設(shè)計點(diǎn)是容錯和保證精準(zhǔn)一次。

Fault tolerance

Celeborn 中最高頻的是 Push data，當(dāng)其失敗時，Celeborn 不會認(rèn)為這個Worker 失敗了，之前推送的數(shù)據(jù)就丟失了，而是認(rèn)為當(dāng)前推送臨時出問題，后面的推送會嘗試申請新的 Worker，繼續(xù)往新 Worker 推送。檢測數(shù)據(jù)是否丟失則是在Commit files 階段完成，Lifecycle Manager 既往之前失敗的 Worker 發(fā)送 commit 請求、也會往新的 Worker 上發(fā)送 commit 請求，只有當(dāng)兩個 Pair 至少都有一個副本 commit 成功，才會判定當(dāng)前數(shù)據(jù)沒有丟失。

Exactly once

另一個問題是，假設(shè) Client 認(rèn)為數(shù)據(jù)推送失敗了，重新申請 Worker 推送數(shù)據(jù)，但實際上 Worker 成功接收數(shù)據(jù)，并且 commit 成功了。這時需要一個機(jī)制來保證精準(zhǔn)一次推送，Celeborn 是通過在每一個推送數(shù)據(jù)的 batch 上添加三元組 header（MapId、AttemptId和BatchId）保證不會有數(shù)據(jù)重復(fù)。Reducer 在讀取數(shù)據(jù)時會過濾掉非 successful attempt id 的數(shù)據(jù)和重復(fù)的 batch id 數(shù)據(jù)。

ShuffleClient 提供了 API 幫助計算引擎集成 Celeborn，主要包括以下四個 API：

（1）pushData

registerShuffle 會包含在 pushData 的實現(xiàn)里，引擎只需要創(chuàng)建 ShuffleClient 實例，調(diào)用 pushData 即可。

（2）mapperEnd

shuffle write 階段，除了 pushData，還需一個 mapperEnd API，告知 Celeborn shuffle write 完成。

（3）readPartition

在 shuffle read 階段，只有一個 readPartition 接口，Celeborn 會返回一個InputStream 對象，計算引擎接收這個 InputStream 完成后續(xù)工作。

（4）unregisterShuffle

當(dāng)計算引擎認(rèn)為一個 shuffle 已經(jīng)完全結(jié)束，后續(xù)也不會再使用，可以調(diào)用 unregisterShuffle 通知 Celeborn 該 shuffle 生命周期結(jié)束。

除了上面介紹的一些核心設(shè)計，Celeborn 還具有如下特性，這里不做展開介紹。

滾動升級：升級時不影響當(dāng)前正在運(yùn)行的作業(yè)，下線時也不需要等待長尾作業(yè)的完成。
優(yōu)雅下線
多租戶
負(fù)載均衡
MapPartition
流量控制
Spark AQE
列式 shuffle

5、效果評估

下圖是 Celeborn 0.3.0 發(fā)布時做的一個純 shuffle 作業(yè)的對比測試結(jié)果，對比的是 Spark 原生的 External Shuffle Service (ESS) 和 Celeborn 0.2.1 版本。

可以看到 Celeborn 在性能上優(yōu)于 ESS，且規(guī)模越大，優(yōu)勢越明顯。同時 0.3.0 版本比 0.2.1 版本的性能也有提升。

對比 ESS，Celeborn 的優(yōu)勢主要在 shuffle read 階段，這與前面介紹的穩(wěn)定性和性能主要在于 shuffle read 時的非常低效的 IO 模式一致。shuffle write 階段，Celeborn 需要通過一層網(wǎng)絡(luò)，而 ESS 直接寫本地文件。但 Celeborn 最新版本在 shuffle write 過程相對于 ESS 并沒有明顯的性能降低。

二、Celeborn 的社區(qū)/用戶/生產(chǎn)介紹

1、Celeborn 多元社區(qū)

Celeborn 于 2022 年 10 月捐贈給 Apache 軟件基金會，隨后（2022-10-18）進(jìn)入孵化器，經(jīng)過一年的發(fā)展，Celeborn 已經(jīng)變成一個多元的社區(qū)。

目前 Celeborn 社區(qū)擁有：

7 位 PPMC，除了阿里，還有兩位分別來自網(wǎng)易和 Shopee。一位是來自 Shopee的朱夷 @Angerszhuuuu，是第一個大規(guī)模使用 Celeborn 的用戶，踩了很多坑，對 Celeborn 的發(fā)展做出了很多貢獻(xiàn)；另一位是來自網(wǎng)易的潘成(Cheng Pan)，也是一位非常重要的 PMC。后面還會有更多 PMC 加入 Celeborn 社區(qū)。
6 位 Committers，其中三位來自阿里之外。
72 位 Contributors。

2、Celeborn 用戶場景

目前用戶使用 Celeborn 的方式有三種：混部、獨(dú)立部署、完全存算分離。

混部：把 Celeborn 集群跟現(xiàn)有的 HDFS/YARN 集群部署在一起。這種一般是已經(jīng)存在 HADOOP 集群，又不想或基于成本考慮不能對集群進(jìn)行升級改造，則可以把 Celeborn 直接部署在原集群上。
Celeborn 獨(dú)立部署：這種方式的好處是可以把 shuffle 相關(guān)的 IO 跟 HDFS 的 IO 隔離開來，能夠有效提升穩(wěn)定性。
完全存算分離：這種方式是把計算集群、Celeborn 集群、存儲集群完全分開，三方集群都可以很方便地進(jìn)行擴(kuò)展，能得到很好的性價比。

3、Celeborn 生產(chǎn)場景

下圖展示了阿里和其他用戶的 Celeborn 真實生產(chǎn)場景的情況，一些大用戶的 Celeborn 集群每天服務(wù)的 shuffle 數(shù)據(jù)總量達(dá)到若干個 PB、數(shù)萬個作業(yè)，單 Shuffle 數(shù)據(jù)達(dá)到數(shù)百 T。

4、 Celeborn 用戶反饋

下面摘錄一些正向的用戶反饋。

當(dāng)然也有一些問題反饋，針對問題反饋，社區(qū)會積極解決。

5、Celeborn 與其他社區(qū)

（1）Kyuubi + Celeborn

接下來介紹 Celeborn 跟其他社區(qū)或項目的結(jié)合情況。

首先是 Kyuubi 和 Celeborn 的結(jié)合使用。Kyuubi 是網(wǎng)易開源的一個很優(yōu)秀的開源項目。阿里云的 EMR 產(chǎn)品較早的時候就引入了 Kyuubi，用戶在創(chuàng)建 EMR 集群時可以直接選擇 Kyuubi 組件。

同時，網(wǎng)易通過 Kyuubi + Celeborn 在內(nèi)部一些生產(chǎn)場景有效提升了性價比。

Kyuubi 社區(qū)和 Celeborn 社區(qū)聯(lián)系緊密。Kyuubi 的一些 PM 和 Committers 也在為 Celeborn 社區(qū)做貢獻(xiàn)。雙方都希望 Kyuubi 聯(lián)合 Celeborn 將來能做出更廣為大眾接受的云原生方案。

（2）Gluten + Celeborn

Spark 社區(qū)近年來一直希望能做出一個 Photon 的翻版組件。Databricks 的 Photon 是一個高效的 Native SQL 執(zhí)行引擎，不開源。

Gluten 就是做這樣一件事情的產(chǎn)品，它將 Velox、ClickHouse 等 Native 的 SQL 執(zhí)行引擎通過友好的方式集成到 Spark 里面。

Gluten 還提供了高效的 Columnar Shuffle 機(jī)制，下圖是 Gluten Columnar Shuffle 的示意圖。 Gluten 采用高效的設(shè)計讓 RecordBatch 數(shù)據(jù)結(jié)構(gòu)有比較高效的 shuffle 性能，但是其在 shuffle 的數(shù)據(jù)傳輸上沿用了 Spark External Shuffle Service 的方式，因此也存在前面介紹的各種問題。

為解決 shuffle 數(shù)據(jù)上的問題，過去一段時間，Gluten 社區(qū)和 Celeborn 社區(qū)做了很好的聯(lián)合，把 Celeborn 集成到了 Gluten 里面。

編譯 Gluten 時指定 RSS Profile，就會默認(rèn)使用 Celeborn 的 Client 接管 shuffle 數(shù)據(jù)。實現(xiàn)方式是在 Gluten 的 Native Partitioner 引入了 Celeborn 的 SDK，在推送和讀取 shuffle 數(shù)據(jù)時，通過 Celeborn SDK 連接 Celeborn Cluster。要說明的一點(diǎn)是，Celeborn 的引入與 Gluten 本身的 Native Partitoner 是正交關(guān)系，并不是替代關(guān)系。