百信銀行基於 Apache Hudi 實時數據湖演進方案

簡介： 本文介紹了百信銀行實時計算平台的建設情況，實時數據湖構建在 Hudi 上的方案和實踐方法，以及實時計算平台集成 Hudi 和使用 Hudi 的方式。

本文介紹了百信銀行實時計算平台的建設情況，實時數據湖構建在 Hudi 上的方案和實踐方法，以及實時計算平台集成 Hudi 和使用 Hudi 的方式。內容包括：

1. 背景
2. 百信銀行基於 Flink 的實時計算平台設計與實踐
3. 百信銀行實時計算平台與實時數據湖的集成實踐
4. 百信銀行實時數據湖的未來
5. 總結

  

一、背景

百信銀行，全稱為 “中信百信銀行股份有限公司”，是首家獲批獨立法人形式的直銷銀行。作為首家國有控股的互聯網銀行，相比於傳統金融行業，百信銀行對數據敏捷性有更高的要求。

數據敏捷，不僅要求數據的准確性，還要求數據到達的實時性，和數據傳輸的安全性。為了滿足我行數據敏捷性的需求，百信銀行大數據部承擔起了建設實時計算平台的職責，保證了數據快速，安全且標准得在線送達。

受益於大數據技術的發展和更新迭代，目前廣為人知的批流一體的兩大支柱分別是：“統一計算引擎” 與 “統一存儲引擎”。

• Flink，作為大數據實時計算領域的佼佼者，1.12 版本的發布讓它進一步提升了統一計算引擎的能力；
• 同時隨着數據湖技術 Hudi 的發展，統一存儲引擎也迎來了新一代技術變革。

在 Flink 和 Hudi 社區發展的基礎上，百信銀行構建了實時計算平台，同時將實時數據湖 Hudi 集成到實時計算平台之上。結合行內數據治理的思路，實現了數據實時在線、安全可靠、標准統一，且有敏捷數據湖的目標。

二、百信銀行基於 Flink 的實時計算平台設計與實踐

1. 實時計算平台的定位

實時計算平台作為行級實時計算平台，由大數據 IaaS 團隊自主研發，是一款實現了實時數據 ”端到端“ 的在線數據加工處理的企業級產品。

• 其核心功能具備了實時采集、實時計算、實時入庫、復雜時間處理、規則引擎、可視化管理、一鍵配置、自主上線，和實時監控預警等。
• 目前其支持的場景有實時數倉、斷點召回、智能風控、統一資產視圖、反欺詐，和實時特征變量加工等。
• 並且，它服務着行內小微、信貸、反欺詐、消金、財務，和風險等眾多業務線。

截止目前，在線穩定運行的有 320+ 的實時任務，且在線運行的任務 QPS 日均達到 170W 左右。

2. 實時計算平台的架構

按照功能來划分的話，實時計算平台的架構主要分為三層：

■ 1）數據采集層

采集層目前主要分為兩個場景：

• 第一個場景是采集 MySQL 備庫的 Binlog 日志到 Kafka 中。我行所使用的數據采集方案並沒有采用業界普遍用的如 Canal，Debezium 等現有的 CDC 方案。

  1、因為我們的 MySQL 版本為百信銀行內部的版本，Binlog 協議有所不同，所以現有的技術方案不能很好的支持兼容我們獲取 Binlog 日志。

  2、同時，為了解決我們數據源 MySQL 的備庫隨時可能因為多機房切換，而造成采集數據丟失的情況。我們自研了讀取 MySQL Binlog 的 Databus 項目，我們也將 Databus 邏輯轉化成了 Flink 應用程序，並將其部署到了 Yarn 資源框架中，使 Databus 數據抽取可以做到高可用，且資源可控。

• 第二個場景是，我們對接了第三方的應用，這個第三方應用會將數據寫入 Kafka，而寫入 Kafka 有兩種方式：

  1、一種方式是依據我們定義的 Json shcema 協議。

