基于Flink table store建構實時數倉的現狀與發展趨勢思考

衆所周知，實時數倉落地是一個難點，尤其是金融行業，還沒有出現真正所謂的實時報表。金融行業個别案例的實時數倉是在較窄場景、較多限制下的嘗試，還不能夠稱之為實時數倉，如銀行普遍的實時報表業務都無法滿足。目前亟需設計實作一套能夠落地的金融行業的實時報表方案，來滿足業務場景對資料時效性越來越高的需求。

本文内容首先介紹了銀行業常見的實時場景和解決方案，然後針對銀行業報表依賴次元表計算的特點，提出了基于 Flink Table Store 作為資料存儲，進而建構流式數倉的解決方案。

01

金融行業實時數倉現狀分析

1.1 動賬場景介紹

數倉模組化有範式模組化和次元模組化，銀行業采用的是次元模組化，其中分為事實表和次元表。

事實表：刻畫行為的，一般用來統計交易筆數，交易金額，業務量等。

次元表：描述結果和狀态的，常見的使用者手機号、身份證号、所屬的機構等不經常更新的資料，但其中銀行業比較重要的有“賬戶餘額”，客戶餘額會随着動賬交易而頻繁更新。

本文以銀行典型的動賬場景為例，一次動賬操作其實是一個事務，至少操作兩張表，第一張比較好了解，就是交易流水表，記錄轉賬的一次行為；第二張則是使用者的屬性表，其中有一個字段是使用者的餘額，需要随着動賬的交易流水表同步更新，上面的兩個表是兩次轉賬的示例。

在這個轉賬場景下進行分析

• 流水表的特點：主要是 Insert 操作，記錄行為資訊，适合增量計算，如統開戶、取款、貸款、購買理财等事件行為。
• 應用的場景有實時營銷，如大額動賬提醒，工資代發，理财産品購買等；實時反欺詐的申請反欺詐、交易反欺詐；在貸後管理也有應用，如監控使用者入賬行為，提供給零貸貸後臨期催收、扣款等。
• 客戶屬性表的特點：主要是 Update 操作，記屬性資訊，客戶的存款、貸款、理财、基金、保險等産品的餘額是在次元表中，是以常使用次元表全量計算資産資訊，如資産餘額類的計算，計算某分支行的總存款餘額等。
• 應用的場景主要是實時報表、實時大屏：如對公 CRM、零售 CRM；經營管理；資産負債管理等。

針對于銀行業這兩種典型的動賬場景，有三種解決方案。下面逐個介紹不同方案适用的場景和有哪些局限。

1.2 基于 Kafka 的 ETL

該架構能夠解決的問題，大多是基于事實表的增量計算，已經在行内有大量的落地案例，但無法解決銀行業的基于次元表的全量計算。另外該方案很難形成規模化的資料分層複用，Kafka 存在資料無法查詢和長期持久化等問題。這種煙囪式的 case by case 開發階段，本行已經經曆過了，生産上也有大量的落地場景，實時任務達到了 300+個。

1.3 基于微批的 ELT

為了解決銀行業大量基于次元表的統計分析場景，來看一下進行了哪些方式的探索。總結來說，是一種先載入後分析，也就是 ELT 的方式。過程是這樣的，先實時采集-> 然後直接實時載入->最後在實時 OLAP 查詢階段進行邏輯的加工。

在ELT探索的的初期，我們采用過微批全量計算的方式，在資料實時地寫入到資料庫後，定時執行全量加工邏輯，類似于離線數倉有跑批的概念，隻不過每天跑批縮短到了小時級别或分鐘級别跑一次批，來達到準實時加工的效果。顯而易見，這種方式是不可取的，存在時效性差、跑批不穩定等問題。

1.4 基于視圖的 ELT

随着 MPP 資料庫的發展，查詢性能得到了極大的提升，本行使用 StarRocks 引擎，通過 View 視圖嵌套加工邏輯的方式也進行了探索，也就是把業務資料庫的資料以 CDC 方式，載入 MPP 資料庫的明細層，查詢分析邏輯使用 View 封裝，在查詢觸發時直接計算，這種方式也可以解決基于次元表的全量計算，但每次查詢資源消耗太大，支撐大資料高頻率的查詢操作比較困難，無法大範圍應用推廣。

