使用 Flink Hudi 構(gòu)建流式數(shù)據(jù)湖

作者：佚名 2021-08-31 10:07:16

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

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

1.背景

2.增量 ETL

3.演示

一、背景

近實(shí)時(shí)

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

分鐘級(jí)別的大屏；
各種 BI 分析 (OLAP)；
機(jī)器學(xué)習(xí)分鐘級(jí)別的特征提取。

增量計(jì)算

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

流處理的時(shí)延低，但是 SQL 的 pattern 比較固定，查詢端的能力（索引、ad hoc）欠缺；
批處理的數(shù)倉能力豐富但是數(shù)據(jù)時(shí)延大。
于是 Hudi 社區(qū)提出基于 mini-batch 的增量計(jì)算模型：

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

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

核心問題

增量模型需要解決的核心問題：

UPSERT 能力：類似 KUDU 和 Hive ACID，Hudi 也提供了分鐘級(jí)的更新能力；
增量消費(fèi)：Hudi 通過湖存儲(chǔ)的多 snapshots 提供增量拉取。
基于 mini-batch 的增量計(jì)算模型可以提升部分場景的時(shí)延、節(jié)省計(jì)算成本，但有一個(gè)很大的限制：對(duì) SQL 的 pattern 有要求。因?yàn)橛?jì)算走的是批，批計(jì)算本身不維護(hù)狀態(tài)，這就要求計(jì)算的指標(biāo)能夠比較方便地 merge，簡單的 count、sum 可以做，但是 avg、count distinct 這些還是需要拉取全量數(shù)據(jù)重算。

隨著流計(jì)算和實(shí)時(shí)數(shù)倉的普及，Hudi 社區(qū)也在積極的擁抱變化，通過流計(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 工具越來越流行，Hudi 社區(qū)也先后集成了流寫，流讀的能力。用戶可以通過 Flink SQL 將 CDC 數(shù)據(jù)實(shí)時(shí)寫入 Hudi 存儲(chǔ)：

用戶既可以通過 Flink CDC connector 直接將 DB 數(shù)據(jù)導(dǎo)入 Hudi；
也可以先將 CDC 數(shù)據(jù)導(dǎo)入 Kafka，再通過 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 流的所有消息變更都在入湖之后保存下來，被用于流式消費(fèi)。Flink 的有狀態(tài)計(jì)算實(shí)時(shí)累加計(jì)算結(jié)果 (state)，通過流式寫 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ǔ)格式都可以通過 compaction 壓縮成列存（Parquet、ORC）以提供 OLAP 分析能力，由于數(shù)據(jù)湖的開放性，壓縮后的格式可以對(duì)接各種查詢引擎：Flink、Spark、Presto、Hive 等。

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

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

三、演示

我們通過一段 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

通過 Flink SQL Client 創(chuàng)建表用來讀取 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 并且打開 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.