?摘要：本文整理自京東高級技術專家韓飛在 Flink Forward Asia 2021 流批一體專場的分享。主要內容包括：

• 整體思考
• 技術方案及優化
• 落地案例
• 未來展望??

01整體思考

提到流批一體，不得不提傳統的大數據平臺 —— Lambda 架構。它能夠有效地支撐離線和實時的數據開發需求，但它流和批兩條數據鏈路割裂所導致的高開發維護成本以及數據口徑不一致是無法忽視的缺陷。

通過一套數據鏈路來同時滿足流和批的數據處理需求是最理想的情況，即流批一體。此外我們認為流批一體還存在一些中間階段，比如只實現計算的統一或者只實現存儲的統一也是有重大意義的。

以只實現計算統一為例，有一些數據應用的實時性要求比較高，比如希望端到端的數據處理延時不超過一秒鐘，這對目前開源的、適合作為流批統一的存儲來說是一個很大的挑戰。以數據湖為例，它的數據可見性與 commit 的間隔相關，進而與 Flink 做 checkpoint 的時間間隔相關，此特性結合數據處理鏈路的長度，可見做到端到端一秒鐘的處理并不容易。因此對于這類需求，只實現計算統一也是可行的。通過計算統一去降低用戶的開發及維護成本，解決數據口徑不一致的問題。

在流批一體技術落地的過程中，面臨的挑戰可以總結為以下 4 個方面：

• 首先是數據實時性。如何把端到端的數據時延降低到秒級別是一個很大的挑戰，因為它同時涉及到計算引擎及存儲技術。它本質上屬于性能問題，也是一個長期目標。
• 第二個挑戰是如何兼容好在數據處理領域已經廣泛應用的離線批處理能力。此處涉及開發和調度兩個層面的問題，開發層面主要是復用的問題，比如如何復用已經存在的離線表的數據模型，如何復用用戶已經在使用的自定義開發的 Hive UDF 等。調度層面的問題主要是如何合理地與調度系統進行集成。
• 第三個挑戰是資源及部署問題。比如通過不同類型的流、批應用的混合部署來提高資源利用率，以及如何基于 metrics 來構建彈性伸縮能力，進一步提高資源利用率。
• 最后一個挑戰也是最困難的一個：用戶觀念。大多數用戶對于比較新的技術理念通常僅限于技術交流或者驗證，即使驗證之后覺得可以解決實際問題，也需要等待合適的業務來試水。這個問題也催生了一些思考，平臺側一定要多站在用戶的視角看待問題，合理地評估對用戶的現有技術架構的改動成本以及用戶收益、業務遷移的潛在風險等。

上圖是京東實時計算平臺的全景圖，也是我們實現流批一體能力的載體。中間的 Flink 基于開源社區版本深度定制。基于該版本構建的集群，外部依賴包含三個部分，JDOS、HDFS/CFS 和 Zookeeper。

• JDOS 是京東的 Kubernetes 平臺，目前我們所有 Flink 計算任務容器化的，都運行在這套平臺之上；
• Flink 的狀態后端有 HDFS 和 CFS 兩種選擇，其中 CFS 是京東自研的對象存儲；
• Flink 集群的高可用是基于 Zookeeper 構建的。

在應用開發方式方面，平臺提供 SQL 和 Jar 包兩種方式，其中 Jar 的方式支持用戶直接上傳 Flink 應用 Jar 包或者提供 Git 地址由平臺來負責打包。除此之外我們平臺化的功能也相對比較完善，比如基礎的元數據服務、SQL 調試功能，產品端支持所有的參數配置，以及基于 metrics 的監控、任務日志查詢等。

連接數據源方面，平臺通過 connector 支持了豐富的數據源類型，其中  JDQ 基于開源 Kafka 定制，主要應用于大數據場景的消息隊列；JMQ 是京東自研，主要應用于在線系統的消息隊列；JimDB 是京東自研的分布式 KV 存儲。

