簡(jiǎn)介：?本文介紹了 Flink Hudi 通過(guò)流計(jì)算對(duì)原有基于 mini-batch 的增量計(jì)算模型的不斷優(yōu)化演進(jìn)。

本文介紹了 Flink Hudi 通過(guò)流計(jì)算對(duì)原有基于 mini-batch 的增量計(jì)算模型不斷優(yōu)化演進(jìn)。用戶可以通過(guò) Flink SQL 將 CDC 數(shù)據(jù)實(shí)時(shí)寫(xiě)入 Hudi 存儲(chǔ)，且在即將發(fā)布的 0.9 版本 Hudi 原生支持 CDC format。主要內(nèi)容為：

背景

增量 ETL

演示

一、背景

近實(shí)時(shí)

從 2016 年開(kāi)始，Apache Hudi 社區(qū)就開(kāi)始通過(guò) Hudi 的 UPSERT 能力探索近實(shí)時(shí)場(chǎng)景的使用案例 [1]。通過(guò) MR/Spark 的批處理模型，用戶可以實(shí)現(xiàn)小時(shí)級(jí)別的數(shù)據(jù)注入 HDFS/OSS。在純實(shí)時(shí)場(chǎng)景，用戶通過(guò)流計(jì)算引擎 Flink + KV/OLAP 存儲(chǔ)的架構(gòu)可以實(shí)現(xiàn)端到端的秒級(jí) (5分鐘級(jí)) 實(shí)時(shí)分析。然而在秒級(jí) (5分鐘級(jí)) 到小時(shí)級(jí)時(shí)的場(chǎng)景還存在大量的用例，我們稱之為 NEAR-REAL-TIME (近實(shí)時(shí))。

在實(shí)踐中有大量的案例都屬于近實(shí)時(shí)的范疇：

分鐘級(jí)別的大屏；

各種 BI 分析 (OLAP)；

機(jī)器學(xué)習(xí)分鐘級(jí)別的特征提取。

增量計(jì)算

解決近實(shí)時(shí)的方案當(dāng)前是比較開(kāi)放的。

• 流處理的時(shí)延低，但是 SQL 的 pattern 比較固定，查詢端的能力（索引、ad hoc）欠缺；
• 批處理的數(shù)倉(cāng)能力豐富但是數(shù)據(jù)時(shí)延大。

于是 Hudi 社區(qū)提出基于 mini-batch 的增量計(jì)算模型：

增量數(shù)據(jù)集 =>?增量計(jì)算結(jié)果?merge?已存結(jié)果 => 外存

這套模型通過(guò)湖存儲(chǔ)的 snapshot 拉取增量的數(shù)據(jù)集 (兩個(gè) commits 之前的數(shù)據(jù)集)，通過(guò) Spark/Hive 等批處理框架計(jì)算增量的結(jié)果 (比如簡(jiǎn)單的 count) 再 merge 到已存結(jié)果中。

核心問(wèn)題

增量模型需要解決的核心問(wèn)題：

UPSERT 能力：類似 KUDU 和 Hive ACID，Hudi 也提供了分鐘級(jí)的更新能力；

增量消費(fèi)：Hudi 通過(guò)湖存儲(chǔ)的多 snapshots 提供增量拉取。

基于 mini-batch 的增量計(jì)算模型可以提升部分場(chǎng)景的時(shí)延、節(jié)省計(jì)算成本，但有一個(gè)很大的限制：對(duì) SQL 的 pattern 有要求。因?yàn)橛?jì)算走的是批，批計(jì)算本身不維護(hù)狀態(tài)，這就要求計(jì)算的指標(biāo)能夠比較方便地 merge，簡(jiǎn)單的 count、sum 可以做，但是 avg、count distinct 這些還是需要拉取全量數(shù)據(jù)重算。

隨著流計(jì)算和實(shí)時(shí)數(shù)倉(cāng)的普及，Hudi 社區(qū)也在積極的擁抱變化，通過(guò)流計(jì)算對(duì)原有基于 mini-batch 的增量計(jì)算模型不斷優(yōu)化演進(jìn)：在 0.7 版本引入了流式數(shù)據(jù)入湖，在 0.9 版本支持了原生的 CDC format。

