基於Apache Hudi 的CDC數據入湖

• 2021 年 10 月 24 日
• 筆記

作者：李少鋒

文章目錄：

一、CDC背景介紹

二、CDC數據入湖

三、Hudi核心設計

四、Hudi未來規劃

1. CDC背景介紹

首先我們介紹什麼是CDC？CDC的全稱是Change data Capture，即變更數據捕獲，它是資料庫領域非常常見的技術，主要用於捕獲資料庫的一些變更，然後可以把變更數據發送到下游。它的應用比較廣，可以做一些數據同步、數據分發和數據採集，還可以做ETL，今天主要分享的也是把DB數據通過CDC的方式ETL到數據湖。

對於CDC，業界主要有兩種類型：

• 基於查詢，客戶端會通過SQL方式查詢源庫表變更數據，然後對外發送。
• 基於日誌，這也是業界廣泛使用的一種方式，一般是通過binlog方式，變更的記錄會寫入binlog，解析binlog後會寫入消息系統，或直接基於Flink CDC進行處理。

它們兩者是有區別的，基於查詢比較簡單，是入侵性的，而基於日誌是非侵入性，對數據源沒有影響，但binlog的解析比較複雜一些。

基於查詢和基於日誌，分別有四種實現技術，有基於時間戳、基於觸發器和快照，還有基於日誌的，這是實現CDC的技術，下面是幾種方式的對比。

通過這個表格對比可以發現基於日誌的綜合最優，但解析比較複雜，但業界有很多開源的binlog的解析器，比較通用和流行的有Debezium、Canal，以及Maxwell。基於這些binlog解析器就可以構建ETL管道。

下面來看下業界比較流行的一種CDC入倉架構。

整個數據入倉是分實時流是離線流，實時流解析binlog，通過Canal解析binlog，然後寫入Kafka，然後每個小時會把Kafka數據同步到Hive中；另外就是離線流，離線流需要對同步到Hive的貼源層的表進行拉取一次全量，如果只有前面的實時流是數據是不全的，必須通過離線流的SQL Select把全量導入一次數據，對每張ODS表會把存量數據和增量數據做一個Merge。這裡可以看到對於ODS層的實時性不夠，存在小時、天級別的延遲。而對ODS層這個延時可以通過引入Apache Hudi做到分鐘級。

2. CDC數據入湖方法

基於CDC數據的入湖，這個架構非常簡單。上游各種各樣的數據源，比如DB的變更數據、事件流，以及各種外部數據源，都可以通過變更流的方式寫入表中，再進行外部的查詢分析，整個架構非常簡單。

架構雖然簡單，但還是面臨很多挑戰。以Apache Hudi數據湖為例，數據湖是通過文件存儲各種各樣的數據， 對於CDC的數據處理需要對湖裡某部分文件進行可靠地、事務性變更，這樣可以保證下游查詢不會看到部分結果，另外對CDC數據需要高效的做更新、刪除操作，這就需要快速定位到更改的文件，另外是對於每小批量的數據寫入，希望能夠自動處理小文件，避免繁雜的小文件處理，還有面向查詢的布局優化，可以通過一些技術手段如Clustering改造文件布局，對外提供更好的查詢性能。

而Apache Hudi是怎麼應對這些挑戰的呢？首先支援事務性寫入，包括讀寫之間的MVCC機制保證寫不影響讀，也可以控制事務及並發保證，對於並發寫採用OCC樂觀鎖機制，對更新刪除，內置一些索引及自定義保證更新、刪除比較高效。另外是面向查詢優化，Hudi內部會自動做小文件的管理，文件會自動長到用戶指定的文件大小，如128M，這對Hudi來說也是比較核心的特性。另外Hudi提供了Clustering來優化文件布局的功能。

下圖是典型CDC入湖的鏈路。上面的鏈路是大部分公司採取的鏈路，前面CDC的數據先通過CDC工具導入Kafka或者Pulsar，再通過Flink或者是Spark流式消費寫到Hudi里。第二個架構是通過Flink CDC直聯到MySQL上游數據源，直接寫到下游Hudi表。

其實，這兩條鏈路各有優缺點。第一個鏈路統一數據匯流排，擴展性和容錯性都很好。對於第二條鏈路，擴展性和容錯性會稍微差點，但由於組件較少，維護成本相應較低。

