簡介： 阿裡雲技術專家李少鋒(風澤)在Apache Hudi 與 Apache Pulsar 聯合 Meetup 杭州站上的演講整理稿件，本議題将介紹典型 CDC 入湖場景，以及如何使用 Pulsar/Hudi 來建構資料湖，同時将會分享 Hudi 核心設計、新願景以及社群最新動态。
本文PPT下載下傳連結：
李少鋒(風澤) - 阿裡雲技術專家-《基于Apache Hudi的CDC資料入湖》.pdf
其他幹貨：
王烨(萌豆)-阿裡雲進階技術專家 -《阿裡雲基于Hudi建構Lakehouse實踐》.pdf
翟佳-StreamNative 聯合創始人、Apache Pulsar PMC 成員-《Pulsar 2.8.0 功能特性概述及規劃》.pdf
盛宇帆-StreamNative 軟體工程師-《基于 Flink 的全新 Pulsar Connector 的設計、開發和使用》.pdf
一、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做到分鐘級。
二、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異常情況下也不會出現資料重複問題。
三、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時查詢待更新檔案的開銷，也是上雲必備。
四、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。
在核心側的優化，包括了基于Metadata Table全局記錄級别索引。還有位元組跳動小夥伴做的寫入支援Bucket，這樣的好處就是做資料更新的時候，可以通過主鍵找到對應Bucket，隻要把對應Bucket的parquet檔案的Bloomfilter讀取出來就可以了，減少了查找更新時候的開銷。
還有更智能地Clustering政策，在我們内部也做了這部分工作，更智能的Clustering可以基于之前的負載情況，動态的開啟Clustering優化，另外還包括基于Metadata Table建構二級索引，以及Full Schema Evolution和跨表事務。
現在Hudi社群發展得比較快，代碼重構量非常大，但都是為了更好的社群發展，從0.7.0到0.9.0版本Flink內建Hudi子產品基本上完全重構了，如果有興趣的同學可以參與到社群，共同建設更好的資料湖平台。
原文連結：https://developer.aliyun.com/article/789432?
版權聲明： 本文内容由阿裡雲實名注冊使用者自發貢獻，版權歸原作者所有，阿裡雲開發者社群不擁有其著作權，亦不承擔相應法律責任。具體規則請檢視《阿裡雲開發者社群使用者服務協定》和《阿裡雲開發者社群知識産權保護指引》。如果您發現本社群中有涉嫌抄襲的内容，填寫侵權投訴表單進行舉報，一經查實，本社群将立刻删除涉嫌侵權内容。