二、增量 ETL

DB 數(shù)據(jù)入湖

隨著 CDC 技術(shù)的成熟，debezium 這樣的 CDC 工具越來(lái)越流行，Hudi 社區(qū)也先后集成了流寫(xiě)，流讀的能力。用戶可以通過(guò) Flink SQL 將 CDC 數(shù)據(jù)實(shí)時(shí)寫(xiě)入 Hudi 存儲(chǔ)：

• 用戶既可以通過(guò) Flink CDC connector 直接將 DB 數(shù)據(jù)導(dǎo)入 Hudi；
• 也可以先將 CDC 數(shù)據(jù)導(dǎo)入 Kafka，再通過(guò) Kafka connector 導(dǎo)入 Hudi。

第二種方案的容錯(cuò)和擴(kuò)展性會(huì)好一些。

數(shù)據(jù)湖 CDC

在即將發(fā)布的 0.9 版本，Hudi 原生支持 CDC format，一條 record 的所有變更記錄都可以保存，基于此，Hudi 和流計(jì)算系統(tǒng)結(jié)合的更加完善，可以流式讀取 CDC 數(shù)據(jù) [2]：

源頭 CDC 流的所有消息變更都在入湖之后保存下來(lái)，被用于流式消費(fèi)。Flink 的有狀態(tài)計(jì)算實(shí)時(shí)累加計(jì)算結(jié)果 (state)，通過(guò)流式寫(xiě) Hudi 將計(jì)算的變更同步到 Hudi 湖存儲(chǔ)，之后繼續(xù)對(duì)接 Flink 流式消費(fèi) Hudi 存儲(chǔ)的 changelog， 實(shí)現(xiàn)下一層級(jí)的有狀態(tài)計(jì)算。近實(shí)時(shí)端到端 ETL pipeline：

這套架構(gòu)將端到端的 ETL 時(shí)延縮短到分鐘級(jí)，并且每一層的存儲(chǔ)格式都可以通過(guò) compaction 壓縮成列存（Parquet、ORC）以提供 OLAP 分析能力，由于數(shù)據(jù)湖的開(kāi)放性，壓縮后的格式可以對(duì)接各種查詢引擎：Flink、Spark、Presto、Hive 等。

一張 Hudi 數(shù)據(jù)湖表具備兩種形態(tài)：

• 表形態(tài)：查詢最新的快照結(jié)果，同時(shí)提供高效的列存格式
• 流形態(tài)：流式消費(fèi)變更，可以指定任意點(diǎn)位流讀之后的 changelog

三、演示

我們通過(guò)一段 Demo 演示 Hudi 表的兩種形態(tài)。

環(huán)境準(zhǔn)備

• Flink SQL Client
• Hudi master 打包?hudi-flink-bundle?jar
• Flink 1.13.1

這里提前準(zhǔn)備一段 debezium-json 格式的 CDC 數(shù)據(jù)

