搜索

Flink+hudi 构架沧湖一体化解决方案

程序源代码

共 10777字，需浏览 22分钟

 · 2021-12-14

简介

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和建模

新架构与湖仓一体

通过湖仓一体、流批一体，准实时场景下做到了：数据同源、同计算引擎、同存储、同计算口径。数据的时效性可以到分钟级，能很好的满足业务准实时数仓的需求。下面是架构图：

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 架构已经足够。

最佳实践

版本搭配

版本选择，这个问题可能会成为困扰大家的第一个绊脚石，下面是hudi中文社区推荐的版本适配：

flinkhudi
1.12.20.9.0
1.13.10.10.0

建议用 hudi master +flink 1.13 这样可以和 cdc connector 更好地适配。

下载hudi

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如下


<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.0modelVersion>

    <groupId>org.examplegroupId>
    <artifactId>flink_hudi_testartifactId>
    <version>1.0-SNAPSHOTversion>

    <properties>
        <maven.compiler.source>8maven.compiler.source>
        <maven.compiler.target>8maven.compiler.target>
        <flink.version>1.13.1flink.version>
        <hudi.version>0.10.0hudi.version>
        <hadoop.version>2.10.1hadoop.version>
    properties>

    <dependencies>
        <dependency>
            <groupId>org.apache.hadoopgroupId>
            <artifactId>hadoop-clientartifactId>
            <version>${hadoop.version}version>
        dependency>
        <dependency>
            <groupId>org.apache.hadoopgroupId>
            <artifactId>hadoop-hdfsartifactId>
            <version>${hadoop.version}version>
        dependency>
        <dependency>
            <groupId>org.apache.hadoopgroupId>
            <artifactId>hadoop-commonartifactId>
            <version>${hadoop.version}version>
        dependency>


        <dependency>
            <groupId>org.apache.flinkgroupId>
            <artifactId>flink-coreartifactId>
            <version>${flink.version}version>
        dependency>
        <dependency>
            <groupId>org.apache.flinkgroupId>
            <artifactId>flink-streaming-java_2.11artifactId>
            <version>${flink.version}version>
        dependency>

        <dependency>
            <groupId>org.apache.flinkgroupId>
            <artifactId>flink-connector-jdbc_2.11artifactId>
            <version>${flink.version}version>
        dependency>

        <dependency>
            <groupId>org.apache.flinkgroupId>
            <artifactId>flink-javaartifactId>
            <version>${flink.version}version>
        dependency>
        <dependency>
            <groupId>org.apache.flinkgroupId>
            <artifactId>flink-clients_2.11artifactId>
            <version>${flink.version}version>
        dependency>
        <dependency>
            <groupId>org.apache.flinkgroupId>
            <artifactId>flink-table-api-java-bridge_2.11artifactId>
            <version>${flink.version}version>
        dependency>
        <dependency>
            <groupId>org.apache.flinkgroupId>
            <artifactId>flink-table-commonartifactId>
            <version>${flink.version}version>
        dependency>

        <dependency>
            <groupId>org.apache.flinkgroupId>
            <artifactId>flink-table-planner_2.11artifactId>
            <version>${flink.version}version>
        dependency>

        <dependency>
            <groupId>org.apache.flinkgroupId>
            <artifactId>flink-table-planner-blink_2.11artifactId>
            <version>${flink.version}version>
        dependency>
        <dependency>
            <groupId>org.apache.flinkgroupId>
            <artifactId>flink-table-planner-blink_2.11artifactId>
            <version>${flink.version}version>
            <type>test-jartype>
        dependency>

        <dependency>
            <groupId>com.ververicagroupId>
            <artifactId>flink-connector-mysql-cdcartifactId>
            <version>2.0.0version>
        dependency>

        <dependency>
            <groupId>org.apache.hudigroupId>
            <artifactId>hudi-flink-bundle_2.11artifactId>
            <version>${hudi.version}version>
            <scope>systemscope>
            <systemPath>${project.basedir}/libs/hudi-flink-bundle_2.11-0.10.0-SNAPSHOT.jarsystemPath>
        dependency>

        <dependency>
            <groupId>mysqlgroupId>
            <artifactId>mysql-connector-javaartifactId>
            <version>5.1.49version>
        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 


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 仓湖一体化方案的原型构建完成，感谢大家看到这里。

Hi，我是王知无，一个大数据领域的原创作者。 

放心关注我，获取更多行业的一手消息。



八千里路云和月 | 从零到大数据专家学习路径指南

互联网最坏的时代可能真的来了

我在B站读大学，大数据专业

我们在学习Flink的时候，到底在学习什么？

193篇文章暴揍Flink，这个合集你需要关注一下

Flink生产环境TOP难题与优化，阿里巴巴藏经阁YYDS

Flink CDC我吃定了耶稣也留不住他！| Flink CDC线上问题小盘点

我们在学习Spark的时候，到底在学习什么？

在所有Spark模块中，我愿称SparkSQL为最强！

