Apache Hudi使用簡介

Apache Hudi使用簡介

• Apache Hudi使用簡介
• 資料實時處理和實時的資料
• 業務場景和技術選型
• Apache hudi簡介
  • 使用Aapche Hudi整體思路
  • Hudi表資料結構
    • 資料檔案
    • .hoodie檔案
  • Hudi記錄Id
  • COW和MOR
    • Copy On Write Table
    • Merge On Read Table
• 基于hudi的代碼實作
  • binlog資料寫入Hudi表
  • 歷史資料同步以及表元資料同步至hive
    • 同步歷史資料至hudi表
    • 同步hudi表結構至hive meta
• 一些踩坑
  • hive相關配置
  • spark streaming的一些調優
• 未來改進
• 參考資料

資料實時處理和實時的資料

實時分為處理的實時和資料的實時
即席分析是要求對資料實時的處理，馬上要得到對應的結果
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，它提供了一系列引數，配置程式的執行行為

一個使用的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提供了如下一些引數，用于配置程式執行行為

一個程式執行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，考慮到目前程式只支持按天磁區，所以同步工具可以一天執行一次即可，引數配置如下

一個程式執行demo

java -jar hudi-learn-1.0-SNAPSHOT.jar --hive-db-name streaming --hive-table-name crm__order --hive-user-name hive --hive-pwd hive --hive-jdbc-url jdbc:hive2://192.168.16.181:10000 --hudi-table-path hdfs://192.168.16.181:8020/hudi_table/crm__order --hive-site-path /lib/hive/conf/hive-site.xml

一些踩坑

hive相關配置

有些hive集群的hive.input.format配置，默認是org.apache.hadoop.hive.ql.io.CombineHiveInputFormat，這會導致掛載Hudi資料的Hive外表讀取到所有Hudi的Parquet資料，從而導致最終的讀取結果重復，需要將hive的format改為org.apache.hadoop.hive.ql.io.HiveInputFormat，為了避免在整個集群層面上更改對其余離線Hive Sql造成不必要的影響，建議只對當前hive session設定set hive.input.format=org.apache.hadoop.hive.ql.io.HiveInputFormat;

spark streaming的一些調優

由于binlog寫入Hudi表的是基于Spark streaming實作的，這里給出了一些spark 和spark streaming層面的配置，它能使整個程式作業更穩定

未來改進

• 支持無磁區，或非日期磁區表，目前只支持日期磁區表
• 多資料型別支持，目前為了程式的穩定性，會將Mysql中的欄位全部以String型別存盤至Hudi

參考資料

https://hudi.apache.org/

歡迎關注我的個人公眾號"西北偏北UP"，記錄代碼人生，行業思考，科技評論

標籤：其他

上一篇：redis-復制

下一篇：sqlserver 腳本和批處理指令小結