{"before":null,"after":{"id":101,"ts":1000,"name":"scooter","description":"Small 2-wheel scooter","weight":3.140000104904175},"source":{"version":"1.1.1.Final","connector":"mysql","name":"dbserver1","ts_ms":0,"snapshot":"true","db":"inventory","table":"products","server_id":0,"gtid":null,"file":"mysql-bin.000003","pos":154,"row":0,"thread":null,"query":null},"op":"c","ts_ms":1589355606100,"transaction":null} {"before":null,"after":{"id":102,"ts":2000,"name":"car battery","description":"12V car battery","weight":8.100000381469727},"source":{"version":"1.1.1.Final","connector":"mysql","name":"dbserver1","ts_ms":0,"snapshot":"true","db":"inventory","table":"products","server_id":0,"gtid":null,"file":"mysql-bin.000003","pos":154,"row":0,"thread":null,"query":null},"op":"c","ts_ms":1589355606101,"transaction":null} {"before":null,"after":{"id":103,"ts":3000,"name":"12-pack drill bits","description":"12-pack of drill bits with sizes ranging from #40 to #3","weight":0.800000011920929},"source":{"version":"1.1.1.Final","connector":"mysql","name":"dbserver1","ts_ms":0,"snapshot":"true","db":"inventory","table":"products","server_id":0,"gtid":null,"file":"mysql-bin.000003","pos":154,"row":0,"thread":null,"query":null},"op":"c","ts_ms":1589355606101,"transaction":null} {"before":null,"after":{"id":104,"ts":4000,"name":"hammer","description":"12oz carpenter's hammer","weight":0.75},"source":{"version":"1.1.1.Final","connector":"mysql","name":"dbserver1","ts_ms":0,"snapshot":"true","db":"inventory","table":"products","server_id":0,"gtid":null,"file":"mysql-bin.000003","pos":154,"row":0,"thread":null,"query":null},"op":"c","ts_ms":1589355606101,"transaction":null} {"before":null,"after":{"id":105,"ts":5000,"name":"hammer","description":"14oz carpenter's hammer","weight":0.875},"source":{"version":"1.1.1.Final","connector":"mysql","name":"dbserver1","ts_ms":0,"snapshot":"true","db":"inventory","table":"products","server_id":0,"gtid":null,"file":"mysql-bin.000003","pos":154,"row":0,"thread":null,"query":null},"op":"c","ts_ms":1589355606101,"transaction":null} {"before":null,"after":{"id":106,"ts":6000,"name":"hammer","description":"16oz carpenter's hammer","weight":1},"source":{"version":"1.1.1.Final","connector":"mysql","name":"dbserver1","ts_ms":0,"snapshot":"true","db":"inventory","table":"products","server_id":0,"gtid":null,"file":"mysql-bin.000003","pos":154,"row":0,"thread":null,"query":null},"op":"c","ts_ms":1589355606101,"transaction":null} {"before":null,"after":{"id":107,"ts":7000,"name":"rocks","description":"box of assorted rocks","weight":5.300000190734863},"source":{"version":"1.1.1.Final","connector":"mysql","name":"dbserver1","ts_ms":0,"snapshot":"true","db":"inventory","table":"products","server_id":0,"gtid":null,"file":"mysql-bin.000003","pos":154,"row":0,"thread":null,"query":null},"op":"c","ts_ms":1589355606101,"transaction":null} {"before":null,"after":{"id":108,"ts":8000,"name":"jacket","description":"water resistent black wind breaker","weight":0.10000000149011612},"source":{"version":"1.1.1.Final","connector":"mysql","name":"dbserver1","ts_ms":0,"snapshot":"true","db":"inventory","table":"products","server_id":0,"gtid":null,"file":"mysql-bin.000003","pos":154,"row":0,"thread":null,"query":null},"op":"c","ts_ms":1589355606101,"transaction":null} {"before":null,"after":{"id":109,"ts":9000,"name":"spare tire","description":"24 inch spare tire","weight":22.200000762939453},"source":{"version":"1.1.1.Final","connector":"mysql","name":"dbserver1","ts_ms":0,"snapshot":"true","db":"inventory","table":"products","server_id":0,"gtid":null,"file":"mysql-bin.000003","pos":154,"row":0,"thread":null,"query":null},"op":"c","ts_ms":1589355606101,"transaction":null} {"before":{"id":106,"ts":6000,"name":"hammer","description":"16oz carpenter's hammer","weight":1},"after":{"id":106,"ts":10000,"name":"hammer","description":"18oz carpenter hammer","weight":1},"source":{"version":"1.1.1.Final","connector":"mysql","name":"dbserver1","ts_ms":1589361987000,"snapshot":"false","db":"inventory","table":"products","server_id":223344,"gtid":null,"file":"mysql-bin.000003","pos":362,"row":0,"thread":2,"query":null},"op":"u","ts_ms":1589361987936,"transaction":null} {"before":{"id":107,"ts":7000,"name":"rocks","description":"box of assorted rocks","weight":5.300000190734863},"after":{"id":107,"ts":11000,"name":"rocks","description":"box of assorted rocks","weight":5.099999904632568},"source":{"version":"1.1.1.Final","connector":"mysql","name":"dbserver1","ts_ms":1589362099000,"snapshot":"false","db":"inventory","table":"products","server_id":223344,"gtid":null,"file":"mysql-bin.000003","pos":717,"row":0,"thread":2,"query":null},"op":"u","ts_ms":1589362099505,"transaction":null} {"before":null,"after":{"id":110,"ts":12000,"name":"jacket","description":"water resistent white wind breaker","weight":0.20000000298023224},"source":{"version":"1.1.1.Final","connector":"mysql","name":"dbserver1","ts_ms":1589362210000,"snapshot":"false","db":"inventory","table":"products","server_id":223344,"gtid":null,"file":"mysql-bin.000003","pos":1068,"row":0,"thread":2,"query":null},"op":"c","ts_ms":1589362210230,"transaction":null} {"before":null,"after":{"id":111,"ts":13000,"name":"scooter","description":"Big 2-wheel scooter ","weight":5.179999828338623},"source":{"version":"1.1.1.Final","connector":"mysql","name":"dbserver1","ts_ms":1589362243000,"snapshot":"false","db":"inventory","table":"products","server_id":223344,"gtid":null,"file":"mysql-bin.000003","pos":1394,"row":0,"thread":2,"query":null},"op":"c","ts_ms":1589362243428,"transaction":null} {"before":{"id":110,"ts":12000,"name":"jacket","description":"water resistent white wind breaker","weight":0.20000000298023224},"after":{"id":110,"ts":14000,"name":"jacket","description":"new water resistent white wind breaker","weight":0.5},"source":{"version":"1.1.1.Final","connector":"mysql","name":"dbserver1","ts_ms":1589362293000,"snapshot":"false","db":"inventory","table":"products","server_id":223344,"gtid":null,"file":"mysql-bin.000003","pos":1707,"row":0,"thread":2,"query":null},"op":"u","ts_ms":1589362293539,"transaction":null} {"before":{"id":111,"ts":13000,"name":"scooter","description":"Big 2-wheel scooter ","weight":5.179999828338623},"after":{"id":111,"ts":15000,"name":"scooter","description":"Big 2-wheel scooter ","weight":5.170000076293945},"source":{"version":"1.1.1.Final","connector":"mysql","name":"dbserver1","ts_ms":1589362330000,"snapshot":"false","db":"inventory","table":"products","server_id":223344,"gtid":null,"file":"mysql-bin.000003","pos":2090,"row":0,"thread":2,"query":null},"op":"u","ts_ms":1589362330904,"transaction":null} {"before":{"id":111,"ts":16000,"name":"scooter","description":"Big 2-wheel scooter ","weight":5.170000076293945},"after":null,"source":{"version":"1.1.1.Final","connector":"mysql","name":"dbserver1","ts_ms":1589362344000,"snapshot":"false","db":"inventory","table":"products","server_id":223344,"gtid":null,"file":"mysql-bin.000003","pos":2443,"row":0,"thread":2,"query":null},"op":"d","ts_ms":1589362344455,"transaction":null}