硬刚Hive | 4万字基础调优面试小总结

数据治理方法论和实践小百科全书

标签体系下的用户画像建设小指南

4万字长文 | ClickHouse基础&实践&调优全视角解析

【面试&个人成长】2021年过半，社招和校招的经验之谈

大数据方向另一个十年开启 |《硬刚系列》第一版完结

我写过的关于成长/面试/职场进阶的文章

当我们在学习Hive的时候在学习什么？「硬刚Hive续集」

浏览 34
点赞
评论
收藏
分享

手机扫一扫分享

举报
评论
图片
表情
推荐
Windows格式化对话框是一个使用了30年的 “临时解决方案”
戴夫-普卢默（Dave Plummer）是微软的资深工程师，曾创造了任务管理器、Windows 弹球、原生 ZIP 支持（微软出钱买断该功能后，他用这笔钱购买了一辆红色克尔维特）等传奇。近日他在自己的 X 账户上分享了创建 "格式化" 对话框的故事 —— 称其是一个使用了长达 30 年的 “临时解决
开源Linux
0
多人同时导出 Excel 干崩服务器！新来的阿里大佬给出的解决方案太优雅了！
点击关注公众号，Java 干货及时推送↓推荐阅读：面试辅导，我们出大成果了！来源：juejin.cn/post/7259249904777838629前言 业务诉求：考虑到数据库数据日渐增多，导出会有全量数据的导出，多人同时导出可以会对服务性能造成影响，导出涉及到mysql查询的io操作，
Java技术栈
1
IT 运维整体解决方案（PPT）
本文作为一个IT运维整体解决方案PPT，包括了步骤规划、平台设计、事件处理流程、监控、各设备及系统管理、工作流引擎等各个要素，对撰写运维方案具有参考价值，部分内容可作为模板借鉴。（*因成文时间所限，文中举例的部分产品技术可能过时，但产品技术并非本文重点。）图片点击可放大原题：IT运维整体解决方案，点
数据中心运维管理
0
一站式解决方案：基于 Arthas 实现服务发现和权限控制
来源：juejin.cn/post/7281849496983994383👉 欢迎加入小哈的星球 ，你将获得: 专属的项目实战 / Java 学习路线 / 一对一提问 / 学习打卡 /  赠书福利全栈前后端分离博客项目 2.0 版本完结啦， 演示链接
小哈学Java
0
编辑选择：工业互联网标识解析应用优秀解决方案（四款）
随着工业互联网的快速发展，标识解析技术作为实现设备互联、数据互通及服务互操作的重要支撑，已然成为物理世界与数字世界之间的关键桥梁。它不仅推动着工业资源的高效配置和生产流程的持续优化，而且助力整个产业链的深度融合。但面对复杂且多变的工业环境，现有标识解析解决方案仍面临标准化、成熟度不一以及安全隐私方面
工业互联网世界
0
Google 应用无法安装解决方案
和你一起终身学习，这里是程序员Android经典好文推荐，通过阅读本文，您将收获以下知识点:一、 国家或地区限制导致无法安装二、手机feature不支持导致无法安装Google PlayStore应用无法安装的原因：一、 国家或地区限制导致无法安装某些应用只在某些国家和地区才能使用，所以Google
程序员Android
10
湖蓝色吊带时尚搭配芭比裤
在盛夏的灿烂阳光下，时尚绝非仅仅是衣物的搭配，它更是一种深入骨髓的生活态度，一种悠然自得的享受。今天，就让我们携手漫步于时尚的殿堂，共同领略其中那无尽的魅力与风情。身姿轻盈飘逸，宛如风中的仙子，她优雅...
全栈架构社区
0
SpaceX被曝正在建设间谍卫星网络；国家数据局：加快构建全国一体化算力网；华为星闪应用今年将首推家居场景
智次方 追踪智能产业新趋势知情人士：马斯克的SpaceX公司正在为美国情报机构建立一个间谍卫星网络据五名知情人士透露，SpaceX公司正在与美国情报机构签订一份机密合同，建立一个由数百颗间谍卫星组成的网络，这表明...
物联网头条君
0
EdgeOne一体化服务,安全加速两不误
作为一位网络技术的爱好者，我一直对腾讯云开发的EdgeOne产品充满好奇。最近看到腾讯云有征文活动，有幸亲自体验了这款革命性的产品，并感受到了它所带来的便捷与安全。接下来，我将以一个新手的视角，分享我的使用...
人生不止有技术
0
解决全屋智能落地难题 多品牌一站式解决方案亮相2024深圳智博会
首届深圳国际智能家居博览会已于深圳国际会展中心盛大开幕，展期持续2024年3月17-20日，共4天。作为驱动中国智能家居产业技术革新与创新应用的会展引擎，首届深圳智博会聚焦开源鸿蒙、星闪技术、电力线载波、人工智...
天极网
0
点赞
评论
收藏
分享

手机扫一扫分享

举报