Flinkcdc 实现 MySQL 写入 Doris

一、环境配置

Doris：3.0.4 + JDK 17
MySQL （业务数据库）：5.7
MySQL（本地数据库）：5.7
Flink：flink-1.19.1
flinkcdc：flink-cdc-3.3.0

二、环境搭建

下载 Flink 1.19.1

wget https://archive.apache.org/dist/flink/flink-1.19.1/flink-1.19.1-bin-scala_2.12.tgz

解压后得到 flink-1.19.1 目录，设置 FLINK_HOME 为 flink-1.19.1 所在目录

sudo vim /etc/profile.d/my_env.sh

添加 Flink 路径

export JAVA_HOME=/opt/module/java
export PATH=$PATH:$JAVA_HOME/bin
export FLINK_HOME=/opt/module/flink-1.19.1
export PATH=$PATH:$FLINK_HOME/bin

读入数据通过在 conf/flink-conf.yaml 配置文件追加下列参数开启 checkpoint，每隔 3 秒做一次 checkpoint。

execution.checkpointing.interval: 3000

使用下面的命令启动 Flink 集群

./bin/start-cluster.sh

启动成功的话，可以在 http://localhost:8081/访问到 Flink Web UI，如下所示：
在这里插入图片描述
多次执行 start-cluster.sh 可以拉起多个 TaskManager。

部署 Doris 和 MySQL 这部分省略

添加配置文件
Flink lib 目录下：
在这里插入图片描述
flinkcdc 3.3.0 lib 目录下

三、通过 Flink CDC CLI 提交任务

编写任务配置 yaml 文件。下面给出了一个整库同步的示例文件 mysql-to-doris.yaml：

################################################################################
# Description: Sync MySQL all tables to Doris
################################################################################
source:type: mysqlhostname: 172.16.11.154port: 3306username: rootpassword: xxxtables: app_db.\.*server-id: 5400-5404server-time-zone: 'Asia/Shanghai'
sink:type: dorisfenodes: 172.16.10.181:8030benodes: 172.16.10.181:8040username: rootpassword: 123456table.create.properties.light_schema_change: truetable.create.properties.replication_num: 1pipeline:name: Sync MySQL Database to Dorisparallelism: 1

其中： source 中的 tables: app_db..* 通过正则匹配同步 app_db 下的所有表。 sink 添加 table.create.properties.replication_num 参数是由于 Docker 镜像中只有一个 Doris BE 节点。

最后，通过命令行提交任务到 Flink Standalone cluster

bash bin/flink-cdc.sh mysql-to-doris.yaml

提交成功后，返回信息如：

Pipeline has been submitted to cluster.
Job ID: ae30f4580f1918bebf16752d4963dc54
Job Description: Sync MySQL Database to Doris

在 Flink Web UI，可以看到一个名为 Sync MySQL Database to Doris 的任务正在运行。
在这里插入图片描述
后续相关信息，参考官网即可
https://nightlies.apache.org/flink/flink-cdc-docs-release-3.3/zh/docs/get-started/quickstart/mysql-to-doris/