通過(guò) Flink SQL Client 創(chuàng)建表用來(lái)讀取 CDC 數(shù)據(jù)文件

Flink SQL> CREATE TABLE debezium_source( > id INT NOT NULL, > ts BIGINT, > name STRING, > description STRING, > weight DOUBLE > ) WITH ( > 'connector' = 'filesystem', > 'path' = '/Users/chenyuzhao/workspace/hudi-demo/source.data', > 'format' = 'debezium-json' > ); [INFO] Execute statement succeed.

執(zhí)行 SELECT 觀察結(jié)果，可以看到一共有 20 條記錄，中間有一些 UPDATE s，最后一條消息是 DELETE

Flink SQL> select * from debezium_source; +----+-------------+----------------------+--------------------------------+--------------------------------+--------------------------------+ | op | id | ts | name | description | weight | +----+-------------+----------------------+--------------------------------+--------------------------------+--------------------------------+ | +I | 101 | 1000 | scooter | Small 2-wheel scooter | 3.140000104904175 | | +I | 102 | 2000 | car battery | 12V car battery | 8.100000381469727 | | +I | 103 | 3000 | 12-pack drill bits | 12-pack of drill bits with ... | 0.800000011920929 | | +I | 104 | 4000 | hammer | 12oz carpenter's hammer | 0.75 | | +I | 105 | 5000 | hammer | 14oz carpenter's hammer | 0.875 | | +I | 106 | 6000 | hammer | 16oz carpenter's hammer | 1.0 | | +I | 107 | 7000 | rocks | box of assorted rocks | 5.300000190734863 | | +I | 108 | 8000 | jacket | water resistent black wind ... | 0.10000000149011612 | | +I | 109 | 9000 | spare tire | 24 inch spare tire | 22.200000762939453 | | -U | 106 | 6000 | hammer | 16oz carpenter's hammer | 1.0 | | +U | 106 | 10000 | hammer | 18oz carpenter hammer | 1.0 | | -U | 107 | 7000 | rocks | box of assorted rocks | 5.300000190734863 | | +U | 107 | 11000 | rocks | box of assorted rocks | 5.099999904632568 | | +I | 110 | 12000 | jacket | water resistent white wind ... | 0.20000000298023224 | | +I | 111 | 13000 | scooter | Big 2-wheel scooter | 5.179999828338623 | | -U | 110 | 12000 | jacket | water resistent white wind ... | 0.20000000298023224 | | +U | 110 | 14000 | jacket | new water resistent white w... | 0.5 | | -U | 111 | 13000 | scooter | Big 2-wheel scooter | 5.179999828338623 | | +U | 111 | 15000 | scooter | Big 2-wheel scooter | 5.170000076293945 | | -D | 111 | 16000 | scooter | Big 2-wheel scooter | 5.170000076293945 | +----+-------------+----------------------+--------------------------------+--------------------------------+--------------------------------+ Received a total of 20 rows