在當前 Lambda 架構中，假設實時鏈路的數據存儲在 JDQ，離線鏈路的數據存在 Hive 表中，即便計算的是同一業務模型，元數據的定義也常常是存在差異的，因此我們引入統一的邏輯模型來兼容實時離線兩邊的元數據。

在計算環節，通過 FlinkSQL 結合 UDF 的方式來實現業務邏輯的流批統一計算，此外平臺會提供大量的公用 UDF，同時也支持用戶上傳自定義 UDF。針對計算結果的輸出，我們同樣引入統一的邏輯模型來屏蔽流批兩端的差異。對于只實現計算統一的場景，可以將計算結果分別寫入流批各自對應的存儲，以保證數據的實時性與先前保持一致。

對于同時實現計算統一和存儲統一的場景，我們可以將計算的結果直接寫入到流批統一的存儲。我們選擇了 Iceberg 作為流批統一的存儲，因為它擁有良好的架構設計，比如不會綁定到某一個特定的引擎等。

在兼容批處理能力方面，我們主要進行了以下三個方面的工作：

第一，復用離線數倉中的 Hive 表。

以數據源端為例，為了屏蔽上圖左側圖中流、批兩端元數據的差異，我們定義了邏輯模型 gdm_order_m 表，并且需要用戶顯示地指定 Hive 表和 Topic 中的字段與這張邏輯表中字段的映射關系。這里映射關系的定義非常重要，因為基于 FlinkSQL 的計算只需面向這張邏輯表，而無需關心實際的 Hive 表與 Topic 中的字段信息。在運行時通過 connector 創建流表和批表的時候，邏輯表中的字段會通過映射關系被替換成實際的字段。

在產品端，我們可以給邏輯表分別綁定流表和批表，通過拖拽的方式來指定字段之間的映射關系。這種模式使得我們的開發方式與之前有所差異，之前的方式是先新建一個任務并指定是流任務還是批任務，然后進行 SQL 開發，再去指定任務相關的配置，最后發布任務。而在流批一體模式下，開發模式變為了首先完成 SQL 的開發，其中包括邏輯的、物理的 DDL 的定義，以及它們之間的字段映射關系的指定，DML 的編寫等，然后分別指定流批任務相關的配置，最后發布成流批兩個任務。

第二，與調度系統打通。

離線數倉的數據加工基本是以 Hive/Spark 結合調度的模式，以上圖中居中的圖為例，數據的加工被分為 4 個階段，分別對應數倉的 BDM、FDM、GDM 和 ADM 層。隨著 Flink 能力的增強，用戶希望把 GDM 層的數據加工任務替換為 FlinkSQL 的批任務，這就需要把 FlinkSQL 批任務嵌入到當前的數據加工過程中，作為中間的一個環節。

為了解決這個問題，除了任務本身支持配置調度規則，我們還打通了調度系統，從中繼承了父任務的依賴關系，并將任務自身的信息同步到調度系統中，支持作為下游任務的父任務，從而實現了將 FlinkSQL 的批任務作為原數據加工的其中一個環節。

第三，對用戶自定義的 Hive UDF、UDAF 及 UDTF 的復用。

對于現存的基于 Hive 的離線加工任務，如果用戶已經開發了 UDF 函數，那么最理想的方式是在遷移 Flink 時對這些 UDF 進行直接復用，而不是按照 Flink UDF 定義重新實現。

在 UDF 的兼容問題上，針對使用 Hive 內置函數的場景，社區提供了 load hive modules 方案。如果用戶希望使用自己開發的 Hive UDF，可以通過使用 create catalog、use catalog、create function，最后在 DML 中調用的方式來實現, 這個過程會將 Function 的信息注冊到 Hive 的 Metastore 中。從平臺管理的角度，我們希望用戶的 UDF 具備一定的隔離性，限制用戶 Job 的粒度，減少與 Hive Metastore 交互以及產生臟函數元數據的風險。

