網(wǎng)易傳媒大數(shù)據(jù)實(shí)際業(yè)務(wù)中，存在著大量的準(zhǔn)實(shí)時計(jì)算需求場景，業(yè)務(wù)方對于數(shù)據(jù)的實(shí)效性要求一般是分鐘級；這種場景下，用傳統(tǒng)的離線數(shù)倉方案不能滿足用戶在實(shí)效性方面的要求，而使用全鏈路的實(shí)時計(jì)算方案又會帶來較高的資源占用。

基于對開源數(shù)據(jù)湖方案的調(diào)研，我們注意到了網(wǎng)易數(shù)帆開源的基于 Apache Iceberg 構(gòu)建的 Arctic 數(shù)據(jù)湖解決方案。Arctic 能相對較好地支持與服務(wù)于流批混用的場景，其開放的疊加式架構(gòu)，可以幫助我們非常平滑地過渡與實(shí)現(xiàn) Hive 到數(shù)據(jù)湖的升級改造，且由于傳媒離線數(shù)倉已接入有數(shù)，通過 Arctic 來改造現(xiàn)有業(yè)務(wù)的成本較低，于是我們準(zhǔn)備通過引入 Arctic ，嘗試解決 push 業(yè)務(wù)場景下的痛點(diǎn)。

1

項(xiàng)目背景

以傳媒 push 實(shí)時數(shù)倉為例，新聞推送在地域、時間、頻次等因素上有較高的不確定性，非常容易出現(xiàn)偶發(fā)的流量洪峰，尤其是在出現(xiàn)突發(fā)性社會熱點(diǎn)新聞的時候。如果采用全鏈路的實(shí)時計(jì)算方案來處理，則需要預(yù)留出較多的資源 buffer 來應(yīng)對。

由于推送時機(jī)的不確定性，push 業(yè)務(wù)的數(shù)據(jù)指標(biāo)一般不是增量型的，而是以當(dāng)天截止到當(dāng)前的各種累計(jì)型指標(biāo)為主，計(jì)算窗口通常為十五分鐘到半小時不等，統(tǒng)計(jì)維度區(qū)分發(fā)送類型、內(nèi)容分類、發(fā)送票數(shù)、發(fā)送廠商、首啟方式、用戶活躍度、AB 實(shí)驗(yàn)等，具有流量波動大和數(shù)據(jù)口徑繁多等特點(diǎn)。

此前采用的全鏈路 Flink 實(shí)時計(jì)算方案中，主要遇到以下問題：

（1）資源占用成本高

為應(yīng)對流量洪峰，需要為實(shí)時任務(wù)分配預(yù)留出較高的資源，且多個聚合任務(wù)需要消費(fèi)同一個上游數(shù)據(jù)，存在讀放大問題。push 相關(guān)的實(shí)時計(jì)算流程占到了實(shí)時任務(wù)總量的 18+%，而資源使用量占到了實(shí)時資源總使用量的近 25%。

（2）大狀態(tài)帶來的任務(wù)穩(wěn)定性下降

push 業(yè)務(wù)場景下進(jìn)行窗口計(jì)算時，大流量會帶來大狀態(tài)的問題，而大狀態(tài)的維護(hù)在造成資源開支的同時比較容易影響任務(wù)的穩(wěn)定性。

（3）任務(wù)異常時難以及時的進(jìn)行數(shù)據(jù)修復(fù)

實(shí)時任務(wù)出現(xiàn)異常時，以實(shí)時方式來回溯數(shù)據(jù)時效慢且流程復(fù)雜；而以離線流程來修正，則會帶來雙倍的人力和存儲成本。

2

項(xiàng)目思路和方案

2.1項(xiàng)目思路

我們通過對數(shù)據(jù)湖的調(diào)研，期望利用數(shù)據(jù)實(shí)時入湖的特點(diǎn)，同時使用 Spark 等離線資源完成計(jì)算，用較低的成本滿足業(yè)務(wù)上對準(zhǔn)實(shí)時計(jì)算場景的需求。我們以 push 業(yè)務(wù)場景作為試點(diǎn)進(jìn)行方案的探索落地，再逐漸將方案推廣至更多類似業(yè)務(wù)場景。