創(chuàng)建 Hudi 表，這里設(shè)置表的形態(tài)為?MERGE_ON_READ?并且打開(kāi) changelog 模式屬性?changelog.enabled

Flink SQL> CREATE TABLE hoodie_table( > id INT NOT NULL PRIMARY KEY NOT ENFORCED, > ts BIGINT, > name STRING, > description STRING, > weight DOUBLE > ) WITH ( > 'connector' = 'hudi', > 'path' = '/Users/chenyuzhao/workspace/hudi-demo/t1', > 'table.type' = 'MERGE_ON_READ', > 'changelog.enabled' = 'true', > 'compaction.async.enabled' = 'false' > ); [INFO] Execute statement succeed.

查詢

通過(guò) INSERT 語(yǔ)句將數(shù)據(jù)導(dǎo)入 Hudi，開(kāi)啟流讀模式，并執(zhí)行查詢觀察結(jié)果

Flink SQL> select * from hoodie_table/*+ OPTIONS('read.streaming.enabled'='true')*/; +----+-------------+----------------------+--------------------------------+--------------------------------+--------------------------------+ | op | id | ts | name | description | weight | +----+-------------+----------------------+--------------------------------+--------------------------------+--------------------------------+ | +I | 101 | 1000 | scooter | Small 2-wheel scooter | 3.140000104904175 | | +I | 102 | 2000 | car battery | 12V car battery | 8.100000381469727 | | +I | 103 | 3000 | 12-pack drill bits | 12-pack of drill bits with ... | 0.800000011920929 | | +I | 104 | 4000 | hammer | 12oz carpenter's hammer | 0.75 | | +I | 105 | 5000 | hammer | 14oz carpenter's hammer | 0.875 | | +I | 106 | 6000 | hammer | 16oz carpenter's hammer | 1.0 | | +I | 107 | 7000 | rocks | box of assorted rocks | 5.300000190734863 | | +I | 108 | 8000 | jacket | water resistent black wind ... | 0.10000000149011612 | | +I | 109 | 9000 | spare tire | 24 inch spare tire | 22.200000762939453 | | -U | 106 | 6000 | hammer | 16oz carpenter's hammer | 1.0 | | +U | 106 | 10000 | hammer | 18oz carpenter hammer | 1.0 | | -U | 107 | 7000 | rocks | box of assorted rocks | 5.300000190734863 | | +U | 107 | 11000 | rocks | box of assorted rocks | 5.099999904632568 | | +I | 110 | 12000 | jacket | water resistent white wind ... | 0.20000000298023224 | | +I | 111 | 13000 | scooter | Big 2-wheel scooter | 5.179999828338623 | | -U | 110 | 12000 | jacket | water resistent white wind ... | 0.20000000298023224 | | +U | 110 | 14000 | jacket | new water resistent white w... | 0.5 | | -U | 111 | 13000 | scooter | Big 2-wheel scooter | 5.179999828338623 | | +U | 111 | 15000 | scooter | Big 2-wheel scooter | 5.170000076293945 | | -D | 111 | 16000 | scooter | Big 2-wheel scooter | 5.170000076293945 |