這是阿里雲資料庫OLAP團隊的CDC入湖鏈路，因為我們我們做Spark的團隊，所以我們採用的Spark Streaming鏈路入湖。整個入湖鏈路也分為兩個部分：首先有一個全量同步作業，會通過Spark做一次全量數據拉取，這裡如果有從庫可以直連從庫做一次全量同步，避免對主庫的影響，然後寫到Hudi。然後會啟動一個增量作業，增量作業通過Spark消費阿里雲DTS里的binlog數據來將binlog准實時同步至Hudi表。全量和增量作業的編排藉助了Lakehouse的作業自動編排能力，協調全量和增量作業，而對於全量和增量銜接時利用Hudi的Upsert語義保證全增量數據的最終的一致性，不會出現數據偏多和偏少的問題。

在Lakehouse的CDC入湖鏈路中，我們團隊也做了一些優化。

第一個是原庫的Schema變更處理，我們對接的客戶某些列的增加、刪除或者修改某些列的場景。在Spark寫Hudi之前會做Schema的檢驗，看這個Schema是不是合法，如果合法就可以正常寫入，如果不合法的話，則會寫入失敗，而刪除欄位會導致Schema校驗不合法，導致作業失敗，這樣穩定性是沒有保證的。因此我們會捕捉Schema Validation的異常，如果發現是減少了欄位，我們會把之前的欄位做自動補全，然後做重試，保證鏈路是穩定的。

第二個有些客戶表沒有主鍵或者主鍵不合理，比如採用更新時間欄位作為主鍵，或者設置會變化的分區欄位，這時候就會導致寫入Hudi的數據和源庫表數據對不上。因此我們做了一些產品層面的優化，允許用戶合理設置主鍵和分區映射，保證同步到Hudi里和源庫是數據完全對齊的。

還有一個常見需求是用戶在上游庫中增加一個表，如果使用表級別同步的話，新增表在整個鏈路是無法感知的，也就無法同步到Hudi中，而在Lakehouse中，我們可以對整庫進行同步，因此在庫中新增表時，會自動感知新增表，將新增表數據自動同步到Hudi，做到原庫增加表自動感知的能力。

還有一個是對CDC寫入時候性能優化，比如拉取的一批數據中包含Insert、Update、Delete等事件，是否一直使用Hudi的Upsert方式寫入呢？這樣控制比較簡單，並且Upsert有數據去重能力，但它帶來的問題是找索引的效率低，而對於Insert方式而言，不需要找索引，效率比較高。因此對於每一批次數據會判斷是否都是Insert事件，如果都是Insert事件就直接Insert方式寫入，避免查找文件是否更新的開銷，數據顯示大概可以提升30%~50%的性能。當然這裡也需要考慮到DTS異常，重新消費數據時，恢復期間不能直接使用Insert方式，否則可能會存在數據重複，對於這個問題我們引入了表級別的Watermark，保證即使在DTS異常情況下也不會出現數據重複問題。

3. Hudi核心設計

接著介紹下Hudi 的定位，根據社區最新的願景，Hudi的定義是流式數據湖平台，它支援海量數據更新，內置表格式以及支援事務的儲存，一系列列表服務包括Clean、Archive、Compaction、Clustering等，以及開箱即用的數據服務，以及本身自帶的運維工具和指標監控，提供很好的運維能力。

這是Hudi官網的圖，可以看到Hudi在整個生態里是做湖存儲，底層可以對接HDFS以及各種雲廠商的對象存儲，只要兼容Hadoop協議接。上游是入湖的變化事件流，對上可以支援各種各樣的數據引擎，比如presto、Spark以及雲上產品；另外可以利用Hudi的增量拉取能力藉助Spark、Hive、Flink構建派生表。

整個Hudi體系結構是非常完備的，其定位為增量的處理棧。典型的流式是面向行，對數據逐行處理，處理非常高效。

但面向行的數據里沒有辦法做大規模分析做掃描優化，而批處理可能需要每天全量處理一次，效率相對比較低。而Hudi引入增量處理的概念，處理的數據都是某一時間點之後的，和流處理相似，又比批處理高效很多，並且本身是面向數據湖中的列存數據，掃描優化非常高效。

而回顧Hudi的發展歷史。2015年社區主席發表了一篇增量處理的文章，16年在Uber開始投入生產，為所有資料庫關鍵業務提供了支撐；2017年，在Uber支撐了100PB的數據湖，2018年隨著雲計算普及，吸引了中國外的使用者；19年Uber把它捐贈到Apache進行孵化；2020年一年左右的時間就成為了頂級項目，採用率增長了超過10倍；2021年Uber最新資料顯示Hudi支援了500PB數據湖，同時對Hudi做了很多增強，像Spark SQL DML和Flink的集成。最近字節跳動推薦部門分享的基於Hudi的數據湖實踐單表超過了400PB，總存儲超過了1EB，日增PB級別。

