一、方案描述

由Flink CDC来监测到源数据库数据变更并将其发送到Kafka或同步到目标数据库中，再由后续消费者或其它应用来使用数据。

二、Flink CDC

1.1 什么是CDC

CDC是Change Data Capture（变更数据获取）的简称。核心思想是，监测并捕获数据库的变动（包括数据或数据表的插入、更新以及删除等），将这些变更按发生的顺序完整记录下来，写入到消息中间件中以供其他服务进行订阅及消费。

1.2 什么是Flink CDC

官方文档地址：[项目介绍 | Apache Flink CDC](Introduction | Apache Flink CDC)
官方描述：Flink CDC 是一个基于流的数据集成工具，旨在为用户提供一套功能更加全面的编程接口（API）。 该工具使得用户能够以 YAML 配置文件的形式，优雅地定义其 ETL（Extract, Transform, Load）流程，并协助用户自动化生成定制化的 Flink 算子并且提交 Flink 作业。 Flink CDC 在任务提交过程中进行了优化，并且增加了一些高级特性，如表结构变更自动同步（Schema Evolution）、数据转换（Data Transformation）、整库同步（Full Database Synchronization）以及 精确一次（Exactly-once）语义。

1.3 其它CDC

1.4 FlinkCDC所支持的数据库情况

Flink CDC 提供了可用于 YAML 作业的 Pipeline Source 和 Sink 连接器来与外部系统交互。可以直接使用这些连接器，只需将 JAR 文件添加到您的 Flink CDC 环境中，并在 YAML Pipeline 定义中指定所需的连接器。

Flink社区开发了flink-cdc-connectors组件，这是一个可以直接从MySQL、PostgreSQL等数据库直接读取全量数据和增量变更数据的source组件（其中一些组件是基于Debezium来获取数据变更，它可以充分利用Debezium的能力）。使用这些组件可以通过Flink SQL或代码开发的方式获取目标数据库的全量数据和增量变更数据。

二、使用Pipeline连接器实时获取数据

2.1 环境介绍

我们下面将主要展示通过使用Pipeline连接器来获取实时数据的流程。
准备环境：

• 单节点的standalone模式的Flink集群
• Flink CDC
• 单节点Kafka
• Doris快速体验版数据库
• Mysql测试数据库

2.2 相关版本信息

• Flink 1.18
• Flink CDC 3.11
• Kafka 3.6.1
• Doris doris-2.0.3-rc06

2.3 详细步骤

引入所需依赖包
将 flink-cdc-pipeline-connector-doris-3.1.1.jar flink-cdc-pipeline-connector-kafka-3.1.1.jar flink-cdc-pipeline-connector-mysql-3.1.1.jar放入flink cdc的lib文件夹下

2.3.1 实时获取MySQL数据并发送到Kafka

1.编写同步变更配置文件
将yaml文件放入到flink-cdc下的job文件夹中

# 数据来源
source:
  type: mysql
  hostname: xxx.xxx.xxx.xxx
  port: 3306
  username: root
  password: "password"
  tables: doris_test.\.*
  server-id: 5400-5404
  server-time-zone: UTC+8

 # 数据去向
sink:
  type: kafka
  topic: test003
  properties.bootstrap.servers: xxx.xxx.xxx.xxx:9092
  format: json

pipeline:
  name: Sync MySQL Data to KAFKA
  parallelism: 2

2.启动Flink集群

# 在flink/bin下执行
./start-cluster.sh

3.启动Flink CDC 任务

# 在flink-cdc-3.1.1/bin下运行
./flink-cdc.sh ../job/mysql-to-kafka.yaml

启动成功

4.启动Kafka消费者

kafka-console-consumer.sh --bootstrap-server xxx.xxx.xxx.xxx:9092 --topic test003

5.在源数据库中修改数据并观察Kafka消费者
当在源数据库testfid表进行数据新增，删除或修改，Kafka消费者即能消费到对应数据

2.3.2 实时获取MySQL数据并同步到Doris数据库

1.编写同步变更配置文件

# 数据来源
source:
  type: mysql
  hostname: xxx.xxx.xxx.xxx
  port: 3306
  username: root
  password: "password"
  tables: doris_test.\.*
  server-id: 5400-5404
  server-time-zone: UTC+8

# 数据去向
sink:
  type: doris
  fenodes: xxx.xxx.xxx.xxx:8030
  username: root
  password: "password"
  table.create.properties.light_schema_change: true
  table.create.properties.replication_num: 1

pipeline:
  name: Sync MySQL Database to Doris
  parallelism: 2
     将yaml文件放入到flink-cdc下的job文件夹中

2.在Doris中创建数据库 doris_test

create database doris_test;

3.启动Flink CDC任务

# 在/app/path/flink-cdc-3.1.1/bin下执行
./flink-cdc.sh ../job/mysql-to-doris.yaml

4.进行数据变更并观察结果
先查看任务启动前源库MySQL和目标库Doris的数据情况，源库MySQL中共有两个表且表中已存在一些数据，Doris中没有表


启动任务后，两个表及数据都已同步到Doris中，当源表数据变更及表结构变更时，也都会实时同步到Doris中

5.进行路由变更后再进行测试并观察结果
Flink CDC Pipeline连接器也支持将两个同样表结构表的数据同步到目标数据库的一个表中

source:
  type: mysql
  hostname: xxx.xxx.xxx.xxx
  port: 3306
  username: root
  password: "password"
  tables: doris_test.\.*
  server-id: 5400-5404
  server-time-zone: UTC+8

sink:
  type: doris
  fenodes: xxx.xxx.xxx.xxx:8030
  username: root
  password: "password"
  table.create.properties.light_schema_change: true
  table.create.properties.replication_num: 1

# 将源库中testfid和testfid_copy1表的数据同步到目标库的表route_test中
route:
  - source-table: doris_test.testfid
    sink-table: doris_test.route_test
  - source-table: doris_test.testfid_copy1
    sink-table: doris_test.route_test

pipeline:
  name: Sync MySQL Database to Doris
  parallelism: 2

源数据

启动任务后Doris中的数据

源库中两个表的数据被合并同步到目标库的一个表中，但这只适用于相同表结构的合并，如果是不同表结构合并会造成数据错乱。