【使用 Flink Hudi 构建流式数据湖】本文介绍了 Flink Hudi 通过流计算对原有基于 mini-batch 的增量计算模型不断优化演进。用户可以通过 Flink SQL 将 CDC 数据实时写入 Hudi 存储，且在即将发布的 0.9 版本 Hudi 原生支持 CDC format。主要内容为：

1. 背景
2. 增量 ETL
3. 演示

一、背景 近实时
从 2016 年开始，Apache Hudi 社区就开始通过 Hudi 的 UPSERT 能力探索近实时场景的使用案例 [1]。通过 MR/Spark 的批处理模型，用户可以实现小时级别的数据注入 HDFS/OSS。在纯实时场景，用户通过流计算引擎 Flink + KV/OLAP 存储的架构可以实现端到端的秒级 (5分钟级) 实时分析。然而在秒级 (5分钟级) 到小时级时的场景还存在大量的用例，我们称之为 NEAR-REAL-TIME (近实时)。

文章图片

在实践中有大量的案例都属于近实时的范畴：

1. 分钟级别的大屏；
2. 各种 BI 分析 (OLAP)；
3. 机器学习分钟级别的特征提取。

增量计算
解决近实时的方案当前是比较开放的。

• 流处理的时延低，但是 SQL 的 pattern 比较固定，查询端的能力（索引、ad hoc）欠缺；
• 批处理的数仓能力丰富但是数据时延大。

于是 Hudi 社区提出基于 mini-batch 的增量计算模型：
增量数据集 => 增量计算结果 merge 已存结果 => 外存
这套模型通过湖存储的 snapshot 拉取增量的数据集 (两个 commits 之前的数据集)，通过 Spark/Hive 等批处理框架计算增量的结果 (比如简单的 count) 再 merge 到已存结果中。
核心问题
增量模型需要解决的核心问题：

1. UPSERT 能力：类似 KUDU 和 Hive ACID，Hudi 也提供了分钟级的更新能力；
2. 增量消费：Hudi 通过湖存储的多 snapshots 提供增量拉取。

基于 mini-batch 的增量计算模型可以提升部分场景的时延、节省计算成本，但有一个很大的限制：对 SQL 的 pattern 有要求。因为计算走的是批，批计算本身不维护状态，这就要求计算的指标能够比较方便地 merge，简单的 count、sum 可以做，但是 avg、count distinct 这些还是需要拉取全量数据重算。
随着流计算和实时数仓的普及，Hudi 社区也在积极的拥抱变化，通过流计算对原有基于 mini-batch 的增量计算模型不断优化演进：在 0.7 版本引入了流式数据入湖，在 0.9 版本支持了原生的 CDC format。
二、增量 ETL DB 数据入湖
随着 CDC 技术的成熟，debezium 这样的 CDC 工具越来越流行，Hudi 社区也先后集成了流写，流读的能力。用户可以通过 Flink SQL 将 CDC 数据实时写入 Hudi 存储：

文章图片

• 用户既可以通过 Flink CDC connector 直接将 DB 数据导入 Hudi；
• 也可以先将 CDC 数据导入 Kafka，再通过 Kafka connector 导入 Hudi。

第二种方案的容错和扩展性会好一些。
数据湖 CDC
在即将发布的 0.9 版本，Hudi 原生支持 CDC format，一条 record 的所有变更记录都可以保存，基于此，Hudi 和流计算系统结合的更加完善，可以流式读取 CDC 数据 [2]：

文章图片

源头 CDC 流的所有消息变更都在入湖之后保存下来，被用于流式消费。Flink 的有状态计算实时累加计算结果 (state)，通过流式写 Hudi 将计算的变更同步到 Hudi 湖存储，之后继续对接 Flink 流式消费 Hudi 存储的 changelog， 实现下一层级的有状态计算。近实时端到端 ETL pipeline：

文章图片

这套架构将端到端的 ETL 时延缩短到分钟级，并且每一层的存储格式都可以通过 compaction 压缩成列存（Parquet、ORC）以提供 OLAP 分析能力，由于数据湖的开放性，压缩后的格式可以对接各种查询引擎：Flink、Spark、Presto、Hive 等。
一张 Hudi 数据湖表具备两种形态：

• 表形态：查询最新的快照结果，同时提供高效的列存格式
• 流形态：流式消费变更，可以指定任意点位流读之后的 changelog

三、演示 我们通过一段 Demo 演示 Hudi 表的两种形态。
环境准备

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

这里提前准备一段 debezium-json 格式的 CDC 数据

{"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 创建表用来读取 CDC 数据文件

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.

执行 SELECT 观察结果，可以看到一共有 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

创建 Hudi 表，这里设置表的形态为 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.

查询
通过 INSERT 语句将数据导入 Hudi，开启流读模式，并执行查询观察结果

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 类型，这里我们打开 TABLE HINTS 功能，方便动态设置表参数。
继续使用 batch 读模式，执行查询观察输出结果，可以看到中间的变更被合并。

Flink SQL> select * from hoodie_table; 2021-08-20 20:51:25,052 INFOorg.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

聚合
Batch 读模式下计算 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 读模式下计算 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 模式下的计算结果是一致的。
Reference
[1] https://www.oreilly.com/conte...
[2] https://hudi.apache.org/blog/...

推荐阅读

• 

  海门大桥
• 

  胃癌|胃癌偏爱这4类人，若你在其中，做好这5点，或许能幸免！
• 

  王者荣耀下载最新版安卓，王者荣耀助手安卓版下载
• 

  电商怎么做 粮油电商是什么意思，粮油是啥
• 

  弈星重塑啦！从围棋的对弈体验入手，天才少年的成长的关键词是：以身入局，____ 王者荣耀8月5日每日一题答案
• 

  安卓 强制前台,手机回收模式下安卓应用如何继续在后台运行?
• 

  电脑调哪几个设置可以让电脑变快速度 电脑调哪几个设置可以让电脑变快
• 

  老款5角硬币图片 「新款5元硬币」
• 

  西瓜汁放冰箱可以保存多久
• 

  qqbot源码分析,nginx源码分析
• 

  梦见抱着男孩
• 

  雨隐村曾经有几位首领 火影忍者手游10月28日每日一题答案
• 

  你看过最好看的美剧是什么？
• 

  php长数据发送 php传输数据的方式
• 

  行政处罚的种类包括人身罚行为罚财产罚和申诫罚 行政处罚的种类
• 

  mysql批量查询语句 mysql查询批量替换
• 

  六月黄吃到什么时候
• 

  小森生活游戏介绍 小森生活好玩吗
• 

  鸡丸的做法大全 鸡丸怎么做
• 

  秦时明月世界怎么转职 秦时明月世界转职业

• 由浅入深理解AOP
• 【译】20个更有效地使用谷歌搜索的技巧
• mybatisplus如何在xml的连表查询中使用queryWrapper
• MybatisPlus|MybatisPlus LambdaQueryWrapper使用int默认值的坑及解决
• MybatisPlus使用queryWrapper如何实现复杂查询
• iOS中的Block
• Linux下面如何查看tomcat已经使用多少线程
• 使用composer自动加载类文件
• android|android studio中ndk的使用
• 使用协程爬取网页，计算网页数据大小