經過幾年的發展，中國外採用Hudi的公司非常多，比如公有雲的華為雲、阿里雲、騰訊雲以及AWS，都集成了Hudi，阿里雲也基於Hudi構建Lakehouse。字節跳動的整個數倉體系往湖上遷移也是基於Hudi構建的，後面也會有相應的文章分享他們基於Flink+Hudi的數據湖的日增PB數據量的實踐。同時像百度、快手頭部互聯網大廠都有在使用。同時我們了解銀行、金融行業也有工商銀行、農業銀行、百度金融、百信銀行也有落地。遊戲領域包括了三七互娛、米哈游、4399，可以看到Hudi在各行各業都有比較廣泛的應用。

Hudi的定位是一套完整的數據湖平台，最上層面向用戶可以寫各種各樣的SQL，Hudi作為平台提供的各種能力，下面一層是基於SQL以及編程的API，再下一層是Hudi的內核，包括索引、並發控制、表服務，後面社區要構建的基於Lake Cache構建快取，文件格式是使用的開放Parquet、ORC、HFile存儲格式，整個數據湖可以構建在各種雲上。

後面接著介紹Hudi的關鍵設計，這對我們了解Hudi非常有幫助。首先是文件格式，它最底層是基於Fileslice的設計，翻譯過來就是文件片，文件片包含基本文件和增量日誌文件。基本文件就是一個Parquet或者是ORC文件，增量文件是log文件，對於log文件的寫入Hudi里編碼了一些block，一批Update可以編碼成一個數據塊，寫到文件里。而基礎文件是可插拔，可以基於Parquet，最新的9.0版本已經支援了ORC。還有基於HFile，HFile可用作元數據表。

Log文件里保存了一系列各種各樣的數據塊，它是有點類似於資料庫的重做日誌，每個數據版本都可以通過重做日誌找到。對於基礎文件和Log文件通過壓縮做合併形成新的基礎文件。Hudi提供了同步和非同步的兩種方式，這為用戶提供了很靈活的選擇，比如做可以選擇同步Compaction，如果對延遲不敏感，而不需要額外非同步起一個作業做Compaction，或者有些用戶希望保證寫入鏈路的延遲，可以非同步做Compaction而不影響主鏈路。

Hudi基於File Slice上有個File Group的概念，File Group會包含有不同的File Slice，也File Slice構成了不同的版本，Hudi提供了機制來保留元數據個數，保證元數據大小可控。

對於數據更新寫入，盡量使用append，比如之前寫了一個Log文件，在更新時，會繼續嘗試往Log文件寫入，對於HDFS這種支援append語義的存儲非常友好，而很多雲上對象存儲不支援append語義，即數據寫進去之後不可更改，只能新寫Log文件。對於每個文件組也就是不同FileGroup之間是互相隔離的，可以針對不同的文件組做不同的邏輯，用戶可以自定義演算法實現，非常靈活。

基於Hudi FileGroup的設計可以帶來不少收益。比如基礎文件是100M，後面對基礎文件進行了更新50M數據，就是4個FileGroup，做Compaction合併開銷是600M，50M只需要和100M合，4個150M開銷就是600M，這是有FileGroup設計。還是有4個100M的文件，也是做了更新，每一次合，比如25M要和400M合併，開銷是1200M，可以看到採用FileGroup的設計，合併開銷減少一半。

還有表格式。表格式的內容是文件在Hudi內是怎麼存的。首先定義了表的根路徑，然後寫一些分區，和Hive的文件分區組織是一樣的。還有對錶的Schema定義，表的Schema變更，有一種方式是元數據記錄在文件里，也有的是藉助外部KV存儲元數據，兩者各有優缺點。

Hudi基於Avro格式表示Schema，因此對Schema的Evolution能力完全等同於Avro Schema的Evolution能力，即可以增加欄位以及向上兼容的變更，如int變成long是兼容的，但long變成int是不兼容的。

當前現在社區已經有方案支援Full Schema Evolution，即可以增加一個欄位，刪去一個欄位，重命名，也就是變更一個欄位。

還有一個是Hudi的索引設計。每一條數據寫入Hudi時，都會維護數據主鍵到一個文件組ID的映射，這樣在做更新、刪除時可以更快的定位到變更的文件。

右邊的圖裡有個訂單表，可以根據日期寫到不同的分區里。下面就是用戶表，就不需要做分區，因為它的數據量沒有那麼大，變更沒那麼頻繁，可以使用非分區的表。

對於分區表及變更頻繁的表，在使用Flink寫入時，利用Flink State構建的全局索引效率比較高。整個索引是可插拔的，包括Bloomfilter、 HBase高性能索引。在位元組場景中， Bloomfilter過濾器完全不能滿足日增PB的索引查找，因此他們使用HBase高性能索引，因此用戶可根據自己的業務形態靈活選擇不同索引的實現。在有不同類型索引情況下可以以較低代價支援遲到的更新、隨機更新的場景。

