流批一體技術簡介

摘 要 ： 本文由阿裏雲 Flink 團隊蘇軒楠老師撰寫，旨在向 Flink 使用者整體介紹 Flink 流批一體的技術和挑戰。內容主要分為以下三個部份：

1. 流批一體技術簡介

2. 面臨的挑戰

3. 總結

Tips： 點選 「閱讀原文」 跳轉阿裏雲即時計算 Flink～

Flink 的流批一體概念相信大家並不陌生，盡管流批一體被廣泛討論，但在很多使用者心中，Flink 更多的是作為流計算引擎的事實標準，常有朋友提到 Flink，也是更熟悉其在流計算場景的使用。今天，本文旨在全面向讀者介紹 Flink 的流批一體技術及其所面臨的技術挑戰。在未來的分享中，我們將進一步探討流批一體在不同場景下的套用，以及透過公開渠道收集到的一些企業使用 Flink 流批一體功能的落地情況。

01

流批一體技術簡介

在這一章節我們會簡要地介紹一下流批一體相關的技術，讓大家了解流批一體的整體架構以及架構中各個元件的技術選擇。

1.1 流批一體的架構演進

首先我們來看看，流批一體的架構是怎樣演進而來的。

■ Lambda 架構

Lambda 架構通常是在原有的離線計算中發展而來的。在離線計算的鏈路（批次處理層）上，為了滿足業務數據即時性的要求，會在離線鏈路的基礎上再增加一條即時計算鏈路（速度處理層）。最後，在對外提供服務的時候，會合並批次處理層和速度層的檢視（服務層）。雖然 Lambda 架構可以在不改變原有離線計算架構的基礎上，同時享受到離線和即時計算帶來的好處，但是它在使用的過程中存在著以下的問題：

圖片來源：Questioning the Lambda Architecture [1]

■ Kappa 架構

Kappa 架構相當於在 Lambda 架構上去掉了批次處理層（Batch Layer），只留下單獨的流處理層（Speed Layer）。透過訊息佇列的數據保留功能，來實作上遊重放（回溯）能力。Kappa 架構解決了上面提到的 Lambda 架構所面臨的問題，但是在這個架構下，就無法享受到離線計算帶來的好處了。比如透過流計算來回溯的效能會比使用批計算的效能要差。而且對於沒有即時需求的作業，使用 Kappa 架構的流計算會造成不必要的資源浪費。

圖片來源：Questioning the Lambda Architecture [1]

■ 流批一體架構

透過使用流批一體的計算引擎和流批一體的儲存格式，我們可以很好地解決 Lambda 和 Kappa 架構中存在的問題。

圖片來源：Flink Forward Asia 2020 - 流批一體技術在天貓雙11的套用 [2]

介紹完流批一體的架構之後，讓我們來看一下流批一體架構中最重要的兩個元件：計算引擎和儲存。

1.2 流批一體的計算引擎

Apache Flink 從設計之初就提出了「批次處理是流處理的特殊情況」。使用者可以使用 DataStream API 和 Flink 的 SQL API 來同時定義流作業和批作業。Flink 在流批一體這個方向上已經做了非常多的工作來提高使用者的體驗，作業的穩定性和效能，目前 Flink Batch 已經在很多公司的生產環境上落地。Flink 社群未來也會持續投入 Flink 流批一體的發展。

Apache Spark 也是最早提出流批一體理念的計算引擎之一，可以用作流批一體計算引擎。與 Flink 不同的是，它的流計算是基於微批（mini-batch）來實作的，在流計算語意的支持和端到端延遲上會差一些，面對復雜、大規模即時計算場景的極致需求可能會力不從心。Apache Spark 雖然也在探索使用 Continuous Processing 來支持流計算，降低延遲，但目前還是屬於實驗階段 [3]，並且經過筆者的觀察 Continuous Processing 的投入一直不大，2021 年之後就幾乎停滯了。

1.3 流批一體的湖表格式

在流批一體這個大場景下，計算引擎只是其中的一環，流批一體的儲存格式更是不可或缺的一部份。Flink 在流批一體的儲存格式上做了許多探索，對接了多個不同的儲存格式，目前在開源社群主流的支持 Flink 流批一體的儲存格式有下面這些：

Apache Paimon 是流批一體的湖儲存格式。可以使用 Flink CDC 來一鍵入湖到 Paimon 中，也可以透過 Flink SQL 或 Spark SQL 來批寫、流寫到 Paimon 當中。Paimon 也可以被 Flink 或 Spark 流讀，這也是它作為流式數據湖的特有能力之一。它有著強大的流讀流寫支持，給流式湖儲存帶來僅 1-5 分鐘的延遲 [4]。

Apache Hudi 原生支持多引擎，因此既可以對批流進行讀寫消費，也可以使用Presto進行互動式分析 [5]。Flink 接入之後，把 Hudi 的時延可以達到十分鐘級[4]。