1.5 動賬場景總結

基于事實表的增量計算已經在生産進行了大量的落地和實踐，本文主要是讨論銀行業基于次元表的全量計算場景，上述兩種解決方案雖然能夠解決一部分實時場景，但局限很大，目前階段來到了優化更新和未來方向選擇的節點。

為了解決銀行業基于次元表的實時 OLAP，必須把部分計算向前移動，在 Flink側計算。湖存儲 Flink Table Store 的出現，使基于次元表的全量計算成為了可能。也就是底層一部分轉化工作在Flink中計算，另一部分聚合計算等工作在 OLAP 資料庫中計算，兩者分攤一下計算時間和資源消耗。在未來，還是希望把全部加工邏輯，全部在Flink端分層完成，向着存算分離、流批一體的流式數倉方向發展。

02

基于 Flink Table Store

的金融行業流式數倉

2.1 Flink Table Store 介紹

2022 年釋出的 Flink Table Store，能夠很好地解決之前遇到的大量資料更新、全量存儲等問題，Table Store 是一個統一的存儲，用于在 Flink 中建構流式處理和批處理的動态表，支援高速資料攝取和快速的資料查詢。

• 是一種湖存儲格式，存儲和計算分離，導入資料時雙寫到資料檔案和日志系統。
• 支援流批寫入、流批讀取，支援快速 Update 操作。
• 還支援豐富的 OLAP 引擎，Hive、Trino 等，目前 StarRocks 也在支援湖存儲查詢分析，相信在不久的将來，StarRocks 也是能夠支援查詢 Flink Table Store。

了解詳情，請移步到官網：https://nightlies.apache.org/flink/flink-table-store-docs-release-0.2/

2.2 導入資料

在銀行業，業務資料庫仍然是以 Oracle 為主，全量資料初始化到 Flink Table Store 中，使用的 Oracle Connector 需要開發才能使用，同時需要支援 Filter、Project 等操作。采用 JDBC 連接配接以流式讀取資料庫的方式進行全量寫入到 Flink Table Store 中，同時在建表配置項中配置 changelog-producer = input，儲存完整的 changelog，為後續流寫和流讀作準備。

在完成了全量資料的初始化，後續增量的更新資料需要持續地寫入到 Flink Table Store 中，首先從 Oracle 中把資料實時地抽取出來，以 JSON 格式寫入到 Kafka，供後續多個場景複用 Topic。在銀行業，資料庫管理較為嚴格，能夠實時擷取業務資料比其它行業要解決更多方面的困難。下面模拟一下動賬過程：

1. 客戶表初始狀态為客戶 1、2、3 的餘額分别為 100、200、300。
2. 客戶 1 轉入 100 元，則客戶表執行 Update 操作，使客戶 1 的餘額從 100 -> 200。
3. 客戶 2 轉出 100 元，則客戶表執行 Update 操作，使客戶 2 的餘額從 200 -> 100。
4. 資料庫的 Update 操作，使用 CDC 工具把 changelog 資訊以 json 格式寫入到 Kafka 隊列。後續啟動 Flink SQL 任務消費 Kafka，将 changelog 流寫入到 Flink Table Store 中。

在拿到增量的 CDC 資料後，需要把增量更新資料和曆史全量資料進行融合，才能夠得到完整最新的全量資料。這裡有兩個問題需要探讨：

第一：全量資料和增量資料為什麼分開寫入呢？

• 避免實時資料抽取多份，統一寫入 Kafka，後續多個實時場景可以複用；
• 離線資料全量初始化可能是一個經常性的操作，比如每周進行一次全量的初始化。

第二：全量資料和增量資料如何保證銜接正确呢？

• 次元表正常情況下是有主鍵的表，這樣就能夠保證有幂等的特性，隻需要保證增量資料早于全量資料就行了。比如增量資料5點開始啟動寫入到 Kafka，全量資料 6 點開始全量同步，增量寫入任務在全量同步結束後開始指定早于 6 點的資料開始消費就可以保證資料的完整性了。

另外在寫入 Flink Table Store 時需要配置 table.exec.sink.upsert-materialize= none，避免産生 Upsert 流，以保證 Flink Table Store 中能夠儲存完整的 changelog，為後續的流讀操作做準備。