另外一個設計是並發控制。並發控制是在0.8之後才引入的。Hudi提供樂觀鎖機制來處理並發寫問題，在提交的時候檢查兩個變更是否衝突，如果衝突就會寫入失敗。對於表服務如Compaction或者是Clustering內部沒有鎖，Hudi內部有一套協調機制來避免鎖競爭問題。比如做Compaction，可以先在timeline上先打一個點，後面完全可以和寫入鏈路解耦，非同步做Compaction。

例如左邊是數據攝取鏈路，數據每半個小時攝取一次，右邊是非同步刪除作業，也會變更表，並且很有可能和寫入修改衝突，會導致這個鏈路一直失敗，平台無故的消耗CPU資源，現在社區針對這種情況也有改進方案，希望儘早檢測並發寫入的衝突，提前終止，減少資源浪費。

另外一個設計是元數據表。因為Hudi最開始是基於HDFS構建和設計，沒有太多考慮雲上存儲場景，導致在雲上FileList非常慢。因此在0.8版本，社區引入了Metadata Table，Metadata Table本身也是一張Hudi表，它構建成一張Hudi，可以復用Hudi表等各種表服務。Metadata Table表文件里會存分區下有的所有文件名以及文件大小，每一列的統計資訊做查詢優化，以及現在社區正在做的，基於Meta Table表構建全局索引，每條記錄對應每個文件ID都記錄在Meta table，減少處理Upsert時查詢待更新文件的開銷，也是上雲必備。

4. Hudi未來規劃

對未來的規劃，如基於Pulsar、Hudi構建Lakehouse，這是StreamNative CEO提出的Proposal，想基於Hudi去構建Pulsar分層的存儲。在Hudi社區，我們也做了一些工作，想把Hudi內置的工具包DeltaStreamar內置Pulsar Source，現在已經有PR了，希望兩個社區聯繫可以更緊密。Pular分層存儲內核部分StreamNative有同學正在做。

最近幾天已經發布了0.9.0重要的優化和改進。首先集成了Spark SQL，極大降低了數據分析人員使用Hudi的門檻。

Flink集成Hudi的方案早在Hudi的0.7.0版本就有了，經過幾個版本的迭代，Flink集成Hudi已經非常成熟了，在字節跳動等大公司已經在生產使用。Blink團隊做的一個CDC的Format集成，直接把Update、Deltete事件直接存到Hudi。還有就是做存量數據的一次性遷移，增量了批量導入能力，減少了序列化和反序列化的開銷。

另外現在有一些用戶會覺得Hudi存一些元數據欄位，比如_hoodie_commit_time等元資訊，這些資訊都是從數據資訊里提取的，有部分存儲開銷，現在支援虛擬鍵，元數據欄位不會再存數據了，它帶來的限制就是不能使用增量ETL，無法獲取Hudi某一個時間點之後的變更數據。

另外很多小夥伴也在希望Hudi支援ORC格式，Hudi最新版本支援了ORC格式，同時這部分格式的是可插拔的，後續可以很靈活接入更多的格式。還做了Metadata Table的寫入和查詢優化，通過Spark SQL查詢的時候，避免Filelist，直接通過Metadata Table獲取整個文件列表資訊。

從更遠來看社區未來的規劃包括對於Spark集成升級到Data SourceV2，現在Hudi基於V1，無法用到V2的性能優化。還有Catalog集成，可以通過Catalog管理表，可以創建、刪除、更新，表格元數據的管理通過Spark Catalog集成。

Flink模組Blink團隊有專職同學負責，後續會把流式數據里的Watremark推到Hudi表裡。

另外是與Kafka Connect Sink的集成，後續直接通過Java客戶把Kafka的數據寫到Hudi，而不用拉起一個Spark/Flink集群作業。

在內核側的優化，包括了基於Metadata Table全局記錄級別索引。還有字節跳動小夥伴做的寫入支援Bucket，這樣的好處就是做數據更新的時候，可以通過主鍵找到對應Bucket，只要把對應Bucket的parquet文件的Bloomfilter讀取出來就可以了，減少了查找更新時候的開銷。

還有更智慧地Clustering策略，在我們內部也做了這部分工作，更智慧的Clustering可以基於之前的負載情況，動態的開啟Clustering優化，另外還包括基於Metadata Table構建二級索引，以及Full Schema Evolution和跨表事務。

現在Hudi社區發展得比較快，程式碼重構量非常大，但都是為了更好的社區發展，從0.7.0到0.9.0版本Flink集成Hudi模組基本上完全重構了，如果有興趣的同學可以參與到社區，共同建設更好的數據湖平台。