Flink 使用介绍相关文档目录

Flink 使用介绍相关文档目录

环境信息

• Hadoop 3.1.1
• Flink 1.17.2
• Hudi 0.15.0
• MySQL 5.7.x

MySQL开启Binlog

具体步骤参见：Flink 使用之 MySQL CDC

编译Hudi

首先clone HUDI项目。

git clone https://github.com/apache/hudi.git

切换到release-0.15.0分支。然后执行如下编译命令：

mvn clean package -Dflink1.17 -Dscala2.12 -Dspark3.3 -DskipTests -Pflink-bundle-shade-hive3

编译完毕后输出的hudi-flink1.17-bundle-0.15.0.jar位于hudi/packaging/hudi-flink-bundle/target中。复制走备用。

Flink配置

下载Flink 1.17.2二进制包解压到服务器任意目录备用。

配置checkpoint

编辑$FLINK_HOME/conf/flink-conf.yaml文件，加入如下配置启用checkpoint。示例checkpoint间隔时间为3s。

execution.checkpointing.interval: 3s

添加依赖

添加如下依赖到$FLINK_HOME/lib目录中。

• flink-sql-connector-mysql-cdc-3.1.0.jar 点我下载
• hudi-flink1.17-bundle-0.15.0.jar
• mysql-connector-java-8.0.27.jar 点我下载

操作演示

下面以student表为例，演示MySQL CDC到Hudi全过程。

student表有3个字段：

• id int类型
• name varchar类型
• score int类型

首先我们创建MySQL表，并写入初始数据：

create table student(
    id int,
    name varchar(50),
    score int
);

insert into student values(1, 'Paul', 123456),(2, 'Kate', 654321),(3, 'Peter', 222222);

接下来进入Flink SQL Client，创建student CDC表和hudi_student表，然后插入CDC表的内容到Hudi表中。需要执行如下SQL：

CREATE TABLE student (
    id int,
    name string,
    score int,
    PRIMARY KEY(id) NOT ENFORCED
) WITH (
    'connector' = 'mysql-cdc',
    'hostname' = '192.168.0.1',
    'port' = '3306',
    'username' = 'root',
    'password' = '123456',
    'database-name' = 'demo',
    'table-name' = 'student',
    'scan.incremental.snapshot.chunk.key-column' = 'id'
);

CREATE TABLE hudi_student (
    id int,
    name string,
    score int,
    PRIMARY KEY(id) NOT ENFORCED
) with (
    'connector' = 'hudi',
    'path' = 'hdfs:///hudi_student',
    'table.type' = 'MERGE_ON_READ',
    'compaction.async.enabled' = 'true',
    'compaction.trigger.strategy' = 'num_commits',
    'compaction.delta_commits' = '5'
);


insert into hudi_student select * from student;

需要注意的是，如果MySQL表没有指定主键约束，CDC表的属性必须要添加'scan.incremental.snapshot.chunk.key-column'配置项用来指定主键。

最后执行：

select * from hudi_student;

可以查看同步到hudi_student表中的全量数据。

          id                           name       score
           1                           Paul      123456
           2                           Kate      654321
           3                          Peter      222222

新增/修改/删除同步演示

新增数据

在MySQL控制台执行：

insert into student values(4, Tom, 444444);

等待一段时间之后，在Flink SQL Client执行：

select * from hudi_student;

得到查询结果：

          id                           name       score
           1                           Paul      123456
           2                           Kate      654321
           3                          Peter      222222
           4                           Tom       444444

新的数据已经追加到了Hudi表中。

修改数据

在MySQL控制台执行：

update student set score=333333 where id=2;

等待一段时间之后，在Flink SQL Client执行：

select * from hudi_student;

得到查询结果：

          id                           name       score
           1                           Paul      123456
           2                           Kate      333333
           3                          Peter      222222
           4                           Tom       444444

发现id为2的数据已经修改。

删除数据

在MySQL控制台执行：

delete from student where id=2;

等待一段时间之后，在Flink SQL Client执行：

select * from hudi_student;

得到查询结果：

          id                           name       score
           1                           Paul      123456
           3                          Peter      222222
           4                           Tom       444444

发现id为2的数据已经删除。

MySQL CDC实时增量入湖的配置和演示过程到此结束。

附录

Oracle CDC情况说明

经本人试验Oracle CDC入湖存在如下问题：

Oracle CDC使用Flink SQL的方式只能获取到最后一条数据。无法读取到初始的全量数据。应为Flink CDC bug所致。参见：https://developer.aliyun.com/ask/623649。

Oracle CDC的延迟非常高。即便是按照文档添加了：

'debezium.log.mining.strategy' = 'online_catalog',
'debezium.log.mining.continuous.mine' = 'true'

效果仍然不好。

Oracle CDC增量修改的数据同步到Hudi表之后会导致Hudi表数据消失，原因未知。

本人结论，不建议使用Oracle CDC实时增量入湖，现阶段问题较多。