一、Celeborn 是什么

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

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

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

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

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

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

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

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

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

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

2、 Celeborn 的前世今生

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

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

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

3、整體架構(gòu)

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

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

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

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

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

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

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

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

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

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

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

4、核心設(shè)計(jì)

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

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

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

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

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

Fault tolerance

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

Exactly once

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

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

（1）pushData

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

（2）mapperEnd

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

（3）readPartition

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

（4）unregisterShuffle

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

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

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

5、效果評(píng)估

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

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

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

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

1、Celeborn 多元社區(qū)

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

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

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

2、Celeborn 用戶場(chǎng)景

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

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

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

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

4、 Celeborn 用戶反饋

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

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

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

（1）Kyuubi + Celeborn

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

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

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

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

（2）Gluten + Celeborn

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

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

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

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

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

（3）MR3 + Celeborn

MR3 是一個(gè)韓國教授團(tuán)隊(duì)通過 API 自發(fā)的集成 Celeborn，社區(qū)也幫助 MR3 團(tuán)隊(duì)解決了很多問題。MR3 + Celeborn 已完成第一個(gè) Release 版本的發(fā)布。

三、Celeborn 未來發(fā)展規(guī)劃

Celeborn 未來的發(fā)展規(guī)劃包括三個(gè)方向。

1、社區(qū)

希望有更多 PPMC、Committer 和 Contributor 能夠加入社區(qū)，大家一起把社區(qū)做好，也希望社區(qū)更多元化，有更多公司的同事加入社區(qū)。

大家可以順手點(diǎn)個(gè) star https://github.com/apache/incubator-celeborn。

2、用戶

持續(xù)推廣用戶，特別是希望有機(jī)會(huì)推廣到海外。

3、Feature 規(guī)劃

• 實(shí)現(xiàn) Spill / Cache Data
• 對(duì)接更多的引擎，比如：Tez/Trino/Ray/...
• 繼續(xù)完善多層存儲(chǔ)
• 實(shí)現(xiàn)認(rèn)證和安全隔離，Linkedin 團(tuán)隊(duì)已在開發(fā)中
• Stage Rerun（螞蟻團(tuán)隊(duì)）
• 更好的AQE支持
• 實(shí)現(xiàn) C++ SDK，使得在集成 Ray 和 Native 時(shí)更加順暢

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