此外，當元信息已經被注冊過，希望下次能在 Flink 平臺端正常使用，如果不使用 if not exist 語法，通常需要先 drop function，再進行 create 操作。但是這種方式不夠優雅，同時也對用戶的使用方式有限制。另一種解決方法是用戶可以注冊臨時的 Hive UDF，在 Flink1.12 中注冊臨時 UDF 的方式是 create temporary function，但是該 Function 需要實現 UserDefinedFunction 接口后才能通過后面的校驗，否則會注冊失敗。

所以我們并沒有使用 create temporary function，而是對 create function 做了一些調整，擴展了 ExtFunctionModule，將解析出來的 FunctionDefinition 注冊到 ExtFunctionModule 中，做了一次 Job 級別的臨時注冊。這樣的好處就是不會污染 Hive Metastore，提供了良好的隔離性，同時也沒有對用戶的使用習慣產生限制，提供了良好的體驗。

不過這個問題在社區 1.13 的版本已經得到了綜合的解決。通過引入 Hive 解析器等擴展，已經可以把實現 UDF、GenericUDF 接口的自定義 Hive 函數通過 create temporary function 語法進行注冊和使用。

資源占用方面，流處理和批處理是天然錯峰的。對于批處理，離線數倉每天 0 點開始計算過去一整天的數據，所有的離線報表的數據加工會在第二天上班前全部完成，所以通常 00:00 到 8:00 是批計算任務大量占用資源的時間段，而這個時間段通常在線的流量都比較低。流處理的負載與在線的流量是正相關的，所以這個時間段流處理的資源需求是比較低的。上午 8 點到晚上 0 點，在線的流量比較高，而這個時間段批處理的任務大部分都不會被觸發執行。

基于這種天然的錯峰，我們可以通過在專屬的 JDOS Zone 中進行不同類型的流批應用的混部來提升資源的使用率，并且如果統一使用 Flink 引擎來處理流批應用，資源的使用率會更高。

同時為了使應用可以基于流量進行動態調整，我們還開發了自動彈性伸縮的服務 (Auto-Scaling Service)。它的工作原理如下：運行在平臺上的 Flink 任務上報 metrics 信息到 metrics 系統，Auto-Scaling Service 會基于 metrics 系統中的一些關鍵指標，比如 TaskManager 的 CPU 使用率、任務的背壓情況等來判定任務是否需要增減計算資源，并把調整的結果反饋給 JRC 平臺，JRC 平臺通過內嵌的 fabric 客戶端將調整的結果同步到 JDOS 平臺，從而完成對 TaskManager Pod 個數的調整。此外，用戶可以在 JRC 平臺上通過配置來決定是否為任務開啟此功能。

上圖右側圖表是我們在 JDOS Zone 中進行流批混部并結合彈性伸縮服務試點測試時的 CPU 使用情況。可以看到 0 點流任務進行了縮容，將資源釋放給批任務。我們設置的新任務在 2 點開始執行，所以從 2 點開始直到早上批任務結束這段時間，CPU 的使用率都比較高，最高到 80% 以上。批任務運行結束后，在線流量開始增長時，流任務進行了擴容，CPU 的使用率也隨之上升。

02技術方案及優化

流批一體是以 FlinkSQL 為核心載體，所以我們對于 FlinkSQL 的底層能力也做了一些優化，主要分為維表優化、join 優化、window 優化和 Iceberg connector 優化幾個方面。

首先是維表相關的幾個優化。目前社區版本的 FlinkSQL 只支持部分數據源 sink 算子并行度的修改，并不支持 source 以及中間處理算子的并行度修改。

假設一個 FlinkSQL 任務消費的 topic 有 5 個分區，那么下游算子的實際并行度是 5，算子之間是 forward 的關系。對于數據量比較大的維表 join 場景，為了提高效率，我們希望并行度高一些，希望可以靈活設置它的并行度而不與上游的分區數綁定。

