▼ 关注「Flink 中文社区」，获取更多技术干货 ▼

摘要：本文详细介绍了 Flink + Hudi 湖仓一体化方案的原型构建。主要内容为：

  1. Hudi

  2. 新架构与湖仓一体

  3. 最佳实践

  4. Flink on Hudi

  5. Flink CDC 2.0 on Hudi

Tips：FFA 2021 重磅开启，点击「阅读原文」即可报名～

 GitHub 地址 

欢迎大家给 Flink 点赞送 star~

一、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/artifact/org.apache.Hudi/Hudi-Flink-bundle

目前 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 湖仓一体化方案的原型构建完成。

参考链接

[1] https://blog.csdn.net/qq_37095882/article/details/103714548

[2] https://blog.csdn.net/weixin_49218925/article/details/115511022

  Flink Forward Asia 2021  

报名现已开放

Flink Forward Asia 2021 重磅启动！FFA 2021 将于 12 月 4-5 日在北京·国家会议中心举办，预计将有 3000+ 开发者参与，探讨交流 Flink 最新动态。报名通道已开启，扫描下图二维码，或点击文末「阅读原文」即可报名 FFA 2021～

更多 Flink 相关技术问题，可扫码加入社区钉钉交流群～

  戳我，报名 FFA 2021 大会！

Flink+Hudi 构架湖仓一体化解决方案相关推荐

  1. 华为云FusionInsight湖仓一体解决方案的前世今生

    摘要:华为云发布新一代智能数据湖华为云FusionInsight时再次提到了湖仓一体理念,那我们就来看看湖仓一体的前世今生. 伴随5G.大数据.AI.IoT的飞速发展,数据呈现大规模.多样性的极速增长 ...

  2. 浅谈湖仓一体化对上层机器学习业务的促进

    背景概述 最近湖仓一体化的概念在大数据圈子突然蹿红,知乎上很多大神已经分析了湖仓一体化主要的革新点,今天主要介绍下湖仓一体化对机器学习业务的影响. 还是简单讲下"湖"和" ...

  3. 汽车之家基于 Flink + Iceberg 的湖仓一体架构实践

    简介:由汽车之家实时计算平台负责人邸星星在 4 月 17 日上海站 Meetup 分享的,基于 Flink + Iceberg 的湖仓一体架构实践. 内容简要: 一.数据仓库架构升级的背景 二.基于 ...

  4. 汽车之家：基于 Flink + Iceberg 的湖仓一体架构实践

    简介: 由汽车之家实时计算平台负责人邸星星在 4 月 17 日上海站 Meetup 分享的,基于 Flink + Iceberg 的湖仓一体架构实践. 内容简要: 一.数据仓库架构升级的背景 二.基于 ...

  5. 湖仓一体化的路，很多人都只走了一半

    2022已至,如果回看2021,这一年无疑是数据的价值进一步体现的一年.数据应用场景不断丰富,从工业.交通.金融到制造,几乎无处不在.当然,数据价值的迅速提升也给开发者和相关企业带来了新的问题.数据量 ...

  6. 湖仓一体化：铁打的数据仓 流水的数据湖产品

    国际研究机构MarketsandMarkets的最新研究报告显示,到2024年,全球数据湖市场将突破200亿美元,增至201亿美元,复合年增长率将高达20.6%.可以说,随着数据治理与应用需求激增,数 ...

  7. 技术专家太玄：企业数字化转型中数据底座“湖仓一体化”l 华坤道威专访

    随着大数据领域的迅速发展和普惠化,数据建设越来越受到企业的高度关注,而"湖仓一体化"也成为数据建设中重要的组成部分,越来越多企业把"湖仓一体化"视为数字变革的重 ...

  8. Spark+Flink+Iceberg打造湖仓一体架构实践探索

    数据湖-大数据生态杀青 数据仓库的痛点 只能存储结构化数据,无法采集存储非结构化数据 无法存储原始数据,所有的数据须经过ETL清洗过程 离线数仓的数据表牵一发而动全身,数据调整工程量大 实时数仓存储空 ...

  9. 37 手游基于 Flink CDC + Hudi 湖仓一体方案实践

    简介: 介绍了 37 手游为何选择 Flink 作为计算引擎,并如何基于 Flink CDC + Hudi 构建新的湖仓一体方案. 本文作者是 37 手游大数据开发徐润柏,介绍了 37 手游为何选择 ...

  10. 湖仓一体技术调研(Apache Hudi、Iceberg和Delta lake对比)

    湖仓一体技术调研(Apache Hudi.Iceberg和Delta lake对比) 作者:程哥哥.刘某迎 .杜某安.刘某.施某宇.严某程 1 引 言 ​ 随着当前的大数据技术逐步革新,企业对单一的数 ...

最新文章

  1. 1096 Consecutive Factors (20 分)【难度: 一般 / 爆搜 数论】
  2. 20-javamail
  3. Amazon Aurora 深度探索
  4. c语言使用正则,C语言中使用正则表达式
  5. mysql mos login_MySQL 中常用的函数
  6. Qt5.7 win10环境 调试器未设置问题解决
  7. silverlight学习笔记（一）—— 使用blend实现缩小Listbox或Scrollviewer中的滚动条
  8. Json.net说法——（四）序列化错误处理
  9. archlinux i3wm flameshot使用
  10. 一次手机木马的清除记录（手机刷机）
  11. 解决或者设置网页变灰-CSS-filter属性-哀悼
  12. python判断天数_Python计算两个日期相差天数的方法示例
  13. RISC-V 指令架构 -- 模式切换
  14. 自编程实现决策树（使用ID3算法）
  15. Java应用性能优化！宁波java培训班地址
  16. Xcode支持iOS6、iOS7版本
  17. 个人MBTI性格分析和五大性格特质
  18. 《c语言入门题目3》将摄氏度转化为华氏度
  19. Eplan符号比例缩放（缩小设置）
  20. 2017年计算机一级b试题及答案,2017年全国计算机一级试题及答案「精品」

热门文章

  1. JavaWeb：Ajax和JSON
  2. SQL：MySQL中建立一个新的数据表
  3. Javascript：ES6中新增class类中静态方法如何在类内部被调用
  4. sql语句区分大小写吗？
  5. 论文笔记_S2D.48_2017-IEEE RAL_单视图和多视图深度融合
  6. pytorch在anaconda下安装
  7. 综述 | 基于特征的视觉同步定位和建图
  8. fixed不以body定位，相对于父级容器定位问题
  9. 2017-2018-2 20165218 实验五《网络编程与安全》实验报告
  10. UVa140 - Bandwidth