摘要：本文整理自阿里云智能 Flink SQL和數(shù)據(jù)通道負責人、Apache Flink PMC 伍翀（花名：云邪）老師，在 Flink Forward Asia 2024 主會場的分享。主要分享了一種專為流分析設(shè)計的新一代存儲解決方案——Fluss，并由阿里巴巴開源委員會副主席王峰先生，在 FFA 2024 現(xiàn)場進行了 Fluss 項目的開源。內(nèi)容分為以下五個部分：
一、Kafka 在實時分析場景遇到的問題
二、Fluss：Flink Unified Streaming Storage
三、Fluss 核心特性
四、Fluss 未來規(guī)劃
五、Fluss 開源
當前業(yè)界呈現(xiàn)出一個顯著的趨勢，即大數(shù)據(jù)的處理正在從離線模式轉(zhuǎn)向?qū)崟r化。我們可以觀察到，多個行業(yè)和應(yīng)用場景都在進行實時化的演進。例如，互聯(lián)網(wǎng)、車聯(lián)網(wǎng)和金融等領(lǐng)域都正通過挖掘?qū)崟r數(shù)據(jù)來提升業(yè)務(wù)價值。
在技術(shù)方面，大數(shù)據(jù)計算架構(gòu)經(jīng)歷了顯著的演變。從最初的 Hive 傳統(tǒng)數(shù)據(jù)倉庫，到引入 Lakehouse 湖倉架構(gòu)，再到目前國內(nèi)流行的 Paimon 流式湖倉架構(gòu)，這些演進的核心驅(qū)動力在于提升業(yè)務(wù)的時效性。從傳統(tǒng)的 T+1 天模式，逐步縮短到 T+1 小時，再到 T+1 分鐘。然而，由于湖存儲架構(gòu)是基于文件系統(tǒng)的，其分鐘級延遲幾乎是極限。但是許多業(yè)務(wù)場景，如搜索推薦、廣告歸因和異常檢測，都要求秒級的實時響應(yīng)。因此，業(yè)界亟需能夠支持秒級存儲的解決方案。盡管大數(shù)據(jù)技術(shù)已經(jīng)取得了長足的發(fā)展，但在大數(shù)據(jù)分析場景中，仍然缺乏一款能夠有效支持秒級存儲的解決方案。
那么在大數(shù)據(jù)里面最常用的秒級存儲是什么呢？當然是 Apache Kafka。Flink 與 Kafka 的組合也已經(jīng)成為業(yè)界構(gòu)建實時數(shù)倉的典型架構(gòu)。然而，這個組合在實際應(yīng)用中并不總是那么理想，原因在于當我們將 Kafka 應(yīng)用于大數(shù)據(jù)分析時，會遇到一系列挑戰(zhàn)和問題。
Kafka 在實時分析場景遇到的問題
一個主要的問題是，Kafka 不支持數(shù)據(jù)更新功能。在數(shù)據(jù)倉庫中，“更新”是一個非常重要的功能，對于一個數(shù)倉來說，經(jīng)常需要“更新”的能力去修正一些數(shù)據(jù)。由于 Kafka 不支持更新，所以它只能將主鍵上重復的數(shù)據(jù)都存儲下來。當計算引擎消費這些數(shù)據(jù)時，就會接收到重復的數(shù)據(jù)。
為了確保計算結(jié)果的準確性，計算引擎必須執(zhí)行去重操作。然而，這個去重過程本身是非常耗費資源的。在 Flink 中，這需要使用 State 來物化上游的全部數(shù)據(jù)，并且每次消費 Kafka 數(shù)據(jù)時，都必須承擔去重的成本，這個成本是相當高的。這種高成本的去重要求限制了 Kafka 數(shù)據(jù)的業(yè)務(wù)復用能力。例如，在淘天集團構(gòu)建實時數(shù)據(jù)中間層的過程中，由于 Kafka 的這些限制，他們選擇不構(gòu)建 DWS 層。
第二個主要問題是，Kafka 不支持數(shù)據(jù)探查功能。在數(shù)據(jù)倉庫建設(shè)中，數(shù)據(jù)探查是一個基本能力。無論是排查問題還是進行數(shù)據(jù)探索，都需要進行數(shù)據(jù)查詢。然而，Kafka 本質(zhì)上是一個黑盒，不支持直接查詢。為了解決這個問題，業(yè)界通常采用兩種方案：
同步到 OLAP 系統(tǒng)：將 Kafka 數(shù)據(jù)同步到 OLAP 系統(tǒng)中進行查詢。不過，這種方法會引入額外的系統(tǒng)組件，增加復雜性和成本。此外，數(shù)據(jù)在不同系統(tǒng)間的同步也可能導致不一致性。
使用 Trino 等查詢引擎直接查詢 Kafka：這種方法避免了數(shù)據(jù)同步問題，但由于 Kafka 僅支持 Full Scan，無法實現(xiàn) Data Skipping，因此在處理大規(guī)模數(shù)據(jù)時效率較低。例如，在 1GB 數(shù)據(jù)上進行簡單查詢都可能需要一分鐘，這使得這種方法在大規(guī)模應(yīng)用中基本上不可行。
第三個問題是數(shù)據(jù)回溯的困難。在數(shù)據(jù)倉庫中，數(shù)據(jù)回溯是常見需求，例如在物流行業(yè)中，可能需要回溯幾個月的數(shù)據(jù)進行分析。然而，在 Kafka 中，長時間存儲大量數(shù)據(jù)會導致成本過高，因此通常只能存儲幾天的數(shù)據(jù)。此外，當進行大規(guī)模數(shù)據(jù)回溯時，所有數(shù)據(jù)流量都必須經(jīng)過 Kafka Broker，這會導致回溯操作的性能非常慢。同時，這種操作還會消耗 Broker 的 CPU 資源，污染其頁面緩存（page cache），從而對其他在線業(yè)務(wù)產(chǎn)生負面影響。
最后一個問題是網(wǎng)絡(luò)成本。根據(jù)多項數(shù)據(jù)資料顯示，網(wǎng)絡(luò)成本占據(jù)了 Kafka 成本的 88%。在數(shù)據(jù)倉庫中，一寫多讀是非常常見的操作模式，并且每個消費者通常只消費數(shù)據(jù)的一部分。例如，在阿里巴巴內(nèi)部的數(shù)萬條 Flink SQL 作業(yè)中，平均每個作業(yè)僅使用了上游數(shù)據(jù)的 49% 的列。然而，當用戶需要消費這 49% 的列時，仍然需要讀取所有列的數(shù)據(jù)，這意味著需要承擔 100% 的網(wǎng)絡(luò)帶寬成本。這種情況導致了網(wǎng)絡(luò)資源的極大浪費。
總結(jié)來說，將 Kafka 用于實時分析場景時，會面臨以下核心問題：不支持更新、無法探查、數(shù)據(jù)回溯難、網(wǎng)絡(luò)成本高。這些問題導致 Flink + Kafka 的組合在某些實時分析應(yīng)用場景中并不是最理想的選擇。
那么其本質(zhì)的原因是什么？
這是因為Kafka 是為流消息設(shè)計的，并不是為流分析設(shè)計的。每個系統(tǒng)都有其特定的定位和優(yōu)勢，Kafka 在消息隊列場景中非常高效，因為它通常以行存格式（如 CSV、JSON、AVRO）存儲數(shù)據(jù)。然而，對于需要處理大規(guī)模數(shù)據(jù)和復雜查詢的分析場景來說，行存格式的效率則顯得不足。需要底層存儲具備強大的Data Skipping 能力，以及支持列裁剪和條件下推等特性。在這種情況下，列存格式顯然更為適合。
Fluss：Flink Unified Streaming Storage
在構(gòu)建這樣的四象限矩陣時，我們可以觀察到一個有趣的現(xiàn)象：象限左邊是業(yè)務(wù)型系統(tǒng)，右邊是分析型系統(tǒng)，上面是流存儲，下面是表存儲?？梢钥吹?，業(yè)務(wù)型系統(tǒng)里面不管是數(shù)據(jù)庫，還是流存儲，都采用的是行存，因為行存在這個場景更為高效。相反，像 Iceberg, Snowflake 這些分析型系統(tǒng)都采用的列存，因為列存在分析場景更高效。在這個矩陣中，右上角是一個空白區(qū)域，代表這個市場里空缺了一個存儲，即面向分析場景的流存儲，不出意料的話，這個流存儲采用的會是列存格式。
為了填補這一市場空白，并解決 Flink 在實時流分析場景中的痛點問題，我們在兩年前發(fā)起了一個流存儲項目，命名為 "FLink Unified Streaming Storage"，取了項目名的首字母縮寫，拼成了 Fluss 這個單詞。值得一提的是，F(xiàn)link 這個名字源自德語，意為“敏捷迅速”，而 Fluss 恰巧也是個德語單詞，意為“河流。這種命名不僅向 Flink 項目的起源致敬，也象征著流數(shù)據(jù)如同河流般源源不斷地流動、分發(fā)，并最終匯聚到數(shù)據(jù)湖中。
Fluss 核心特性
接下來介紹一下 Fluss 的一些核心特性：
首先，不出所料，F(xiàn)luss 采用列式的流存儲。在底層文件存儲中采用了 IPC Streaming Format 協(xié)議，而 Arrow 是一種非常優(yōu)秀的流式列存儲格式?；?Arrow，我們實現(xiàn)了非常高效的列裁剪功能。右側(cè)展示了對 Fluss 和 Kafka 的基準測試結(jié)果。橫軸表示讀取列的數(shù)量，縱軸表示讀取吞吐量?？梢钥吹?，隨著裁剪的列數(shù)增加，F(xiàn)luss 的讀取性能成比例上升。當裁剪到 90% 的列時，F(xiàn)luss 的讀取吞吐量已經(jīng)提高了 10 倍。此外，F(xiàn)luss 的列裁剪是在服務(wù)端進行的，這意味著發(fā)送給客戶端的數(shù)據(jù)已經(jīng)是裁剪過的，從而節(jié)省了大量的網(wǎng)絡(luò)成本。
第二點，實時更新和CDC是流分析中非常依賴的存儲能力，我們也對此進行了支持?？梢岳斫鉃?Fluss 的流存儲基礎(chǔ)是一個日志表（Log Tablet），我們在日志之上構(gòu)建了 KV 索引，從而支持高效的實時更新。Log 和 KV 之間的關(guān)系類似于流表的二象性，KV 的更新會生成變更日志（Changelog）寫入 Log Tablet；在故障恢復時，Log Tablet 的數(shù)據(jù)又用于恢復鍵值表（KV Tablet）。KV Tablet 底層實際上是一個 RocksDB 的 LSM 樹。因此，我們將流存儲與 LSM 結(jié)構(gòu)結(jié)合，支持大規(guī)模實時更新以及部分列的更新，從而實現(xiàn)高效的寬表拼接。
此外，KV 生成的 Changelog 可以直接被 Flink 流讀取，無需額外的去重操作，節(jié)省了大量計算資源，實現(xiàn)了數(shù)據(jù)的業(yè)務(wù)復用。由于我們構(gòu)建了 KV 索引，因此可以支持高性能的主鍵點查，并可作為實時處理鏈路中的維表關(guān)聯(lián)。用戶還可以通過點查的 query 語句直接探查 Fluss 數(shù)據(jù)，我們還支持 LIMIT、COUNT 等查詢功能，以滿足用戶的數(shù)據(jù)探查需求。
Fluss 還有一個非常重要的特性是湖流一體。過去，用戶為了搭建實時鏈路和離線鏈路，同樣一份數(shù)據(jù)需要在流存儲和湖存儲冗余存儲，造成成本浪費。湖流一體的概念是指“湖存儲的數(shù)據(jù)”和“流存儲的數(shù)據(jù)”能夠作為一個整體進行管理和消費，從而避免數(shù)據(jù)的冗余存儲，避免數(shù)據(jù)和元數(shù)據(jù)不一致的問題。
在底層，F(xiàn)luss 維護了一個 Compaction Service，該服務(wù)會自動地將 Fluss 數(shù)據(jù)轉(zhuǎn)換為湖存儲的格式，并確保兩邊元數(shù)據(jù)的一致性。此外，它還保證兩邊的數(shù)據(jù)分布也是一致的，即分區(qū)和分區(qū)一一對齊，Bucket 和 Bucket 也一一對齊。這使得在流轉(zhuǎn)湖的過程中，無需引入網(wǎng)絡(luò) Shuffle，只需將 Arrow 文件直接轉(zhuǎn)換為 Parquet 文件即可。這種轉(zhuǎn)換在業(yè)界已有非常成熟且高效的實現(xiàn)。
在擁有湖和流兩層數(shù)據(jù)后，F(xiàn)luss 的一個關(guān)鍵特性是共享數(shù)據(jù)。具體來說，湖存儲作為流存儲的歷史數(shù)據(jù)層，負責存儲長周期、分鐘級延遲的數(shù)據(jù)；而流存儲作為湖存儲的實時數(shù)據(jù)層，負責存儲短周期、毫秒級延遲的數(shù)據(jù)，這兩者的數(shù)據(jù)可以互相共享。當進行流讀取時，湖存儲可以作為歷史數(shù)據(jù)提供高效的回溯性能。在回溯到當前位點后，系統(tǒng)會自動切換到流存儲繼續(xù)讀取，并確保不會讀取重復數(shù)據(jù)。在批查詢分析中，流存儲可以為 Lakehouse 提供實時數(shù)據(jù)的補充，從而實現(xiàn) Lakehouse 秒級新鮮度的分析。我們將這種功能稱為 Union Read。
除此之外，我們同步到湖存儲的格式完全遵循現(xiàn)有湖存儲的開放協(xié)議，因此現(xiàn)有的一些查詢引擎（如 Spark、StarRocks、Trino）可以直接查詢湖存儲中的數(shù)據(jù)，無縫融入用戶已有的 Lakehouse 架構(gòu)中。目前，F(xiàn)luss 已經(jīng)完成了對 Paimon 的完全集成，對 Iceberg 的集成也在計劃中。
這就是我們Fluss整體的架構(gòu)圖，F(xiàn)luss是一個面向?qū)崟r分析的流存儲。Fluss 需要維護一個 Server 集群，提供實時讀寫的能力，同時使用 Remote Storage 來做數(shù)據(jù)的分層，降低數(shù)據(jù)存儲成本。并且跟Lakchouse 做了一個非常無縫的集成來支持豐富的查詢能力。Fluss 的核心特性包括實時的流讀流寫、列式裁剪、流式的更新、CDC訂閱、實時點查、還有湖流一體。
Fluss 的核心特性結(jié)合，實現(xiàn)了一個非常理想的應(yīng)用場景 Delta Join。在 Flink 中，雙流 Join 是一個非?；A(chǔ)的功能，常用于構(gòu)建寬表。然而，這也是一個常常讓開發(fā)人員感到頭疼的功能。因為雙流 Join 需要在 State 中維護上游全量的數(shù)據(jù)，這導致其狀態(tài)通常非常龐大。例如，淘寶最大的 Flink 作業(yè)之一是成交引導的雙流 Join（曝光關(guān)聯(lián)訂單），需要消耗 50TB 的狀態(tài)。但這帶來了很多問題，包括成本高、作業(yè)不穩(wěn)定、Checkpoint超時、重啟恢復慢等等。
因此我們充分利用 Fluss 的 CDC 流讀+索引點查的能力研發(fā)了一套新的 Flink 的 Join 算子實現(xiàn)，叫 Delta Join。Delta Join 可以簡單理解成“雙邊驅(qū)動的維表Join”，就是左邊來了數(shù)據(jù)，就根據(jù)Join Key去點查右表；右邊來了數(shù)據(jù)，就根據(jù) Join Key 去點查左表。全程就像維表Join一樣不需要state，但是實現(xiàn)了雙流Join一樣的語義，即任何一邊有數(shù)據(jù)更新，都會觸發(fā)對關(guān)聯(lián)結(jié)果的更新。
在測試中，我們使用了淘寶最大的雙流 Join 作業(yè)進行性能評估。在從雙流 Join 遷移到 Delta Join 后，成功減免了 50TB 的大狀態(tài)，使得作業(yè)運行更加穩(wěn)定，Checkpoint 也不再超時。在雙十一的數(shù)據(jù)壓測回追中，我們發(fā)現(xiàn)，在保證相同吞吐量的情況下，F(xiàn)link 的資源消耗能夠降低10倍，從2300 CU 減少到200 CU。此外，在回追過程中，我們還可以利用湖流一體歸檔的 Paimon 表加上 Flink Batch Join 進行數(shù)據(jù)回追，將回追1天數(shù)據(jù)的時間從4小時縮短到0.5小時。使用批作業(yè)進行數(shù)據(jù)回追，展示了流批一體的一個非常有前景的應(yīng)用場景。
除了資源的減少和性能的提升，對于用戶最大的收益其實是靈活性的提升。以前的 State 是 Flink 內(nèi)置的黑盒，用戶看不見摸不著，一修改作業(yè)就要重跑 State，耗時耗力。在使用 Delta Join 后，相當于狀態(tài)與作業(yè)進行了解耦，修改作業(yè)不需要重跑 State，所以回追很高效。并且數(shù)據(jù)都在 Fluss 里面，變得可查可分析，提升了業(yè)務(wù)靈活性和開發(fā)效率。目前，我們已經(jīng)在 Flink 社區(qū)提交了 Delta Join 的 FLIP-486 提案，對于這個提案感興趣的朋友可以關(guān)注一下。
Fluss 未來規(guī)劃
關(guān)于 Fluss 的未來規(guī)劃，最重要的有三件事，這三件事分別對應(yīng)了 Fluss 與三個開源軟件之間的關(guān)系：
Kafka 協(xié)議兼容：這是為了幫助已有的流數(shù)據(jù)更好地遷移到 Fluss 上。
與 Flink 的深度協(xié)同優(yōu)化：這一規(guī)劃包括通過存儲+優(yōu)化器+執(zhí)行引擎的協(xié)同優(yōu)化，以解決之前存在的一些難點和痛點。Delta Join 就是一個很好的例子，通過這種深度協(xié)同，F(xiàn)luss 可以與 Flink 緊密結(jié)合，提升整體的流處理性能和穩(wěn)定性。
為 Paimon 提供實時數(shù)據(jù)層：通過打造湖流一體架構(gòu)，F(xiàn)luss 希望與 Paimon 結(jié)合，提供一個實時與離線一體化的存儲解決方案。
Fluss 開源
在11月29日舉辦的 Flink Forward Asia 2024 大會主題演講上，阿里巴巴正式開源了 Fluss 項目（https://github.com/alibaba/fluss）。阿里巴巴開源委員會副主席王峰先生，在現(xiàn)場進行了 Fluss 項目的開源，贏得了現(xiàn)場觀眾的熱烈反響。
Fluss 目前已經(jīng)在 GitHub 上以 Apache 2.0 協(xié)議正式開源，項目地址為：https://github.com/alibaba/fluss，歡迎大家關(guān)注和 Star。并且我們計劃于 2025 年將其捐贈到 Apache 軟件基金會。在此，我們誠摯地邀請各位加入 Fluss 開源社區(qū)，共同促進這一新興項目的成長與發(fā)展。歡迎大家加入 Fluss 社區(qū)釘釘群 109135004351，歡迎大家一起探索，參與開發(fā)和貢獻，并攜手構(gòu)建下一代的流存儲技術(shù)！