基于此，我們開發了預覽拓撲的功能，不論是 Jar 包、SQL 任務都可以解析并生成 StreamGraph 進行預覽，進一步還能支持修改分組、算子 chain 的策略、并行度、設置 uid 等。

借助這個功能，我們還可以調整維表 join 算子的并行度，并且將分區策略由 forward 調整為 rebalance，然后把這些調整后的信息更新到 StreamGraph。此外我們還實現了動態 rebalance 策略，可以基于 backLog 去判斷下游分區中的負載情況，從而選擇最優的分區進行數據分發。

為了提升維表 join 的性能，我們對所有平臺支持的維表數據源類型都實現了異步 IO 并支持在內存中做緩存。不論是原生的 forward 方式還是 rebalance 方式，都存在緩存失效和替換的問題。那么，如何提高維表緩存的命中率以及如何降低維表緩存淘汰的操作?

以原生的 forward 方式為例，forward 意味著每個 subtask 緩存著隨機的維表數據，與 joinkey 的值有關。對維表的 joinkey 做哈希，就能保證下游每一個算子緩存著與 joinkey 相關的、不同的維表數據，從而有效地提升緩存的命中率。

在實現層面我們新增了一條叫 StreamExecLookupHashJoinRule 的優化規則，并且把它添加到物理 rewrite 的階段。在最底層的掃描數據 StreamExecTableSourceScan 和維表 join StreamExecLookupJoin 之間增加了一個 StreamExecChange 節點，由它來完成對維表數據的哈希操作。可以通過在定義維表 DDL 時指定 lookup.hash.enable=true 來開啟這個功能。

我們對于 forward、rebalance、哈希三種方式開啟緩存，進行了相同場景的性能測試。主表一億條數據去 join 維表的 1 萬條數據，在不同的計算資源下，rebalance 相較于原生的 forward 方式有數倍的性能提升，而哈希相較于 rebalance 的方式又有數倍的性能提升，整體效果是比較可觀的。

針對維表 join 單條查詢效率比較低的問題，解決思路也很簡單，就是攢批，按照微批的方式去訪問 (mini-batch)?？梢栽?DDL 的定義中通過設置 lookup.async.batch.size 的值來指定批次的大小。除此之外，我們還在時間維度上引入了 Linger 機制來做限制，防止極端場景出現遲遲無法攢夠一批數據而導致時延比較高的情況，可以通過在 DDL 的定義中設置 lookup.async.batch.linger 的值來指定等待時間。

經過測試，mini-batch 的方式能夠帶來 15% ~ 50% 的性能提升。

Interval join 也是生產上一個使用比較頻繁的場景，這類業務的特點是流量非常大，比如 10 分鐘百 GB 級別。Interval join 兩條流的數據都會緩存在內部 state 中，任意一邊的數據到達都會獲取對面流相應時間范圍的數據去執行 join function，所以這種大流量的任務會有非常大的狀態。

對此我們選用了 RocksDB 來做狀態后端，但是進行了調參優化后效果仍不理想，任務運行一段時間之后會出現背壓，導致 RocksDB 的性能下降，CPU 的使用率也比較高。

通過分析我們發現，根本原因與 Flink 底層掃描 RocksDB 是基于前綴的掃描方式有關。因此解決思路也很簡單，根據查詢條件，精確地構建查詢的上下界，把前綴查詢變為范圍查詢。查詢條件依賴的具體上下界的 key 變為了 keyGroup+joinKey+namespace+timestamp[lower,upper]，可以精確地只查詢某些 timestamp 之間的數據，任務的背壓問題也得到了解決。而且數據量越大，這種優化帶來的性能提升越明顯。

Regular join 使用狀態來保存所有歷史數據，所以如果流量大也會導致狀態數據比較大。而它保存狀態是依賴 table.exec.state.ttl 參數，這個參數值比較大也會導致狀態大。

