摘要：本文详细介绍了 Flink + Hudi 湖仓一体化方案的原型构建。主要内容为：
Hudi
新架构与湖仓一体
最佳实践
Flink on Hudi
Flink CDC 2.0 on Hudi
Tips：FFA 2021 重磅开启，点击「阅读原文」即可报名～
GitHub 地址
欢迎大家给 Flink 点赞送 star~
一、Hudi
Apache Hudi (发音为 “Hoodie”)在 DFS 的数据集上提供以下流原语：
插入更新 (如何改变数据集?)
增量拉取 (如何获取变更的数据?)
Hudi 维护在数据集上执行的所有操作的时间轴 (timeline），以提供数据集的即时视图。Hudi 将数据集组织到与 Hive 表非常相似的基本路径下的目录结构中。数据集分为多个分区，文件夹包含该分区的文件。每个分区均由相对于基本路径的分区路径唯一标识。
分区记录会被分配到多个文件。每个文件都有一个唯一的文件 ID 和生成该文件的提交 (commit)。如果有更新，则多个文件共享相同的文件 ID，但写入时的提交 (commit) 不同。
存储类型 – 处理数据的存储方式
写时复制
纯列式
创建新版本的文件
读时合并
近实时
视图 – 处理数据的读取方式
读取优化视图 - 输入格式仅选择压缩的列式文件
parquet 文件查询性能
500GB 的延迟时间约为 30 分钟
导入现有的 Hive 表
近实时视图
混合、格式化数据
约 1-5 分钟的延迟
提供近实时表
增量视图
数据集的变更
启用增量拉取
Hudi 存储层由三个不同的部分组成：
元数据 – 它以时间轴的形式维护了在数据集上执行的所有操作的元数据，该时间轴允许将数据集的即时视图存储在基本路径的元数据目录下。时间轴上的操作类型包括：
提交 (commit)，一次提交表示将一批记录原子写入数据集中的过程。单调递增的时间戳，提交表示写操作的开始。
清理 (clean)，清理数据集中不再被查询中使用的文件的较旧版本。
压缩 (compaction)，将行式文件转化为列式文件的动作。
索引 - 将传入的记录键快速映射到文件 (如果已存在记录键)。索引实现是可插拔的，Bloom 过滤器 - 由于不依赖任何外部系统，因此它是默认配置，索引和数据始终保持一致。Apache HBase - 对少量 key 更高效。在索引标记过程中可能会节省几秒钟。
数据 - Hudi 以两种不同的存储格式存储数据。实际使用的格式是可插入的，但要求具有以下特征 – 读优化的列存储格式 (ROFormat)，默认值为 Apache Parquet；写优化的基于行的存储格式 (WOFormat)，默认值为 Apache Avro。
Hudi 解决了以下限制：
HDFS 的可伸缩性限制；
需要在 Hadoop 中更快地呈现数据；
没有直接支持对现有数据的更新和删除；
快速的 ETL 和建模；
要检索所有更新的记录，无论这些更新是添加到最近日期分区的新记录还是对旧数据的更新，Hudi 都允许用户使用最后一个检查点时间戳。此过程不用执行扫描整个源表的查询。
HDFS 中的可伸缩性限制；
Hadoop 中数据的快速呈现；
支持对于现有数据的更新和删除；
快速的 ETL 和建模。
以上内容主要引用于：《Apache Hudi 详解》
二、新架构与湖仓一体
通过湖仓一体、流批一体，准实时场景下做到了：数据同源、同计算引擎、同存储、同计算口径。数据的时效性可以到分钟级，能很好的满足业务准实时数仓的需求。下面是架构图：
MySQL 数据通过 Flink CDC 进入到 Kafka。之所以数据先入 Kafka 而不是直接入 Hudi，是为了实现多个实时任务复用 MySQL 过来的数据，避免多个任务通过 Flink CDC 接 MySQL 表以及 Binlog，对 MySQL 库的性能造成影响。
通过 CDC 进入到 Kafka 的数据除了落一份到离线数据仓库的 ODS 层之外，会同时按照实时数据仓库的链路，从 ODS->DWD->DWS->OLAP 数据库，最后供报表等数据服务使用。实时数仓的每一层结果数据会准实时的落一份到离线数仓，通过这种方式做到程序一次开发、指标口径统一，数据统一。
从架构图上，可以看到有一步数据修正 (重跑历史数据) 的动作，之所以有这一步是考虑到：有可能存在由于口径调整或者前一天的实时任务计算结果错误，导致重跑历史数据的情况。
而存储在 Kafka 的数据有失效时间，不会存太久的历史数据，重跑很久的历史数据无法从 Kafka 中获取历史源数据。再者，如果把大量的历史数据再一次推到 Kafka，走实时计算的链路来修正历史数据，可能会影响当天的实时作业。所以针对重跑历史数据，会通过数据修正这一步来处理。
总体上说，这个架构属于 Lambda 和 Kappa 混搭的架构。流批一体数据仓库的各个数据链路有数据质量校验的流程。第二天对前一天的数据进行对账，如果前一天实时计算的数据无异常，则不需要修正数据，Kappa 架构已经足够。
本节内容引用自：37 手游基于 Flink CDC + Hudi 湖仓一体方案实践
三、最佳实践
版本选择，这个问题可能会成为困扰大家的第一个绊脚石，下面是hudi中文社区推荐的版本适配：
| Flink | Hudi |
|---|---|
| 1.12.2 | 0.9.0 |
| 1.13.1 | 0.10.0 |
建议用 Hudi master + Flink 1.13 这样可以和 CDC connector 更好地适配。
https://mvnrepository.com/artifact/org.apache.Hudi/Hudi-Flink-bundle
目前 maven 中央仓库，最新版本是 0.9.0 ，如果需要下载 0.10.0 版本 , 可以加入社区群，在共享文件中下载，也可以下载源码自行编译。
如果将 Hudi-Flink-bundle_2.11-0.10.0.jar 放到了 Flink/lib 下，则只需要如下执行即可，否则会出现各种找不到类的异常
bin/SQL-client.sh embedded
四、Flink on Hudi
新建 maven 工程，修改 pom 如下：
<?xml version="1.0" encoding="UTF-8"?><project xmlns="http://maven.apache.org/POM/4.0.0"xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"><modelVersion>4.0.0</modelVersion><groupId>org.example</groupId><artifactId>Flink_Hudi_test</artifactId><version>1.0-SNAPSHOT</version><properties><maven.compiler.source>8</maven.compiler.source><maven.compiler.target>8</maven.compiler.target><Flink.version>1.13.1</Flink.version><Hudi.version>0.10.0</Hudi.version><hadoop.version>2.10.1</hadoop.version></properties><dependencies><dependency><groupId>org.apache.hadoop</groupId><artifactId>hadoop-client</artifactId><version>${hadoop.version}</version></dependency><dependency><groupId>org.apache.hadoop</groupId><artifactId>hadoop-hdfs</artifactId><version>${hadoop.version}</version></dependency><dependency><groupId>org.apache.hadoop</groupId><artifactId>hadoop-common</artifactId><version>${hadoop.version}</version></dependency><dependency><groupId>org.apache.Flink</groupId><artifactId>Flink-core</artifactId><version>${Flink.version}</version></dependency><dependency><groupId>org.apache.Flink</groupId><artifactId>Flink-streaming-java_2.11</artifactId><version>${Flink.version}</version></dependency><dependency><groupId>org.apache.Flink</groupId><artifactId>Flink-connector-jdbc_2.11</artifactId><version>${Flink.version}</version></dependency><dependency><groupId>org.apache.Flink</groupId><artifactId>Flink-java</artifactId><version>${Flink.version}</version></dependency><dependency><groupId>org.apache.Flink</groupId><artifactId>Flink-clients_2.11</artifactId><version>${Flink.version}</version></dependency><dependency><groupId>org.apache.Flink</groupId><artifactId>Flink-table-api-java-bridge_2.11</artifactId><version>${Flink.version}</version></dependency><dependency><groupId>org.apache.Flink</groupId><artifactId>Flink-table-common</artifactId><version>${Flink.version}</version></dependency><dependency><groupId>org.apache.Flink</groupId><artifactId>Flink-table-planner_2.11</artifactId><version>${Flink.version}</version></dependency><dependency><groupId>org.apache.Flink</groupId><artifactId>Flink-table-planner-blink_2.11</artifactId><version>${Flink.version}</version></dependency><dependency><groupId>org.apache.Flink</groupId><artifactId>Flink-table-planner-blink_2.11</artifactId><version>${Flink.version}</version><type>test-jar</type></dependency><dependency><groupId>com.ververica</groupId><artifactId>Flink-connector-mySQL-CDC</artifactId><version>2.0.0</version></dependency><dependency><groupId>org.apache.Hudi</groupId><artifactId>Hudi-Flink-bundle_2.11</artifactId><version>${Hudi.version}</version><scope>system</scope><systemPath>${project.basedir}/libs/Hudi-Flink-bundle_2.11-0.10.0-SNAPSHOT.jar</systemPath></dependency><dependency><groupId>mySQL</groupId><artifactId>mySQL-connector-java</artifactId><version>5.1.49</version></dependency></dependencies></project>
我们通过构建查询insert into t2 select replace(uuid(),'-',''),id,name,description,now() from mySQL_binlog 将创建的 MySQL 表，插入到 Hudi 里。
package name.lijiaqi;import org.apache.Flink.streaming.api.environment.StreamExecutionEnvironment;import org.apache.Flink.table.api.EnvironmentSettings;import org.apache.Flink.table.api.SQLDialect;import org.apache.Flink.table.api.TableResult;import org.apache.Flink.table.api.bridge.java.StreamTableEnvironment;public class MySQLToHudiExample {public static void main(String[] args) throws Exception {EnvironmentSettings fsSettings = EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build();StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();env.setParallelism(1);StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env, fsSettings);tableEnv.getConfig().setSQLDialect(SQLDialect.DEFAULT);// 数据源表String sourceDDL ="CREATE TABLE mySQL_binlog (\n" +" id INT NOT NULL,\n" +" name STRING,\n" +" description STRING\n" +") WITH (\n" +" 'connector' = 'jdbc',\n" +" 'url' = 'jdbc:mySQL://127.0.0.1:3306/test', \n"+" 'driver' = 'com.mySQL.jdbc.Driver', \n"+" 'username' = 'root',\n" +" 'password' = 'dafei1288', \n" +" 'table-name' = 'test_CDC'\n" +")";// 输出目标表String sinkDDL ="CREATE TABLE t2(\n" +"\tuuid VARCHAR(20),\n"+"\tid INT NOT NULL,\n" +"\tname VARCHAR(40),\n" +"\tdescription VARCHAR(40),\n" +"\tts TIMESTAMP(3)\n"+// "\t`partition` VARCHAR(20)\n" +")\n" +// "PARTITIONED BY (`partition`)\n" +"WITH (\n" +"\t'connector' = 'Hudi',\n" +"\t'path' = 'hdfs://172.19.28.4:9000/Hudi_t4/',\n" +"\t'table.type' = 'MERGE_ON_READ'\n" +")" ;// 简单的聚合处理String transformSQL ="insert into t2 select replace(uuid(),'-',''),id,name,description,now() from mySQL_binlog";tableEnv.executeSQL(sourceDDL);tableEnv.executeSQL(sinkDDL);TableResult result = tableEnv.executeSQL(transformSQL);result.print();env.execute("mySQL-to-Hudi");}}
查询 Hudi
package name.lijiaqi;import org.apache.Flink.streaming.api.environment.StreamExecutionEnvironment;import org.apache.Flink.table.api.EnvironmentSettings;import org.apache.Flink.table.api.SQLDialect;import org.apache.Flink.table.api.TableResult;import org.apache.Flink.table.api.bridge.java.StreamTableEnvironment;public class ReadHudi {public static void main(String[] args) throws Exception {EnvironmentSettings fsSettings = EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build();StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();env.setParallelism(1);StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env, fsSettings);tableEnv.getConfig().setSQLDialect(SQLDialect.DEFAULT);String sourceDDL ="CREATE TABLE t2(\n" +"\tuuid VARCHAR(20),\n"+"\tid INT NOT NULL,\n" +"\tname VARCHAR(40),\n" +"\tdescription VARCHAR(40),\n" +"\tts TIMESTAMP(3)\n"+// "\t`partition` VARCHAR(20)\n" +")\n" +// "PARTITIONED BY (`partition`)\n" +"WITH (\n" +"\t'connector' = 'Hudi',\n" +"\t'path' = 'hdfs://172.19.28.4:9000/Hudi_t4/',\n" +"\t'table.type' = 'MERGE_ON_READ'\n" +")" ;tableEnv.executeSQL(sourceDDL);TableResult result2 = tableEnv.executeSQL("select * from t2");result2.print();env.execute("read_Hudi");}}
展示结果
五、Flink CDC 2.0 on Hudi
上一章节，我们使用代码形式构建实验，在本章节里，我们直接使用官网下载的 Flink 包来构建实验环境。
添加如下依赖到 $Flink_HOME/lib 下：
Hudi-Flink-bundle_2.11-0.10.0-SNAPSHOT.jar (修改 Master 分支的 Hudi Flink 版本为 1.13.2 然后构建)
hadoop-mapreduce-client-core-2.7.3.jar (解决 Hudi ClassNotFoundException)
Flink-SQL-connector-mySQL-CDC-2.0.0.jar
Flink-format-changelog-json-2.0.0.jar
Flink-SQL-connector-Kafka_2.11-1.13.2.jar
注意，在寻找 jar 的时候，CDC 2.0 更新过 group id ，不再试com.alibaba.ververica 而是改成了 com.ververica
创建 MySQL CDC 表
CREATE TABLE mySQL_users (id BIGINT PRIMARY KEY NOT ENFORCED ,name STRING,birthday TIMESTAMP(3),ts TIMESTAMP(3)) WITH ('connector' = 'mySQL-CDC','hostname' = 'localhost','port' = '3306','username' = 'root','password' = 'dafei1288','server-time-zone' = 'Asia/Shanghai','database-name' = 'test','table-name' = 'users');
创建 Hudi 表
CREATE TABLE Hudi_users5(id BIGINT PRIMARY KEY NOT ENFORCED,name STRING,birthday TIMESTAMP(3),ts TIMESTAMP(3),`partition` VARCHAR(20)) PARTITIONED BY (`partition`) WITH ('connector' = 'Hudi','table.type' = 'MERGE_ON_READ','path' = 'hdfs://localhost:9009/Hudi/Hudi_users5');
修改配置，让查询模式输出为表，设置 checkpoint
set execution.result-mode=tableau;set execution.checkpointing.interval=10sec;
进行输入导入
INSERT INTO Hudi_users5(id,name,birthday,ts, `partition`) SELECT id,name,birthday,ts,DATE_FORMAT(birthday, 'yyyyMMdd') FROM mySQL_users;查询数据
select * from Hudi_users5;执行结果
这个问题研究了很久，表面上很正常，日志也没有任何报错，也可以看出来 CDC 起作用了，有数据写入，但是就是卡在 hoodie_stream_write 上一动不动，没有数据下发。感谢社区大佬 Danny Chan 的提点，可能是 checkpoint的问题，于是做了设置：
set execution.checkpointing.interval=10sec;终于正常：
至此，Flink + Hudi 湖仓一体化方案的原型构建完成。
参考链接
[1] https://blog.csdn.net/qq_37095882/article/details/103714548
[2] https://blog.csdn.net/weixin_49218925/article/details/115511022