关于Flink:FlinkHudi-构架湖仓一体化解决方案

本文转载自公众号【麒思妙想】，具体介绍了 Flink + Hudi 湖仓一体化计划的原型构建。次要内容为：

Hudi

新架构与湖仓一体

最佳实际

Flink on Hudi

Flink CDC 2.0 on Hudi

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。

Hudi 解决了以下限度：

HDFS 的可伸缩性限度；
须要在 Hadoop 中更快地出现数据；
没有间接反对对现有数据的更新和删除；
疾速的 ETL 和建模；
要检索所有更新的记录，无论这些更新是增加到最近日期分区的新记录还是对旧数据的更新，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	Hudi
1.12.2	0.9.0
1.13.1	0.10.0

倡议用 Hudi master +Flink 1.13 这样能够和 CDC connector 更好地适配。

https://mvnrepository.com/art…

目前 maven 地方仓库，最新版本是 0.9.0，如果须要下载 0.10.0 版本 , 能够退出社区群，在共享文件中下载，也能够下载源码自行编译。

如果将 Hudi-Flink-bundle_2.11-0.10.0.jar 放到了 Flink/lib 下，则只须要如下执行即可，否则会呈现各种找不到类的异样

bin/SQL-client.sh embedded

新建 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 包来构建试验环境。

增加如下依赖到 $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

创立 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;

执行后果

这个问题钻研了很久，外表上很失常，日志也没有任何报错，也能够看进去 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 相干技术问题，可扫码退出社区钉钉交换群；

第一工夫获取最新技术文章和社区动静，请关注公众号～

关于Flink:FlinkHudi-构架湖仓一体化解决方案

一、Hudi

1. 简介

2. 为什么 Hudi 对于大规模和近实时利用很重要？

3. Hudi 的劣势

二、新架构与湖仓一体

三、最佳实际

1. 版本搭配

2. 下载 Hudi

3. 执行

四、Flink on Hudi

五、Flink CDC 2.0 on Hudi

1. 增加依赖

2. Flink SQL CDC on Hudi

3. 卡执行打算

Just My Socks（注册教程内含优惠码）

关于Flink:FlinkHudi-构架湖仓一体化解决方案

一、Hudi

1. 简介

2. 为什么 Hudi 对于大规模和近实时利用很重要？

3. Hudi 的劣势

二、新架构与湖仓一体

三、最佳实际

1. 版本搭配

2. 下载 Hudi

3. 执行

四、Flink on Hudi

五、Flink CDC 2.0 on Hudi

1. 增加依赖

2. Flink SQL CDC on Hudi

3. 卡执行打算

Just My Socks（注册教程 内含优惠码）

Just My Socks（注册教程内含优惠码）