基于對開源數(shù)據(jù)湖方案的調(diào)研，我們注意到了網(wǎng)易數(shù)帆開源的基于 Apache Iceberg 構(gòu)建的 Arctic 數(shù)據(jù)湖解決方案。Arctic 能相對較好地支持與服務(wù)于流批混用的場景，其開放的疊加式架構(gòu)，可以幫助我們非常平滑地過渡與實(shí)現(xiàn) Hive 到數(shù)據(jù)湖的升級改造，且由于傳媒離線數(shù)倉已接入有數(shù)，通過 Arctic 來改造現(xiàn)有業(yè)務(wù)的成本較低，于是我們準(zhǔn)備通過引入 Arctic ，嘗試解決 push 業(yè)務(wù)場景下的痛點(diǎn)。

Arctic 是由網(wǎng)易數(shù)帆開源的流式湖倉系統(tǒng)，在 Iceberg 和 Hive 之上添加了更多實(shí)時場景的能力。通過 Arctic，用戶可以在 Flink、Spark、Trino、Impala 等引擎上實(shí)現(xiàn)更加優(yōu)化的 CDC、流式更新、OLAP 等功能。

實(shí)現(xiàn) push 業(yè)務(wù)場景下的數(shù)據(jù)湖改造，只需要使用 Arctic 提供的 Flink Connector，便可快速地實(shí)現(xiàn) push 明細(xì)數(shù)據(jù)的實(shí)時入湖。

此時需要我們關(guān)注的重點(diǎn)是，數(shù)據(jù)產(chǎn)出需要滿足分鐘級業(yè)務(wù)需求。數(shù)據(jù)產(chǎn)出延遲由兩部分組成：

數(shù)據(jù)就緒延遲，取決于 Flink 實(shí)時任務(wù)的 Commit 間隔，一般為分鐘級別；

數(shù)據(jù)計(jì)算耗時，取決于計(jì)算引擎和業(yè)務(wù)邏輯：數(shù)據(jù)產(chǎn)出延遲 = 數(shù)據(jù)就緒延遲 + 數(shù)據(jù)計(jì)算耗時

2.2 解決方案

2.2.1 數(shù)據(jù)實(shí)時入湖

Arctic 能夠兼容已有的存儲介質(zhì)（如 HDFS）和表結(jié)構(gòu)（如 Hive、Iceberg），并在之上提供透明的流批一體表服務(wù)。存儲結(jié)構(gòu)上主要為 Basestore 和 Changestore 兩部分：

（1）Basestore 中存儲了表的存量數(shù)據(jù)。它通常由 Spark/Flink 等引擎完成第一次寫入，再之后則通過自動的結(jié)構(gòu)優(yōu)化過程將 Changestore 中的數(shù)據(jù)轉(zhuǎn)化之后寫入。

（2）Changestore 中存儲了表上最近的變更數(shù)據(jù)。Changestore 中存儲了表上最近的變更數(shù)據(jù)。它通常由 Apache Flink 任務(wù)實(shí)時寫入，并用于下游 Flink 任務(wù)進(jìn)行準(zhǔn)實(shí)時的流式消費(fèi)。同時也可以對它直接進(jìn)行批量計(jì)算或聯(lián)合 Basestore 里的數(shù)據(jù)一起通過 Merge-On-Read(以下簡稱為MOR) 的查詢方式提供分鐘級延遲的批量查詢能力。

Arctic 表支持實(shí)時數(shù)據(jù)的流式寫入，數(shù)據(jù)寫入過程中為了保證數(shù)據(jù)的實(shí)效性，寫入側(cè)需要頻繁的進(jìn)行數(shù)據(jù)提交，但因此會產(chǎn)生大量的小文件，積壓的小文件一方面會影響數(shù)據(jù)的查詢性能，另一方面也會對文件系統(tǒng)帶來壓力。這方面，Arctic 支持基于主鍵的行級更新，提供了 Optimizer 來進(jìn)行數(shù)據(jù) Update 和自動的結(jié)構(gòu)優(yōu)化，以幫助用戶解決數(shù)據(jù)湖常見的小文件、讀放大、寫放大等問題。

以傳媒 push 數(shù)倉場景為例，push 發(fā)送、送達(dá)、點(diǎn)擊、展示等明細(xì)數(shù)據(jù)需要通過 Flink 作業(yè)實(shí)時寫入到 Arctic 中。由于上游已經(jīng)做了 ETL 清洗，此階段只需要通過 FlinkSQL 即可方便地將上游數(shù)據(jù)寫入 Changestore。Changestore 內(nèi)包含了存儲插入數(shù)據(jù)的 insert 文件和存儲刪除數(shù)據(jù)的 equality delete 文件，更新數(shù)據(jù)會被拆分為更新前項(xiàng)和更新后項(xiàng)分別存儲在 delete 文件與 insert 文件中。

