本文转载自公众号【麒思妙想】，具体介绍了 Flink + Hudi 湖仓一体化计划的原型构建。次要内容为：
- Hudi
- 新架构与湖仓一体
- 最佳实际
- Flink on Hudi
- Flink CDC 2.0 on Hudi
一、Hudi
1. 简介
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。
2. 为什么 Hudi 对于大规模和近实时利用很重要？
Hudi 解决了以下限度：
- HDFS 的可伸缩性限度；
- 须要在 Hadoop 中更快地出现数据；
- 没有间接反对对现有数据的更新和删除；
- 疾速的 ETL 和建模；
- 要检索所有更新的记录，无论这些更新是增加到最近日期分区的新记录还是对旧数据的更新，Hudi 都容许用户应用最初一个检查点工夫戳。此过程不必执行扫描整个源表的查问。
3. 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 湖仓一体计划实际》）
三、最佳实际
1. 版本搭配
版本抉择，这个问题可能会成为困扰大家的第一个绊脚石，上面是 Hudi 中文社区举荐的版本适配：
| Flink | Hudi |
|---|---|
| 1.12.2 | 0.9.0 |
| 1.13.1 | 0.10.0 |
倡议用 Hudi master +Flink 1.13 这样能够和 CDC connector 更好地适配。
2. 下载Hudi
https://mvnrepository.com/art...
目前 maven 地方仓库，最新版本是 0.9.0 ，如果须要下载 0.10.0 版本 , 能够退出社区群，在共享文件中下载，也能够下载源码自行编译。
3. 执行
如果将 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 包来构建试验环境。
1. 增加依赖
增加如下依赖到 $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
2. 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;
执行后果
3. 卡执行打算
这个问题钻研了很久，外表上很失常，日志也没有任何报错，也能够看进去 CDC 起作用了，有数据写入，然而就是卡在 hoodie_stream_write 上一动不动，没有数据下发。感激社区大佬 Danny Chan 的提点，可能是 checkpoint的问题，于是做了设置
set execution.checkpointing.interval=10sec;
于是终于失常：
至此，Flink + Hudi 湖仓一体化计划的原型构建实现。
参考链接
https://blog.csdn.net/weixin_...
https://blog.csdn.net/qq_3709...
https://mp.weixin.qq.com/s/xo...
更多 Flink 相干技术问题，可扫码退出社区钉钉交换群；
第一工夫获取最新技术文章和社区动静，请关注公众号～