Apache Hudi使用簡介
Apache Hudi使用簡介
目錄資料實時處理和實時的資料
實時分為處理的實時和資料的實時
即席分析是要求對資料實時的處理,馬上要得到對應的結果
Flink、Spark Streaming是用來對實時資料的實時處理,資料要求實時,處理也要迅速
資料不實時,處理也不及時的場景則是我們的數倉T+1資料
而本文探討的Apache Hudi,對應的場景是資料的實時,而非處理的實時。它旨在將Mysql中的時候以近實時的方式對映到大資料平臺,比如Hive中。
業務場景和技術選型
傳統的離線數倉,通常資料是T+1的,不能滿足對當日資料分析的需求
而流式計算一般是基於視窗,並且視窗邏輯相對比較固定。
而筆者所在的公司有一類特殊的需求,業務分析比較熟悉現有事務資料庫的資料結構,並且希望有很多即席分析,這些分析包含當日比較實時的資料。慣常他們是基於Mysql從庫,直接通過Sql做相應的分析計算。但很多時候會遇到如下障礙
- 資料量較大、分析邏輯較為複雜時,Mysql從庫耗時較長
- 一些跨庫的分析無法實現
因此,一些彌合在OLTP和OLAP之間的技術框架出現,典型有TiDB。它能同時支援OLTP和OLAP。而諸如Apache Hudi和Apache Kudu則相當於現有OLTP和OLAP技術的橋樑。他們能夠以現有OLTP中的資料結構儲存資料,支援CRUD,同時提供跟現有OLAP框架的整合(如Hive,Impala),以實現OLAP分析
Apache Kudu,需要單獨部署叢集。而Apache Hudi則不需要,它可以利用現有的大資料叢集比如HDFS做資料檔案儲存,然後通過Hive做資料分析,相對來說更適合資源受限的環境
Apache hudi簡介
使用Aapche Hudi整體思路
Hudi 提供了Hudi 表的概念,這些表支援CRUD操作。我們可以基於這個特點,將Mysql Binlog的資料重放至Hudi表,然後基於Hive對Hudi表進行查詢分析。資料流向架構如下
Hudi表資料結構
Hudi表的資料檔案,可以使用作業系統的檔案系統儲存,也可以使用HDFS這種分散式的檔案系統儲存。為了後續分析效能和資料的可靠性,一般使用HDFS進行儲存。以HDFS儲存來看,一個Hudi表的儲存檔案分為兩類。
- 包含
_partition_key
相關的路徑是實際的資料檔案,按分割槽儲存,當然分割槽的路徑key是可以指定的,我這裡使用的是_partition_key - .hoodie 由於CRUD的零散性,每一次的操作都會生成一個檔案,這些小檔案越來越多後,會嚴重影響HDFS的效能,Hudi設計了一套檔案合併機制。 .hoodie資料夾中存放了對應的檔案合併操作相關的日誌檔案。
資料檔案
Hudi真實的資料檔案使用Parquet檔案格式儲存
.hoodie檔案
Hudi把隨著時間流逝,對錶的一系列CRUD操作叫做Timeline。Timeline中某一次的操作,叫做Instant。Instant包含以下資訊
- Instant Action 記錄本次操作是一次資料提交(COMMITS),還是檔案合併(COMPACTION),或者是檔案清理(CLEANS)
- Instant Time 本次操作發生的時間
- state 操作的狀態,發起(REQUESTED),進行中(INFLIGHT),還是已完成(COMPLETED)
.hoodie資料夾中存放對應操作的狀態記錄
Hudi記錄Id
hudi為了實現資料的CRUD,需要能夠唯一標識一條記錄。hudi將把資料集中的唯一欄位(record key ) + 資料所在分割槽 (partitionPath) 聯合起來當做資料的唯一鍵
COW和MOR
基於上述基礎概念之上,Hudi提供了兩類表格式COW和MOR。他們會在資料的寫入和查詢效能上有一些不同
Copy On Write Table
簡稱COW。顧名思義,他是在資料寫入的時候,複製一份原來的拷貝,在其基礎上新增新資料。正在讀資料的請求,讀取的是是近的完整副本,這類似Mysql 的MVCC的思想。
上圖中,每一個顏色都包含了截至到其所在時間的所有資料。老的資料副本在超過一定的個數限制後,將被刪除。這種型別的表,沒有compact instant,因為寫入時相當於已經compact了。
- 優點 讀取時,只讀取對應分割槽的一個數據檔案即可,較為高效
- 缺點 資料寫入的時候,需要複製一個先前的副本再在其基礎上生成新的資料檔案,這個過程比較耗時。且由於耗時,讀請求讀取到的資料相對就會滯後
Merge On Read Table
簡稱MOR。新插入的資料儲存在delta log 中。定期再將delta log合併進行parquet資料檔案。讀取資料時,會將delta log跟老的資料檔案做merge,得到完整的資料返回。當然,MOR表也可以像COW表一樣,忽略delta log,只讀取最近的完整資料檔案。下圖演示了MOR的兩種資料讀寫方式
- 優點 由於寫入資料先寫delta log,且delta log較小,所以寫入成本較低
- 缺點 需要定期合併整理compact,否則碎片檔案較多。讀取效能較差,因為需要將delta log 和 老資料檔案合併
基於hudi的程式碼實現
我在github上放置了基於Hudi的封裝實現,對應的原始碼地址為 https://github.com/wanqiufeng/hudi-learn。
binlog資料寫入Hudi表
- binlog-consumer分支使用Spark streaming消費kafka中的Binlog資料,並寫入Hudi表。Kafka中的binlog是通過阿里的Canal工具同步拉取的。程式入口是CanalKafkaImport2Hudi,它提供了一系列引數,配置程式的執行行為
引數名 | 含義 | 是否必填 | 預設值 |
---|---|---|---|
--base-save-path |
hudi表存放在HDFS的基礎路徑,比如hdfs://192.168.16.181:8020/hudi_data/ | 是 | 無 |
--mapping-mysql-db-name |
指定處理的Mysql庫名 | 是 | 無 |
--mapping-mysql-table-name |
指定處理的Mysql表名 | 是 | 無 |
--store-table-name |
指定Hudi的表名 | 否 | 預設會根據--mapping-mysql-db-name和--mapping-mysql-table-name自動生成。假設--mapping-mysql-db-name 為crm,--mapping-mysql-table-name為order。那麼最終的hudi表名為crm__order |
--real-save-path |
指定hudi表最終儲存的hdfs路徑 | 否 | 預設根據--base-save-path和--store-table-name自動生成,生成格式為'--base-save-path'+'/'+'--store-table-name' ,推薦預設 |
--primary-key |
指定同步的mysql表中能唯一標識記錄的欄位名 | 否 | 預設id |
--partition-key |
指定mysql表中可以用於分割槽的時間欄位,欄位必須是timestamp 或dateime型別 | 是 | 無 |
--precombine-key |
最終用於配置hudi的hoodie.datasource.write.precombine.field |
否 | 預設id |
--kafka-server |
指定Kafka 叢集地址 | 是 | 無 |
--kafka-topic |
指定消費kafka的佇列 | 是 | 無 |
--kafka-group |
指定消費kafka的group | 否 | 預設在儲存表名前加'hudi'字首,比如'hudi_crm__order' |
--duration-seconds |
由於本程式使用Spark streaming開發,這裡指定Spark streaming微批的時長 | 否 | 預設10秒 |
一個使用的demo如下
/data/opt/spark-2.4.4-bin-hadoop2.6/bin/spark-submit --class com.niceshot.hudi.CanalKafkaImport2Hudi \
--name hudi__goods \
--master yarn \
--deploy-mode cluster \
--driver-memory 512m \
--executor-memory 512m \
--executor-cores 1 \
--num-executors 1 \
--queue hudi \
--conf spark.executor.memoryOverhead=2048 \
--conf "spark.executor.extraJavaOptions=-XX:+HeapDumpOnOutOfMemoryError -XX:HeapDumpPath=\tmp\hudi-debug" \
--conf spark.core.connection.ack.wait.timeout=300 \
--conf spark.locality.wait=100 \
--conf spark.streaming.backpressure.enabled=true \
--conf spark.streaming.receiver.maxRate=500 \
--conf spark.streaming.kafka.maxRatePerPartition=200 \
--conf spark.ui.retainedJobs=10 \
--conf spark.ui.retainedStages=10 \
--conf spark.ui.retainedTasks=10 \
--conf spark.worker.ui.retainedExecutors=10 \
--conf spark.worker.ui.retainedDrivers=10 \
--conf spark.sql.ui.retainedExecutions=10 \
--conf spark.yarn.submit.waitAppCompletion=false \
--conf spark.yarn.maxAppAttempts=4 \
--conf spark.yarn.am.attemptFailuresValidityInterval=1h \
--conf spark.yarn.max.executor.failures=20 \
--conf spark.yarn.executor.failuresValidityInterval=1h \
--conf spark.task.maxFailures=8 \
/data/opt/spark-applications/hudi_canal_consumer/hudi-canal-import-1.0-SNAPSHOT-jar-with-dependencies.jar --kafka-server local:9092 --kafka-topic dt_streaming_canal_xxx --base-save-path hdfs://192.168.2.1:8020/hudi_table/ --mapping-mysql-db-name crm --mapping-mysql-table-name order --primary-key id --partition-key createDate --duration-seconds 1200
歷史資料同步以及表元資料同步至hive
history_import_and_meta_sync
分支提供了將歷史資料同步至hudi表,以及將hudi表資料結構同步至hive meta的操作
同步歷史資料至hudi表
這裡採用的思路是
- 將mysql全量資料通過注入sqoop等工具,匯入到hive表。
- 然後採用分支程式碼中的工具HiveImport2HudiConfig,將資料匯入Hudi表
HiveImport2HudiConfig提供瞭如下一些引數,用於配置程式執行行為
引數名 | 含義 | 是否必填 | 預設值 |
---|---|---|---|
--base-save-path |
hudi表存放在HDFS的基礎路徑,比如hdfs://192.168.16.181:8020/hudi_data/ | 是 | 無 |
--mapping-mysql-db-name |
指定處理的Mysql庫名 | 是 | 無 |
--mapping-mysql-table-name |
指定處理的Mysql表名 | 是 | 無 |
--store-table-name |
指定Hudi的表名 | 否 | 預設會根據--mapping-mysql-db-name和--mapping-mysql-table-name自動生成。假設--mapping-mysql-db-name 為crm,--mapping-mysql-table-name為order。那麼最終的hudi表名為crm__order |
--real-save-path |
指定hudi表最終儲存的hdfs路徑 | 否 | 預設根據--base-save-path和--store-table-name自動生成,生成格式為'--base-save-path'+'/'+'--store-table-name' ,推薦預設 |
--primary-key |
指定同步的hive歷史表中能唯一標識記錄的欄位名 | 否 | 預設id |
--partition-key |
指定hive歷史表中可以用於分割槽的時間欄位,欄位必須是timestamp 或dateime型別 | 是 | 無 |
--precombine-key |
最終用於配置hudi的hoodie.datasource.write.precombine.field |
否 | 預設id |
--sync-hive-db-name |
全量歷史資料所在hive的庫名 | 是 | 無 |
--sync-hive-table-name |
全量歷史資料所在hive的表名 | 是 | 無 |
--hive-base-path |
hive的所有資料檔案存放地址,需要參看具體的hive配置 | 否 | /user/hive/warehouse |
--hive-site-path |
hive-site.xml配置檔案所在的地址 | 是 | 無 |
--tmp-data-path |
程式執行過程中臨時檔案存放路徑。一般預設路徑是/tmp。有可能出現/tmp所在磁碟太小,而導致歷史程式執行失敗的情況。當出現該情況時,可以通過該引數自定義執行路徑 | 否 | 預設作業系統臨時目錄 |
一個程式執行demo
nohup java -jar hudi-learn-1.0-SNAPSHOT.jar --sync-hive-db-name hudi_temp --sync-hive-table-name crm__wx_user_info --base-save-path hdfs://192.168.2.2:8020/hudi_table/ --mapping-mysql-db-name crm --mapping-mysql-table-name "order" --primary-key "id" --partition-key created_date --hive-site-path /etc/lib/hive/conf/hive-site.xml --tmp-data-path /data/tmp > order.log &
同步hudi表結構至hive meta
需要將hudi的資料結構和分割槽,以hive外表的形式同步至Hive meta,才能是Hive感知到hudi資料,並通過sql進行查詢分析。Hudi本身在消費Binlog進行儲存時,可以順帶將相關表元資料資訊同步至hive。但考慮到每條寫入Apache Hudi表的資料,都要讀寫Hive Meta ,對Hive的效能可能影響很大。所以我單獨開發了HiveMetaSyncConfig工具,用於同步hudi表元資料至Hive。考慮到目前程式只支援按天分割槽,所以同步工具可以一天執行一次即可。引數配置如下
| 引數名 | 含義 | 是否必填 |預設值|
| :-------- | --------