Apache Iceberg 早在 2020 年，阿裏雲就試圖把 Flink 融入 Iceberg 中，在 Iceberg 中做了很多 Flink 的整合。在把 Flink 融入 Iceberg 後，Iceberg 就有了 Flink 流讀流寫的力。目前 Flink 寫入 Iceberg，並不能太即時，因此更推薦在 1 小時左右的更新 SLA 保障[4]。

1.4 流批一體的數倉

除了流批一體的湖表格式，還有流批一體的數倉也可以作為流批一體的儲存，例如開源的 Starrocks, Clickhouse, Apache Druid 等。還有類似阿裏雲的商業化產品 Hologres 也可以作為流批一體的數倉。但是數倉的儲存成本比湖儲存更高，我們又看到一些的做法是把數據透過湖表格式寫入到數據湖中，然後透過數倉來分析數據湖中的數據做 OLAP 分析，例如 Startrocks + Paimon[6], Hologres + Paimon [7] 等湖倉一體方案。

02

面臨的挑戰

在大家使用 Flink Batch 流批一體實踐的過程中，難免會遇到各種各樣的問題和挑戰。Flink 社群也在積極地解決大家在使用過程中遇到的問題，對 Flink 的批作業能力進行打磨，使Flink 的流批引擎的能力逐步地完善。下面介紹了近些年來，大家在 Flink 流批一體實踐中遇到的挑戰，以及社群的解決方案。

2.1 流批 Shuffle 差異

Flink 流作業的 shuffle 與 批作業的 shuffle 通常是不一樣的，在流作業的情況下使用的是 Pipeline Shuffle，Pipeline Shuffle 的數據是不用落盤的，但是這要求作業啟動的時候所有的算子都要啟動起來，這與常見的 Batch 作業排程需求不匹配。因此，在批模式下，通常都是使用 Blocking Shuffle，這樣上遊 task 會把 Shuffle 數據寫到離線檔中，等下遊 task 啟動以後，再來消費 Shuffle 的數據。

Flink 內部預設的實作是使用 Internal Shuffle， 是把上遊計算節點數據寫到 TaskManager 本地盤，下遊節點連線到上遊 TaskManager 上讀取 Shuffle 檔。這會導致 TaskManager 計算工作完成以後，不能立刻結束，要等下遊消費完 Shuffle 檔後才能釋放掉。這樣不僅造成了資源浪費，而且容錯代價大。

因此，Flink 社群在已開始設計 Shuffle Service 的時候就把他作為一個 pluggable [8]，以便於使用者能夠方便的拓展來實作 Remote Shuffle Service。Remote Shuffle Service 透過單獨的集群提供數據的 Shuffle 服務，可以避免 TaskManager Shuffle 的資源利用率低和容錯開銷大的問題。目前 Apache Celeborn [9] 支持作為 Flink 的 remote shuffle.

圖片來源：Flink Forward Asia 2022 - Flink Shuffle 3.0 Vision, Roadmap & Progress [10]

同時社群也在推動 Shuffle 3.0 中提出了 Hybrid Shuffle，Hybrid Shuffle 將流式 Pipeline Shuffle 跟批式 Blocking Shuffle 的特點結合在一起，讓使用者在寫數據時，既可以寫入記憶體透過記憶體直接進行消費，也可以在記憶體中存放不下這麽多數據、下遊消費不夠及時的時候，將數據寫入到磁盤當中進行後期消費。透過自適應切換，在上遊產出數據的過程中和完成後，下遊可以隨時消費，從而徹底消除資源碎片的情況 [10]。

2.2 Batch 效能

Flink batch 作業的效能也是使用者在使用 Flink batch 中最關心的問題之一，Flink batch 在作業效能最佳化上面做了非常多的改進，使得 Flink batch 在 TPC-DS benchmark 上的表現在每個版本都有很大的提升。例如使用 Operator Fusion Codegen [11] 來最佳化 SQL planner 生成的程式碼，透過 adaptive local hash aggregate[12] 來動態決定是否使用 local aggregation，透過 runtime filter 和 dynamic data prune 來最佳化數據處理的效率，實作 Adaptive Execution Plan（AQE） 做了自動並行推斷，動態負載均衡等[13]。

圖片來源：Flink Forward Asia 2023 - 深入解析 Flink Batch 新進展 [13]

2.3 慢節點問題

在一個分布式系統裏，因為個別的機器故障、資源緊張或者是網路問題，可能導致單個並行的效能下降，這些慢的節點可能成為整個作業的瓶頸。和傳統 MapReduce、Spark 的思路類似，Flink 1.17 版本引入了推測執行來解決慢節點的問題 [14]。當檢測到長尾任務後，在非熱的機器上部署長尾任務的映像例項。哪個先執行完就用哪個結果，並把其他的映像任務取消掉。

2.4 並行配置易用性

