Flink+Hudi 构架湖仓一体化解决方案

本文涉及的产品
实时计算 Flink 版,5000CU*H 3个月
简介: 详细介绍了 Flink + Hudi 湖仓一体化方案的原型构建。

本文转载自公众号【麒思妙想】,详细介绍了 Flink + Hudi 湖仓一体化方案的原型构建。主要内容为:

  1. Hudi
  2. 新架构与湖仓一体
  3. 最佳实践
  4. Flink on Hudi
  5. 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/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>
AI 代码解读

我们通过构建查询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");
    }
}
AI 代码解读

查询 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");
    }
}
AI 代码解读

展示结果

图片

五、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'   
);
AI 代码解读

创建 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'
);
AI 代码解读

修改配置,让查询模式输出为表,设置 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_49218925/article/details/115511022

https://blog.csdn.net/qq_37095882/article/details/103714548

https://mp.weixin.qq.com/s/xoucbJxzO2Zkq_b2_WDUbA

相关实践学习
基于Hologres轻松玩转一站式实时仓库
本场景介绍如何利用阿里云MaxCompute、实时计算Flink和交互式分析服务Hologres开发离线、实时数据融合分析的数据大屏应用。
Linux入门到精通
本套课程是从入门开始的Linux学习课程,适合初学者阅读。由浅入深案例丰富,通俗易懂。主要涉及基础的系统操作以及工作中常用的各种服务软件的应用、部署和优化。即使是零基础的学员,只要能够坚持把所有章节都学完,也一定会受益匪浅。
目录
打赏
0
0
0
0
36033
分享
相关文章
基于 Flink+Paimon+Hologres 搭建淘天集团湖仓一体数据链路
本文整理自淘天集团高级数据开发工程师朱奥在Flink Forward Asia 2024的分享,围绕实时数仓优化展开。内容涵盖项目背景、核心策略、解决方案、项目价值及未来计划五部分。通过引入Paimon和Hologres技术,解决当前流批存储不统一、实时数据可见性差等痛点,实现流批一体存储与高效近实时数据加工。项目显著提升了数据时效性和开发运维效率,降低了使用门槛与成本,并规划未来在集团内推广湖仓一体架构,探索更多技术创新场景。
695 2
基于 Flink+Paimon+Hologres 搭建淘天集团湖仓一体数据链路
Lalamove基于Flink实时湖仓演进之路
本文由货拉拉国际化技术部资深数据仓库工程师林海亮撰写,围绕Flink在实时数仓中的应用展开。文章首先介绍了Lalamove业务背景,随后分析了Flink在实时看板、数据服务API、数据监控及数据分析中的应用与挑战,如多数据中心、时区差异、上游改造频繁及高成本问题。接着阐述了实时数仓架构从无分层到引入Paimon湖仓的演进过程,解决了数据延迟、兼容性及资源消耗等问题。最后展望未来,提出基于Fluss+Paimon优化架构的方向,进一步提升性能与降低成本。
166 11
Lalamove基于Flink实时湖仓演进之路
京东物流基于Flink & StarRocks的湖仓建设实践
本文整理自京东物流高级数据开发工程师梁宝彬在Flink Forward Asia 2024的分享,聚焦实时湖仓的探索与建设、应用实践、问题思考及未来展望。内容涵盖京东物流通过Flink和Paimon等技术构建实时湖仓体系的过程,解决复杂业务场景下的数据分析挑战,如多维OLAP分析、大屏监控等。同时,文章详细介绍了基于StarRocks的湖仓一体方案,优化存储成本并提升查询效率,以及存算分离的应用实践。最后,对未来数据服务的发展方向进行了展望,计划推广长周期数据存储服务和原生数据湖建设,进一步提升数据分析能力。
274 1
京东物流基于Flink & StarRocks的湖仓建设实践
Flink + Doris 实时湖仓解决方案
本文整理自SelectDB技术副总裁陈明雨在Flink Forward Asia 2024的分享,聚焦Apache Doris与湖仓一体解决方案。内容涵盖三部分:一是介绍Apache Doris,一款高性能实时分析数据库,支持多场景应用;二是基于Doris、Flink和Paimon的湖仓解决方案,解决批流融合与数据一致性挑战;三是Doris社区生态及云原生发展,包括存算分离架构与600多位贡献者的活跃社区。文章深入探讨了Doris在性能、易用性及场景支持上的优势,并展示了其在多维分析、日志分析和湖仓分析中的实际应用案例。
294 17
Flink + Doris 实时湖仓解决方案
​Flink+Paimon+Hologres,面向未来的一体化实时湖仓平台架构设计
​Flink+Paimon+Hologres,面向未来的一体化实时湖仓平台架构设计
113 4
鹰角基于 Flink + Paimon + Trino 构建湖仓一体化平台实践项目
鹰角基于 Flink + Paimon + Trino 构建湖仓一体化平台实践项目
289 2
Flink基于Paimon的实时湖仓解决方案的演进
Flink基于Paimon的实时湖仓解决方案的演进
客户说|保险极客引入阿里云AnalyticDB,多业务场景效率大幅提升
“通过引入AnalyticDB,我们在复杂数据查询和实时同步方面取得了显著突破,其分布式、弹性与云计算的优势得以充分体现,帮助企业快速响应业务变化,实现降本增效。AnalyticDB的卓越表现保障了保险极客数据服务的品质和效率。”
美的楼宇科技基于阿里云 EMR Serverless Spark 构建 LakeHouse 湖仓数据平台
美的楼宇科技基于阿里云 EMR Serverless Spark 建设 IoT 数据平台,实现了数据与 AI 技术的有效融合,解决了美的楼宇科技设备数据量庞大且持续增长、数据半结构化、数据价值缺乏深度挖掘的痛点问题。并结合 EMR Serverless StarRocks 搭建了 Lakehouse 平台,最终实现不同场景下整体性能提升50%以上,同时综合成本下降30%。
447 58
阿里云 EMR Serverless StarRocks3.x,极速统一的湖仓新范式
阿里云 EMR Serverless StarRocks3.x,极速统一的湖仓新范式
102 0

相关产品

  • 实时计算 Flink版
  • AI助理

    你好,我是AI助理

    可以解答问题、推荐解决方案等

    登录插画

    登录以查看您的控制台资源

    管理云资源
    状态一览
    快捷访问