針對這種場景，我們改為使用外部存儲JimDB存儲狀態數據。目前只做了 inner join 的實現，實現機制如下：兩邊的流對 join 到的數據進行下發的同時，將所有數據以 mini-batch 的方式寫入到 JimDB，join 時會同時掃描內存中以及 JimDB 中對應的數據。此外，可以通過 JimDB ttl 的機制來實現 table.exec.state.ttl 功能，從而完成對過期數據的清理。

上述實現方式優缺點都比較明顯，優點是可以支持非常大的狀態，缺點是目前無法被 Flink checkpoint 覆蓋到。

對于 window 的優化，首先是窗口偏移量。需求最早來源于一個線上場景，比如我們想統計某個指標 2021 年 12 月 4 日 0 點 ~ 2021 年 12 月 5 日 0 點的結果， 但由于線上集群是東 8 區時間，所以實際統計的結果是 2021 年 12 月 4 日早上 8 點 ~ 2021 年 12 月 5 日早上 8 點的結果，這顯然不符合預期。因此這個功能最早是為了修復非本地時區跨天級別的窗口統計錯誤的問題。

在我們增加了窗口偏移量參數后，可以非常靈活地設置窗口的起始時間，能夠支持的需求也更廣泛。

其次，還存在另外一個場景：雖然用戶設定了窗口大小，但是他希望更早看到窗口當前的計算結果，便于更早地去做決策。因此我們新增了增量窗口的功能，它可以根據設置的增量間隔，觸發執行輸出窗口的當前計算結果。

對于端到端實時性要求不高的應用，可以選擇 Iceberg 作為下游的統一存儲。但是鑒于計算本身的特性、用戶 checkpoint 間隔的配置等原因，可能導致產生大量的小文件。Iceberg 的底層我們選用 HDFS 作為存儲，大量的小文件會對 Namenode 產生較大的壓力，所以就有了合并小文件的需求。

Flink 社區本身提供了基于 Flink batch job 的合并小文件的工具可以解決這個問題，但這種方式有點重，所以我們開發了算子級別的小文件合并的實現。思路是這樣的，在原生的 global commit 之后，我們新增了三個算子 compactCoordinator、 compactOperator 和 compactCommitter，其中 compactCoordinator 負責獲取待合并的 snapshot 并下發，compactOperator 負責 snapshot 的合并操作的執行，并且可以多個 compactOperator 并發執行，compactCommitter 負責合并后 datafiles 的提交。

我們在 DDL 的定義中新增了兩個參數，auto-compact 指定是否開啟合并文件的功能，compact.delta.commits 指定每提交多少次 commit 來觸發一次 compaction。

在實際的業務需求中，用戶可能會從 Iceberg 中讀取嵌套數據，雖然可以在 SQL 中指定讀取嵌套字段內部的數據，但是在實際讀取數據時是會將包含當前嵌套字段的所有字段都讀取到，再去獲取用戶需要的字段，而這會直接導致 CPU 和網絡帶寬負載的增高，所以就產生了如下需求：如何只讀取到用戶真正需要的字段？

解決這個問題，要滿足兩個條件，第一個條件是讀取 Iceberg 的數據結構 schema 只包含用戶需要的字段，第二個條件是 Iceberg 支持按列名去讀取數據，而這個本身已經滿足了，所以我們只需要實現第一個條件即可。

如上圖右側所示，結合之前的 tableSchema 和 projectFields 信息重構，生成了一個只包含用戶需要字段的新的數據結構 PruningTableSchema，并且作為 Iceberg schema 的輸入，通過這樣的操作實現了根據用戶的實際使用情況對嵌套結構進行列裁剪。圖中左下部的示例展示了用戶優化前后讀取嵌套字段的對比，可以看到基于 PruningTablesSchema 能夠對無用的字段進行有效的裁剪。

經過上述優化，CPU 使用率降低了 20%~30%。而且，在相同的數據量下，批任務的執行時間縮短了 20%~30%。