具體的，對于有主鍵場景，insert/update_after 消息會寫入 Changestore 的 insert 文件，delete/update_before 會寫入 Arctic 的 delete 文件。當(dāng)進(jìn)行 Optimize 的時候，會先把 delete 文件讀到內(nèi)存中形成一個 delete map, map 的 key 是記錄的主鍵，value 是 record_lsn。然后 再讀取 Basestore 和 Changestore 中的 insert 文件， 對主鍵相同的 row 進(jìn)行 record_lsn 的對比，如果 insert 記錄中 record_lsn 比 deletemap 中相同主鍵的 record_lsn 小，則認(rèn)為這條記錄已經(jīng)被刪除了，不會再追加到 base 里；否則把數(shù)據(jù)寫入到新文件里，最終實(shí)現(xiàn)了行級的更新。

2.2.2 湖水位感知

傳統(tǒng)的離線計(jì)算在調(diào)度方面需要有一個觸發(fā)機(jī)制，一般由作業(yè)調(diào)度系統(tǒng)按照任務(wù)之間的依賴關(guān)系來處理，當(dāng)上游任務(wù)全部成功后自動調(diào)起下游的任務(wù)。但在實(shí)時入湖的場景下，下游任務(wù)缺乏一個感知數(shù)據(jù)是否就緒的途徑。以 push 場景為例，需要產(chǎn)出的指標(biāo)主要為按照指定的時間粒度來計(jì)算一次當(dāng)天累計(jì)的各種統(tǒng)計(jì)值，此時下游如果沒法感知當(dāng)前湖表水位的話，要么需要留出一個較冗余的緩沖時間來保證數(shù)據(jù)就緒，要么則有漏數(shù)據(jù)的可能，畢竟 push 場景的流量變化是非常起伏不定的。

傳媒大數(shù)據(jù)團(tuán)隊(duì)和 Arctic 團(tuán)隊(duì)借鑒了 Flink Watermark 的處理機(jī)制和 Iceberg 社區(qū)討論的方案，將 Watermark 信息寫入到 Iceberg 表的 metadata 文件里，然后由 Arctic 通過消息隊(duì)列或者 API 暴露出來，從而做到下游任務(wù)的主動感知，盡可能地降低了啟動延遲。具體方案如下：

（1）Arctic 表水位感知

當(dāng)前只考慮 Flink 寫入的場景，業(yè)務(wù)在 Flink 的 source 定義事件時間和 Watermark。ArcticSinkConnector 包含兩個算子，一個是負(fù)責(zé)寫文件的多并發(fā)的 ArcticWriter, 一個是負(fù)責(zé)提交文件的的單并發(fā)的 ArcticFileCommitter。當(dāng)執(zhí)行 checkpoint 時，ArcticFileCommitter 算子會進(jìn)行 Watermark 對齊之后取最小的 Watermark。會新建一個類似于 Iceberg 事務(wù)的 AMS Transaction，在這個事務(wù)里除了 AppendFiles 到 Iceberg，同時把 TransactionID，以及 Watermark 通過 AMS 的 thrift 接口上報(bào)給 AMS。

（2）Hive 表水位感知

Hive表里可見的數(shù)據(jù)是經(jīng)過 Optimize 過后的數(shù)據(jù)，Optimize 由 AMS 來調(diào)度，F(xiàn)link 任務(wù)異常執(zhí)行文件的讀寫合并，并且把 Metric 上報(bào)給 AMS， 由 AMS 來把這一次 Optimize 執(zhí)行的結(jié)果 Commit，AMS 天然知道這一次 Optimize 推進(jìn)到了哪次 Transaction， 并且 AMS 本身也存儲了 Transaction 對應(yīng)的 Watermark，也就知道 Hive 表水位推進(jìn)到了哪里。

2.2.3 數(shù)據(jù)湖查詢

Arctic 提供了 Spark/Flink/Trino/Impala 等計(jì)算引擎的 Connector 支持。通過使用Arctic數(shù)據(jù)源，各計(jì)算引擎都可以實(shí)時讀取到已經(jīng) Commit 的文件，Commit 的間隔按照業(yè)務(wù)的需求一般為分鐘級別。下面以 push 業(yè)務(wù)為例介紹幾種場景下的查詢方案和相應(yīng)成本：