可以看到 Hudi 保留了每行的變更記錄，包括 change log 的 operation 類型，這里我們打開(kāi) TABLE HINTS 功能，方便動(dòng)態(tài)設(shè)置表參數(shù)。

繼續(xù)使用 batch 讀模式，執(zhí)行查詢觀察輸出結(jié)果，可以看到中間的變更被合并。

Flink SQL> select * from hoodie_table; 2021-08-20 20:51:25,052 INFO org.apache.hadoop.conf.Configuration.deprecation [] - mapred.job.map.memory.mb is deprecated. Instead, use mapreduce.map.memory.mb +----+-------------+----------------------+--------------------------------+--------------------------------+--------------------------------+ | op | id | ts | name | description | weight | +----+-------------+----------------------+--------------------------------+--------------------------------+--------------------------------+ | +U | 110 | 14000 | jacket | new water resistent white w... | 0.5 | | +I | 101 | 1000 | scooter | Small 2-wheel scooter | 3.140000104904175 | | +I | 102 | 2000 | car battery | 12V car battery | 8.100000381469727 | | +I | 103 | 3000 | 12-pack drill bits | 12-pack of drill bits with ... | 0.800000011920929 | | +I | 104 | 4000 | hammer | 12oz carpenter's hammer | 0.75 | | +I | 105 | 5000 | hammer | 14oz carpenter's hammer | 0.875 | | +U | 106 | 10000 | hammer | 18oz carpenter hammer | 1.0 | | +U | 107 | 11000 | rocks | box of assorted rocks | 5.099999904632568 | | +I | 108 | 8000 | jacket | water resistent black wind ... | 0.10000000149011612 | | +I | 109 | 9000 | spare tire | 24 inch spare tire | 22.200000762939453 | +----+-------------+----------------------+--------------------------------+--------------------------------+--------------------------------+ Received a total of 10 rows

聚合

Bounded Source 讀模式下計(jì)算?count(*)

Flink SQL> select count (*) from hoodie_table; +----+----------------------+ | op | EXPR$0 | +----+----------------------+ | +I | 1 | | -U | 1 | | +U | 2 | | -U | 2 | | +U | 3 | | -U | 3 | | +U | 4 | | -U | 4 | | +U | 5 | | -U | 5 | | +U | 6 | | -U | 6 | | +U | 7 | | -U | 7 | | +U | 8 | | -U | 8 | | +U | 9 | | -U | 9 | | +U | 10 | +----+----------------------+ Received a total of 19 rows

Streaming 讀模式下計(jì)算?count(*)

Flink SQL> select count (*) from hoodie_table/*+OPTIONS('read.streaming.enabled'='true')*/; +----+----------------------+ | op | EXPR$0 | +----+----------------------+ | +I | 1 | | -U | 1 | | +U | 2 | | -U | 2 | | +U | 3 | | -U | 3 | | +U | 4 | | -U | 4 | | +U | 5 | | -U | 5 | | +U | 6 | | -U | 6 | | +U | 7 | | -U | 7 | | +U | 8 | | -U | 8 | | +U | 9 | | -U | 9 | | +U | 8 | | -U | 8 | | +U | 9 | | -U | 9 | | +U | 8 | | -U | 8 | | +U | 9 | | -U | 9 | | +U | 10 | | -U | 10 | | +U | 11 | | -U | 11 | | +U | 10 | | -U | 10 | | +U | 11 | | -U | 11 | | +U | 10 | | -U | 10 | | +U | 11 | | -U | 11 | | +U | 10 |

可以看到 batch 和 streaming 模式下的計(jì)算結(jié)果是一致的。

原文鏈接

本文為阿里云原創(chuàng)內(nèi)容，未經(jīng)允許不得轉(zhuǎn)載。?

總結(jié)

以上是生活随笔為你收集整理的使用 Flink Hudi 构建流式数据湖的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問(wèn)題。

如果覺(jué)得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。