Flink + Hudi，构架仓湖一体化解决方案
Flink + Hudi，构架仓湖一体化解决方案
在《如何利用 Flink CDC 实现数据增量备份到 Clickhouse》里，我们介绍了如何cdc到ck，今天我们依旧使用前文的案例，来sink到hudi，那么我们开始吧。
Hudi
Apache Hudi(发音为“Hoodie”)在DFS的数据集上提供以下流原语
•插入更新 (如何改变数据集?)•增量拉取 (如何获取变更的数据?)
Hudi维护在数据集上执行的所有操作的时间轴（timeline），以提供数据集的即时视图。Hudi将数据集组织到与Hive表非常相似的基本路径下的目录结构中。数据集分为多个分区，文件夹包含该分区的文件。每个分区均由相对于基本路径的分区路径唯一标识。
分区记录会被分配到多个文件。每个文件都有一个唯一的文件ID和生成该文件的提交（commit）。如果有更新，则多个文件共享相同的文件ID，但写入时的提交（commit）不同。
存储类型–处理数据的存储方式
•写时复制•纯列式•创建新版本的文件•读时合并•近实时
视图–处理数据的读取方式
读取优化视图-输入格式仅选择压缩的列式文件
•parquet文件查询性能•500 GB的延迟时间约为30分钟•导入现有的Hive表
近实时视图
•混合、格式化数据•约1-5分钟的延迟•提供近实时表
增量视图
•数据集的变更•启用增量拉取
Hudi存储层由三个不同的部分组成
元数据–它以时间轴的形式维护了在数据集上执行的所有操作的元数据，该时间轴允许将数据集的即时视图存储在基本路径的元数据目录下。时间轴上的操作类型包括
•提交（commit），一次提交表示将一批记录原子写入数据集中的过程。单调递增的时间戳，提交表示写操作的开始。•清理（clean），清理数据集中不再被查询中使用的文件的较旧版本。•压缩（compaction），将行式文件转化为列式文件的动作。•索引，将传入的记录键快速映射到文件（如果已存在记录键）。索引实现是可插拔的，Bloom过滤器-由于不依赖任何外部系统，因此它是默认配置，索引和数据始终保持一致。Apache HBase-对少量key更高效。在索引标记过程中可能会节省几秒钟。•数据，Hudi以两种不同的存储格式存储数据。实际使用的格式是可插入的，但要求具有以下特征–读优化的列存储格式（ROFormat），默认值为Apache Parquet；写优化的基于行的存储格式（WOFormat），默认值为Apache Avro。
为什么Hudi对于大规模和近实时应用很重要？
Hudi解决了以下限制
•HDFS的可伸缩性限制•需要在Hadoop中更快地呈现数据•没有直接支持对现有数据的更新和删除•快速的ETL和建模•要检索所有更新的记录，无论这些更新是添加到最近日期分区的新记录还是对旧数据的更新，Hudi都允许用户使用最后一个检查点时间戳。此过程不用执行扫描整个源表的查询
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 1.12.2/1.13.1 Hudi 0.9.0/0.10.0
官方说的支持版本是这样， 不过目前我的1.13和0.10组合并没有配置成功，所以大家还是尽量选择 1.12.2+0.9.0 吧，配合scala 2.11 。
下载hudi
https://mvnrepository.com/artifact/org.apache.hudi/hudi-flink-bundle
执行
如果将 hudi-flink-bundle_2.11-0.9.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> </properties> <dependencies> <dependency> <groupId>org.apache.hadoop</groupId> <artifactId>hadoop-client</artifactId> <version>2.9.2</version> </dependency> <dependency> <groupId>org.apache.hadoop</groupId> <artifactId>hadoop-hdfs</artifactId> <version>2.9.2</version> </dependency> <dependency> <groupId>org.apache.hadoop</groupId> <artifactId>hadoop-common</artifactId> <version>2.9.2</version> </dependency> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-core</artifactId> <version>1.12.2</version> </dependency> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-streaming-java_2.11</artifactId> <version>1.12.2</version> </dependency> <!-- <dependency>--> <!-- <groupId>org.apache.flink</groupId>--> <!-- <artifactId>flink-jdbc_2.12</artifactId>--> <!-- <version>1.10.3</version>--> <!-- </dependency>--> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-connector-jdbc_2.11</artifactId> <version>1.12.2</version> </dependency> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-java</artifactId> <version>1.12.2</version> </dependency> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-clients_2.11</artifactId> <version>1.12.2</version> </dependency> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-table-api-java-bridge_2.11</artifactId> <version>1.12.2</version> </dependency> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-table-common</artifactId> <version>1.12.2</version> </dependency> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-table-planner_2.11</artifactId> <version>1.12.2</version> </dependency> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-table-planner-blink_2.11</artifactId> <version>1.12.2</version> </dependency> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-table-planner-blink_2.11</artifactId> <version>1.12.2</version> <type>test-jar</type> </dependency> <dependency> <groupId>com.alibaba.ververica</groupId> <artifactId>flink-connector-mysql-cdc</artifactId> <version>1.2.0</version> </dependency><!-- <dependency>--><!-- <groupId>org.apache.hudi</groupId>--><!-- <artifactId>hudi-flink-client</artifactId>--><!-- <version>0.9.0</version>--><!-- </dependency>--><!-- <dependency>--><!-- <groupId>org.apache.hudi</groupId>--><!-- <artifactId>hudi-common</artifactId>--><!-- <version>0.9.0</version>--><!-- </dependency>--><!-- <dependency>--><!-- <groupId>org.apache.hudi</groupId>--><!-- <artifactId>hudi-hadoop-mr-bundle</artifactId>--><!-- <version>0.9.0</version>--><!-- </dependency>--> <dependency> <groupId>org.apache.hudi</groupId> <artifactId>hudi-flink-bundle_2.11</artifactId> <version>0.9.0</version> </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_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
Flink SQL CDC On Hudi
创建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;执行结果
卡执行计划
这个问题坑了我好几天，一度都打算放弃hudi了，表面上很正常，日志也没有任何报错，也可以看出来cdc起作用了，有数据写入，但是就是卡在 hoodie_stream_write 上一动不动了，没有数据下发了。感谢社区大佬 Danny Chan 的提点，可能是 checkpoint的问题，于是做了设置
set execution.checkpointing.interval=10sec;终于正常了
致此，Flink + Hudi 仓湖一体化方案的原型构建完成，感谢大家看到这里，如果对你有点点帮助的话，希望点个关注，转发。您的举手之劳，会对我非常有帮助，非常感谢。
腾讯云开发者