（1）Arctic + Trino/Impala 滿足秒級 OLAP 查詢

OLAP 場景下，用戶一般更關(guān)注計(jì)算上的耗時，對數(shù)據(jù)就緒的敏感度相對不高。針對中小規(guī)模數(shù)據(jù)量的 Arctic 表或較簡單的查詢，通過 Trino/Impala 進(jìn)行 OLAP 查詢是一個相對高效的方案，基本上可以做到秒級 MOR 查詢耗時。成本上，需要搭建 Trino/Impala 集群，如果團(tuán)隊(duì)中已有在使用的話，則可以根據(jù)負(fù)載情況考慮復(fù)用。

Arctic 在開源發(fā)布會上發(fā)布了自己的 benchmark 數(shù)據(jù)，在數(shù)據(jù)庫 CDC 持續(xù)流式攝取的場景下，對比各個數(shù)據(jù)湖 Format 的 OLAP benchmark 性能, 整體上帶 Optimize 的 Arctic 的性能優(yōu)于 Hudi，這主要得益于 Arctic 內(nèi)部有一套高效的文件索引 Arctic Tree，在 MOR 場景下可以做到更細(xì)粒度、精確地 merge。詳細(xì)的對比報(bào)告可以參考：https://arctic.netease.com/ch/benchmark/。

（2）Arctic + Spark 滿足分鐘級預(yù)聚合查詢

針對提供下游數(shù)據(jù)報(bào)表展示的場景，一般需要走預(yù)計(jì)算的流程將結(jié)果持久化下來，對數(shù)據(jù)就緒和計(jì)算耗時的敏感度都較高，而且查詢邏輯相對復(fù)雜，Trino/Impala 集群規(guī)模相對較小，執(zhí)行容易失敗，導(dǎo)致穩(wěn)定性欠佳。這個場景下我們使用了集群部署規(guī)模最大的 Spark 引擎來處理，在不引入新的資源成本的情況下，做到了離線計(jì)算資源的復(fù)用。

數(shù)據(jù)就緒方面，通過 Arctic 表水位感知方案，可以做到較低的分鐘級就緒延遲。

計(jì)算方面，Arctic 對 Spark Connector 提供了一些讀取優(yōu)化，用戶可以通過配置 Arctic 表的 read.split.planning-parallelism 和 read.split.planning-parallelism-factor 這兩個參數(shù)值，來調(diào)整 Arctic Combine Task 的數(shù)量，進(jìn)而控制計(jì)算任務(wù)的并發(fā)度。由于 Spark 離線計(jì)算的資源相對靈活和充足，我們可以通過上述調(diào)整并發(fā)度的方式來保證在 2~3 分鐘內(nèi)完成業(yè)務(wù)的計(jì)算需求。

（3）Hive + Spark 滿足傳統(tǒng)離線數(shù)倉生產(chǎn)鏈路的調(diào)度

Arctic 支持將 Hive 表作為 Basestore，F(xiàn)ull Optimize 時會將文件寫入到 Hive 數(shù)據(jù)目錄下，以達(dá)到更新 Hive 原生讀取內(nèi)容的目的，通過存儲架構(gòu)上的流批一體來降低成本。因此傳統(tǒng)的離線數(shù)倉生產(chǎn)鏈路，可以直接使用對應(yīng)的 Hive 表來作為離線數(shù)倉鏈路的一部分，時效性上相較于 Arctic 表雖缺少了 MOR，但通過 Hive 表水位感知方案，可以做到業(yè)務(wù)能接受的就緒延遲，從而滿足傳統(tǒng)離線數(shù)倉生產(chǎn)鏈路的調(diào)度。

3

項(xiàng)目影響力與產(chǎn)出價(jià)值

3.1 項(xiàng)目影響力

通過 Arctic + X 方案在傳媒的探索和落地，為傳媒準(zhǔn)實(shí)時計(jì)算場景提供了一個新的解決思路。該思路不但減輕了全鏈路 Flink 實(shí)時計(jì)算方案所帶來的實(shí)時資源壓力和開發(fā)運(yùn)維負(fù)擔(dān)，而且還能較好地復(fù)用現(xiàn)有的 HDFS 和 Spark 等存儲計(jì)算資源，做到了降本增效。