2.3 查詢資料

第一種方式，Batch 模式。

曆史存量和實時寫入的資料，均能夠線上 OLAP 分析。支援流寫批讀，Batch 模式讀取資料是從 Snapshot 檔案中讀取，checkpoint interval 周期内可見。支援多種查詢引擎 Hive、Trino、Flink SQL 等，全局有序 Sorted File 的 Data Skipping，Sort Aggregation and Sort Merge Join 特性等。

這裡可以任意時間檢視各個分支行的存款餘額，或者分析客戶的明細資訊等。

第二種方式，Streaming 模式。

以 Streaming 模式啟動查詢時，任務會持續線上運作，當客戶 1 進行轉賬操作時，如轉入 100 元，變成了 200 元。此時在實時數倉發生的過程如下：

這個過程有如下特點：

• 流批統一。存儲統一，Snapshot+Log，存量資料讀取 Snapshot，增量資料讀取 changelog，hybird 讀取全量實時資料。查詢統一，離線和實時使用相同的 SQL 語句。Streaming 模式開啟 mini-batch 減少聚合語句的備援changelog 輸出。
• 減少物化。FTS 中有完整的 changelog，避免 Flink State 中生成物化節點。
• 時延較低。changelog 使用 File 存儲，代價低，時延高；使用 Kafka 存儲，代價高，時延低。
• 資料驅動，而不是時間排程驅動或者查詢時才開始觸發計算。

2.4 導出資料

最終的結果資料，如果查詢頻率不高，可以直接使用 Flink 1.16 提供的 SQL Gateway 功能；如果查詢頻率較高，可以再以流式寫出到外部的資料庫中，提供穩定的線上服務能力。

2.5 未來發展

實作真正端到端的流式數倉，既能夠支援實時資料和完整的 changelog，也支援批量導入離線資料，當資料在源頭發生變化時就能捕捉到這一變化，并支援對它做逐層分析，讓所有資料實時流動起來，并且對所有流動中的資料都可以實時查詢，是以純流的方式而不是微批的方式流動。在這個過程中，資料是主動的，而查詢是被動的，分析由資料的變化來驅動。

數倉的分層可以解決實時資料的複用，多名額随着資料的實時流動而實時變化，從另一種角度說也是在用空間換取時間。離線資料和實時資料共同存儲在 Flink Table Store 中，使用廉價的存儲和存算分離更加靈活的進行彈性計算。離線分析 sql 和實時分析 sql 式完全一樣的，最終達到流批一體的效果。總結如下：

• 存算分離的湖存儲，FTS 提供完善的湖存儲 Table Format，提供準實時 OLAP 分析。
• 能夠存儲全量資料，每層資料能夠可查，支援 Batch 和 Streaming 兩種模式。
• 支援大量資料更新，有序的 LSM 結構，支援快速的 Update 操作。
• 支援流批寫流批讀，尤其是能夠流式讀取，流式資料從 Log System 中讀取。
• 完整的 changelog，支援全部流程傳遞完整的+I、-U、+U、-D 操作，減少 Flink State 中的物化節點。
• 真正實作流批一體，流批統一 Flink SQL，流批統一存儲。

那為什麼不直接采用這種架構進行建構呢？目前階段這個架構還無法完全落地，比如其中聚合計算有大量的撤銷動作、多個層之間的實時資料流動需要大量的資源和調試技能等，不過随着技術的發展，相信流式數倉一定會到來。

2.6 流式數倉落地進展

目前階段，既然多層的流式數倉落地還有一定的距離，那麼在加入 Flink Table Store 後，在原有 ELT 架構的基礎上，進行優化更新，看看帶來了哪些變化。

在整個計算過程中，直接把原始資料寫入 Flink Table Store，使之存儲曆史全量和實時更新的表資料，然後計算邏輯使用 Flink SQL 實作，最後把初步彙總的資料寫入到 StarRocks 中。原始明細資料在 Flink 中計算，極大的減少了 StarRocks 的計算邏輯，Flink 和 OLAP 引擎兩者協調配合，共同提供端到端的實時報表業務。這種架構在我們在生産上也已經行進了初步的嘗試，效果非常顯著。