此外，我們還實現了一些其他優化，比如修復了 interval outer join 數據晚于 watermark 下發、且下游有時間算子時會導致的數據丟失問題，UDF 的復用問題，FlinkSQL 擴展 KeyBy 語法，維表數據預加載以及 Iceberg connector 從指定的 snapshot 去讀取等功能。

03落地案例

京東目前 FlinkSQL 線上任務 700+，占Flink總任務數的 15% 左右，FlinkSQL 任務累計峰值處理能力超過 1.1 億條/秒。目前主要基于社區的 1.12 版本進行了一些定制優化。

3.1 案例一

實時通用數據層 RDDM 流批一體化的建設。RDDM 全稱是 real-time detail data model - 實時明細數據模型，它涉及訂單、流量、商品、用戶等，是京東實時數倉的重要一環，服務了非常多的核心業務，例如黃金眼/商智、JDV、廣告算法、搜推算法等。

RDDM 層的實時業務模型與離線數據中 ADM 和 GDM 層的業務加工邏輯一致。基于此，我們希望通過 FlinkSQL 來實現業務模型的流批計算統一。同時這些業務也具備非常鮮明的特點，比如訂單相關的業務模型都涉及大狀態的處理，流量相關的業務模型對于端到端的實時性要求比較高。此外，某些特殊場景也需要一些定制化的開發來支持。

RDDM 的實現主要有兩個核心訴求：首先它的計算需要關聯的數據比較多的，大量的維度數據都存儲在 HBase 中；此外部分維度數據的查詢存在二級索引，需要先查詢索引表，從中取出符合條件的 key 再去維度表中獲取真正的數據。

針對上述需求，我們通過結合維表數據預加載的功能與維表 keyby 的功能來提升 join 的效率。針對二級索引的查詢需求，我們定制了 connector 來實現。

維表數據預加載的功能指在初始化的階段就將維表數據加載到內存中，這個功能結合 keyby 使用可以非常有效地減少緩存的數量，提高命中率。

部分業務模型關聯的歷史數據比較多，導致狀態數據比較大，目前我們是根據場景進行定制的優化。我們認為根本的解決方案是實現一套高效的基于 KV 的 statebackend，對于此功能的實現正在規劃中。

3.2 案例二

流量買賣黑產的輿情分析。它的主要流程如下：源端通過爬蟲獲取相關信息并寫入到 JMQ，數據同步到 JDQ 以后，通過 Flink 處理然后繼續寫下游的 JDQ。與此同時，通過 DTS 數據傳輸服務，將上游 JDQ 的數據同步到 HDFS，然后通過 Hive 表進行離線的數據加工。

此業務有兩個特點：首先，端到端的實時性要求不高，可以接受分鐘級別的延時；第二，離線和實時的加工邏輯一致。因此，可以直接把中間環節的存儲從 JDQ 換成 Iceberg，然后通過 Flink 去增量讀取，并通過 FlinkSQL 實現業務邏輯加工，即完成了流批兩套鏈路的完全統一。其中 Iceberg 表中的數據也可以供 OLAP 查詢或離線做進一步的加工。

上述鏈路端到端的時延在一分鐘左右，基于算子的小文件合并功能有效地提升了性能，存儲計算成本有了顯著的降低，綜合評估開發維護成本降低了 30% 以上。

04未來規劃

未來規劃主要分為以下兩個方面：

首先，業務拓展方面。我們會加大 FlinkSQL 任務的推廣，探索更多流批一體的業務場景，同時對產品形態進行打磨，加速用戶向 SQL 的轉型。同時，將平臺元數據與離線元數據做更深度的融合，提供更好的元數據服務。

其次，平臺能力方面。我們會繼續深挖 join 場景和大狀態場景，同時探索高效 KV 類型的狀態后端實現，并在統一計算和統一存儲的框架下不斷優化設計，以降低端到端時延。?