此外 Arctic 在音樂、有道等多個 BU 也有落地，比如在音樂公技，用于 ES 冷數(shù)據(jù)的存儲，降低了用戶 ES 的存儲成本；而有道精品課研發(fā)團(tuán)隊(duì)也在積極探索和使用 Arctic 作為其部分業(yè)務(wù)場景下的解決方案。

目前 Arctic 已經(jīng)在 github 上開源，受到了開源社區(qū)與外部用戶的持續(xù)關(guān)注，在 Arctic 的建設(shè)與發(fā)展中，也收到了不少外部用戶提交的高質(zhì)量 PR 。

3.2 項(xiàng)目產(chǎn)出價(jià)值

通過上述方案我們將 push ETL 明細(xì)數(shù)據(jù)通過 Flink 實(shí)時入湖到 Arctic，然后在調(diào)度平臺上配置分鐘級的調(diào)度任務(wù)，按照不同交叉維度進(jìn)行計(jì)算后將累計(jì)型指標(biāo)后寫入關(guān)系數(shù)據(jù)庫，最后通過有數(shù)直連進(jìn)行數(shù)據(jù)展示，做到了業(yè)務(wù)方要求的分鐘級時效數(shù)據(jù)產(chǎn)出。改造后的方案，同原來的全鏈路 Flink 實(shí)時計(jì)算方案相比：

（1）充分復(fù)用離線空閑算力，降低了實(shí)時計(jì)算資源開支

方案利用了空閑狀態(tài)下的離線計(jì)算資源，且基本不會帶來新的資源開支。離線計(jì)算業(yè)務(wù)場景注定了資源使用的高峰在凌晨，而新聞 push 推送及熱點(diǎn)新聞產(chǎn)生的場景大多為非凌晨時段，在滿足準(zhǔn)實(shí)時計(jì)算時效的前提下，通過復(fù)用提升了離線計(jì)算集群的綜合利用率。另外，該方案能幫我們釋放大約 2.4T 左右的實(shí)時計(jì)算內(nèi)存資源。

（2）降低任務(wù)維護(hù)成本，提升任務(wù)穩(wěn)定性

Arctic + Spark 水位感知觸發(fā)調(diào)度的方案可減少 17+ 實(shí)時任務(wù)的維護(hù)成本，減少了 Flink 實(shí)時計(jì)算任務(wù)大狀態(tài)所帶來的穩(wěn)定性問題。通過 Spark 離線調(diào)度任務(wù)可充分利用離線資源池調(diào)整計(jì)算并行度，有效提升了應(yīng)對突發(fā)熱點(diǎn)新聞流量洪峰時的健壯性。

（3）提升數(shù)據(jù)異常時的修復(fù)能力，降低數(shù)據(jù)修復(fù)時間開支

通過流批一體的 Arctic 數(shù)據(jù)湖存儲架構(gòu)，當(dāng)數(shù)據(jù)出現(xiàn)異常需要修正時，可靈活地對異常數(shù)據(jù)進(jìn)行修復(fù)，降低修正成本；而如果通過實(shí)時計(jì)算鏈路回溯數(shù)據(jù)或通過額外的離線流程來修正，則需要重新進(jìn)行狀態(tài)累計(jì)或復(fù)雜的 ETL 流程。

4

項(xiàng)目未來規(guī)劃和展望

當(dāng)前還有一些場景 Arctic 不能做到較好的支持，傳媒大數(shù)據(jù)團(tuán)隊(duì)將和 Arctic 團(tuán)隊(duì)繼續(xù)對以下場景下的解決方案進(jìn)行探索和落地：

（1）當(dāng)前入湖前的 push 明細(xì)數(shù)據(jù)是通過上游多條數(shù)據(jù)流 join 生成的，也同樣會存在大狀態(tài)的問題。而 Arctic 當(dāng)前只能支持行級的更新能力，如果能落地有主鍵表的部分列更新能力，則可以幫助業(yè)務(wù)在入湖的時候，以較低的成本直接實(shí)現(xiàn)多流 join。

（2）進(jìn)一步完善 Arctic 表和 Hive 表的水位定義和感知方案，提升時效，并推廣到更多的業(yè)務(wù)場景中。當(dāng)前的方案只支持單 Spark/Flink 任務(wù)寫入的場景，對于多個任務(wù)并發(fā)寫表的場景，還需要再完善。

審核編輯 ：李倩