（UMF協議：{col_name:””,umf_id":"","umf_ts":,"umf_op_":"i/u/d"}）
協議定義了 ”唯一 id”，”時間戳“ 和 ”操作類型“。根據此協議，用戶可以指定對該消息的操作類型，分別是 "insert"，"update" 和 "delete"，以便下游對消息進行針對性處理。

  2、另外一種方式，用戶直接把 JSON 類型的數據寫到 kafka 中，不區分操作類型。 

■ 2）數據計算轉換層

消費 Kafka 數據進行一層轉換邏輯，支持用戶自定義函數，將數據標准化，做敏感數據的脫敏加密等。

■ 3）數據存儲層

數據存儲到 HDFS，Kudu，TiDB，Kafka，Hudi，MySQL 等儲存介質中。

在上圖所示的架構圖中，我們可以看到整體實時計算平台支持的主要功能有:

• 開發層面：

  1、支持標准化的 DataBus 采集功能，該功能對於支持 MySQL Binglog 同步到 Kafka 做了同步適配，不需要用戶干預配置過多。用戶只需要指定數據源 MySQL 的實例就可以完成到 Kafka 的標准化同步。
  2、支持用戶可視化編輯 FlinkSQL。
  3、支持用戶自定義 Flink UDF 函數。
  4、支持復雜事件處理(CEP)。
  5、支持用戶上傳打包編譯好 Flink 應用程序。

• 運維層面：

  1、支持不同類型任務的狀態管理，支持savepoint。
  2、支持端到端的延遲監控，告警。

在實時計算平台升級迭代的過程中，社區 Flink 版本之間存在一些向下不兼容的情況。為了平滑的升級 Flink 版本，我們對計算引擎的多版本模塊進行統一的抽象，將多版本之間做了嚴格的 JVM 級別隔離，使版本之間不會產生 Jar 包沖突，Flink Api 不兼容的情況。

如上圖所示，我們將不同的 Flink 版本封裝到一個獨立的虛擬機中，使用 Thrift Server 啟動一個獨立的 JVM 虛擬機，每個版本的 Flink 都會有一個獨立的 Thrift Server。在使用的過程中，只要用戶顯示指定的 Flink 版本，Flink 應用程序就會被指定的 Thrift Server 啟動。同時，我們也將實時計算的后端服務嵌入一個常用的 Flink 版本，避免因為啟動 Thrift Server 而占用過多的啟動時間。

同時為了滿足金融系統高可用和多備的需求，實時計算平台也開發了多 Hadoop 集群的支持，支持實時計算任務在失敗后可以遷移到備集群上去。整體的方案是，支持多集群 checkpoint，savepoint，支持任務失敗后，可以在備機房重啟實時任務。

三、百信銀行實時計算平台與實時數據湖集成實踐

在介紹本內容之前，我們先來了解一些我行目前在數據湖的現狀。目前的實時數據湖，我行依然采用主流的 Lambda 架構來構建數據倉庫。

1. Lambda

Lambda 架構下，數倉的缺點：

• 同樣的需求，開發和維護兩套代碼邏輯：批和流兩套邏輯代碼都需要開發和維護，並且需要維護合並的邏輯，且需同時上線；
• 計算和存儲資源占用多：同樣的計算邏輯計算兩次，整體資源占用會增多；
• 數據具有二義性：兩套計算邏輯，實時數據和批量數據經常對不上，准確性難以分辨；
• 重用 Kafka 消息隊列：Kafka 保留往往按照天或者月保留，不能全量保留數據，無法使用現有的 adhoc 查詢引擎分析。

2. Hudi

為了解決 Lambda 架構的痛點，我行准備了新一代的數據湖技術架構，同時我們也花大量的時間調研了現有的數據湖技術，最終選擇 Hudi 作為我們的存儲引擎。

• Update / Delete 記錄：Hudi 使用細粒度的文件/記錄級別索引，來支持 Update / Delete 記錄，同時還提供寫操作的事務保證，支持 ACID 語義。查詢會處理最后一個提交的快照，並基於此輸出結果；
• 變更流：Hudi 對獲取數據變更提供了流的支持，可以從給定的時間點獲取給定表中已 updated / inserted / deleted 的所有記錄的增量流，可以查詢不同時間的狀態數據；
• 技術棧統一：可以兼容我們現有的 adhoc 查詢引擎 presto，spark。
• 社區更新迭代速度快：已經支持 Flink 兩種不同方式的的讀寫操作，如 COW 和 MOR。

在新的架構中可以看到，我們將實時和批處理貼源層的數據全部寫到 Hudi 存儲中，並重新寫入到新的數據湖層 datalake（Hive 的數據庫）。出於歷史的原因，為了兼容之前的數據倉庫的模型，我們依然保留之前的 ODS 層，歷史的數倉模型保持不變，只不過 ODS 貼源層的數據需要從 datalake 層獲取。

• 首先，我們可以看到，對於新的表的入倉邏輯，我們通過實時計算平台使用 Flink 寫入到 datalake 中（新的貼源層，Hudi 格式存儲），數據分析師和數據科學家，可以直接使用 datalake 層的數據進行數據分析和機器學習建模。如果數據倉庫的模型需要使用 datalake 的數據源，需要一層轉換 ODS 的邏輯，這里的轉換邏輯分為兩種情況：

  1、第一種，對於增量模型，用戶只需要將最新 datalake 的分區使用快照查詢放到 ODS 中即可。
  2、第二種，對於全量模型，用戶需要把 ODS 前一天的快照和 datalake 最新的快照查詢的結果進行一次合並，形成最新的快照再放到 ODS 當前的分區中，以此類推。

我們這么做的原因是，對於現有的數倉模型不用改造，只是把 ODS 的數據來源換成 datalake，時效性強。同時滿足了數據分析和數據科學家准實時獲取數據的訴求。

• 另外，對於原始的 ODS 存在的數據，我們開發了將 ODS 層的數據進行了一次初始化入 datalake 的腳本。

  1、如果 ODS 層數據每天是全量的快照，我們只將最新的一次快照數據初始化到 datalake 的相同分區，然后實時入 datalake 的鏈路接入；
  2、如果 ODS 層的數據是增量的，我們暫時不做初始化，只在 datalake 中重新建一個實時入湖的鏈路，然后每天做一次增量日切到 ODS 中。

• 最后，如果是一次性入湖的數據，我們使用批量入湖的工具導入到 datalake 中即可。

整體湖倉轉換的邏輯如圖：

3. 技術挑戰

• 在我們調研的初期，Hudi 對 Flink 的支持不是很成熟，我們對 Spark - StrunctStreaming 做了大量的開發和測試。從我們 PoC 測試結果上看，

  1、如果使用無分區的 COW 寫入的方式，在千萬級寫入量的時候會發現寫入越來越慢；
  2、后來我們將無分區的改為增量分區的方式寫入，速度提升了很多。

之所以會產生這個問題，是因為 spark 在寫入時會讀取 basefile 文件索引，文件越大越多，讀取文件索引就會越慢，因此會產生寫入越來越慢的情況。

• 同時，隨着 Flink 對 hudi 支持越來越好，我們的目標是打算將 Hudi 入湖的功能集成到實時計算平台。因此，我們把實時計算平台對 Hudi 做了集成和測試，期間也遇到一些問題，典型的問題有：

  1、類沖突
  2、不能找到 class 文件
  3、rocksdb 沖突

為了解決這些不兼容的問題，我們將對 Hudi 的依賴，重新構造了一個獨立的模塊，這個工程只是把 Hudi 的依賴打包成一個 shade package。

  4、當有依賴沖突時，我們會把 Flink 模塊相關或者 Hudi 模塊相關的沖突依賴 exclude 掉。 5、而如果有其他依賴包找不到的情況，我們會把相關的依賴通過 pom 文件引入進來。 

• 在使用 Hudi on Flink 的方案中，也遇到了相關的問題，比如，checkpoint 太大導致 checkpoint 時間過長而引起的失敗。這個問題，我們設置狀態的 TTL 時間，把全量 checkpoint 改為增量 checkpoint，且提高並行度來解決。
• COW 和 MOR 的選擇。目前我們使用的 Hudi 表以 COW 居多，之所以選擇 COW，

  1、第一是因為我們目前歷史存量 ODS 的數據都是一次性導入到 datalake 數據表中，不存在寫放大的情況。
  2、另外一個原因是，COW 的工作流比較簡單，不會涉及到 compaction 這樣的額外操作。

如果是新增的 datalake 數據，並且存在大量的 update，並且實時性要求較高的情況下，我們更多的選擇 MOR 格式來寫，尤其寫 QPS 比較大的情況下，我們會采用異步 compaction 的操作，避免寫放大。除了這種情況外，我們還是會更傾向以 COW 的格式來寫。

四、百信銀行實時數據湖的未來

在我行實時數據湖的架構中，我們的目標是將實時數倉的整個鏈路構建在Hudi之上，架構體系如圖：

我們整體的目標規划是替代 kafka，把 Hudi 作為中間存儲，將數倉建設在 Hudi 之上，並以 Flink 作為流批一體計算引擎。這樣做的好處有：

• MQ 不再擔任實時數據倉庫存儲的中間存儲介質，而 Hudi 存儲在 HDFS 上，可以存儲海量數據集；
• 實時數據倉庫中間層可以使用 OLAP 分析引擎查詢中間結果數據；
• 真正意義上的批流一體，數據 T+1 延遲的問題得到解決；
• 讀時 Schema 不再需要嚴格定義 Schema 類型，支持 schema evolution；
• 支持主鍵索引，數據查詢效率數倍增加，並且支持 ACID 語義，保證數據不重復不丟失；
• Hudi 具有 Timeline 的功能，可以更多存儲數據中間的狀態數據，數據完備性更強。

五、總結

本文介紹了百信銀行實時計算平台的建設情況，實時數據湖構建在 Hudi 上的方案和實踐方法，以及實時計算平台集成 Hudi 和使用 Hudi 的方式。

在使用 Hudi 的過程中，也遇到一些問題，由衷感謝社區同學的幫助。特別感謝社區 Danny chan，leesf 解疑答惑。在實時數據湖架構體系下，構建我們實時數倉，流批一體方案還是在摸索中。

僅以此篇，希望能給其他正在建設實時計算平台，和使用 Hudi 構建實時數據湖的同學提供一些參考。

原文鏈接
本文為阿里雲原創內容，未經允許不得轉載。