在 Flink Batch 作業中，為作業節點設定適當的並列性並不是一件容易的事情。在批作業中，並列度設定得太小可能導致太長的執行時間和 Failover 發生時大量的回退。相反，並列度設定得過大，則可能導致資源浪費和任務部署和網路 Shuffle 更多的成本開銷。為了解決這個問題，Flink 在 1.15 中引入了 Adaptive Batch Scheduler [15]，這能讓 Flink 根據消費的數據量大小來自動決定作業節點的並行度，免去了使用者需要手動調整作業並行度的煩惱。

2.5 Hive SQL 相容性

由於 Flink SQL 使用的是標準的 ANSI SQL，並且 Hive SQL 與 ANSI SQL 語法差異較多。不少使用者在遷移 Hive SQL 到 Flink SQL 上的時候會遇到不少的阻礙。雖然 Flink SQL 本身提供了 Hive Dialect [15]，但是在 Flink 1.15 版本，距離完全相容 Hive SQL 仍然有不小的差距。比如，快手在選定了一批準備遷移的作業後，透過解析驗證，就發現了諸多不支持的語法。在快手給出 input 後，社群第一優先級做出了支持。如上圖所示，我們列出了比較重要且很常用的一些語法，比如 CTAS、ADD JAR、USING JAR、宏命令、Transform 等。

Flink 社群在 1.16 版本做了大量工作, 包括 CTAS [16]、ADD JAR、USING JAR [17] 等等，來補全 Hive 語法。經過 qtest 測試，整體相容度能達到 95%，基本能保證使用者現有的 Query 都能遷到 Flink 上來。

03

總結

流批一體是 Flink 非常重要的一個發展方向之一，眾多使用者都在使用的過程中給予了 Flink 社群寶貴的反饋。隨著更多的開發者加入了開源社群的開發工作，這讓 Flink Batch 的能力不斷得到打磨和發展。在社群成員的努力下，很多使用者已經可以十分順利地把 Flink 流批一體在他們的生產環境中落地。我們會在後期分享中，給大家介紹一些流批一體的主要落地場景，以及我們在公開渠道收集到的各個公司使用流批一體的落地情況。

雖然，Flink 流批一體已經達到生產可用的狀態，但是社群也看到仍然有不少需要繼續投入的地方，例如繼續完善 DataStream API batch 的能力，使其能夠完全與 DataSet API 能力對齊；更加深入地與 Apache Celeborn 結合，結合 Flink 特點實作動態切換 Shuffle 的機制，多級儲存引入記憶體、支持 Flink Hybird Shuffle 等；更加深入地與 Apache Paimon 對接，整合流批一體引擎和儲存的能力，讓使用者能夠更加簡單地使用 Apache Flink + Apache Paimon 搭建流批一體的數據湖倉。

[1] https://www.oreilly.com/radar/questioning-the-lambda-architecture/

[2] https://www.bilibili.com/video/BV1164y1o7yc/

[3] https://spark.apache.org/docs/latest/structured-streaming-programming-guide.html#continuous-processing

[4] https://flink-learning.org.cn/article/detail/d17cc1d2a06946b40c51d4301df6e540

[5]

[6] https://flink-learning.org.cn/article/detail/02a574303b7e65fd53e13a82b40a8d8f

[7] https://flink-learning.org.cn/article/detail/84f501725034542a7f41e0670645c714

[8] https://cwiki.apache.org/confluence/display/FLINK/FLIP-31:+Pluggable+Shuffle+Service

[9] https://celeborn.apache.org/

[10] https://flink-learning.org.cn/article/detail/f6449048654123b163e29917e8ad5a79

[11] https://cwiki.apache.org/confluence/display/FLINK/FLIP-315+Support+Operator+Fusion+Codegen+for+Flink+SQL

[12] https://issues.apache.org/jira/browse/FLINK-30542

[13] https://developer.aliyun.com/ebook/8229/115382

[14] https://cwiki.apache.org/confluence/display/FLINK/FLIP-168:+Speculative+Execution+for+Batch+Job

[15] https://cwiki.apache.org/confluence/display/FLINK/FLIP-152

[16] https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=199541185

[17] https://cwiki.apache.org/confluence/display/FLINK/FLIP-214

歡迎大家加入 Flink Batch 交流釘釘群。本群旨在為 Flink Batch 愛好者提供一個交流技術和傳遞資訊的平台，在這裏：

「Flink Batch 交流群」群的釘釘群號：34817520，也可以掃碼加入

活動推薦

阿裏雲基於 Apache Flink 構建的企業級產品-即時計算 Flink 版現開啟活動：

新使用者復制下方連結或者掃描二維碼即可0元免費試用 Flink + Paimon

了解活動詳情： https://free.aliyun.com/?pipCode=sc

▼ 關註「 Apache Flink 」，獲取更多技術幹貨 ▼

點選「閱 讀原文 」跳轉 阿裏雲即時計算 Flink ～