1. 准备工作
1.1 确认 MySQL binlog 模式
确认 MySQL 数据库的 binlog 模式是否为 ROW。可以在 MySQL 命令行中执行以下语句确认：
SHOW GLOBAL VARIABLES LIKE 'binlog_format';
如果返回结果中的 Value 字段为 ROW，则说明 binlog 模式为 ROW。
1.2 下载并安装 Flink
下载并安装 Flink，可以参考官方文档进行安装。
2. 配置 Flink CDC
2.1 配置 MySQL 数据库连接信息
在 Flink 的配置文件 flink-conf.yaml 中添加 MySQL 数据库连接信息，例如：
# MySQL connection configuration
mysql.server-id: 12345
mysql.hostname: localhost
mysql.port: 3306
mysql.username: root
mysql.password: 123456
mysql.database-name: test
2.2 配置 CDC Job
在 Flink 的 CDC Job 配置文件 mysql-cdc.properties 中添加以下配置：
# Flink CDC Job Configuration
name: mysql-cdc-job
flink.parallelism: 1
flink.checkpoint.interval: 60000
flink.checkpoint.mode: EXACTLY_ONCE
# MySQL CDC Source Configuration
debezium.transforms: unwrap
debezium.transforms.unwrap.type: io.debezium.transforms.ExtractNewRecordState
database.hostname: localhost
database.port: 3306
database.user: root
database.password: 123456
database.history.kafka.bootstrap.servers: localhost:9092
database.history.kafka.topic: mysql-cdc-history
database.server.id: 12345
database.server.name: test
database.whitelist: test.user
其中，name 为 CDC Job 的名称，flink.parallelism 为 Flink 的并行度，flink.checkpoint.interval 为 Flink 的 Checkpoint 时间间隔，flink.checkpoint.mode 为 Checkpoint 模式，此处设置为 EXACTLY_ONCE。
debezium.transforms 为 Debezium 转换器的名称，此处设置为 unwrap。database.hostname、database.port、database.user、database.password 分别为 MySQL 数据库的连接信息。database.history.kafka.bootstrap.servers 为 Kafka 的地址信息，database.history.kafka.topic 为 CDC 历史数据记录的 Kafka Topic。database.server.id 为 MySQL 的 Server ID，database.server.name 为 CDC Source 的名称，database.whitelist 为需要进行同步的 MySQL 表的名称。
步骤一：创建 MySQL 数据库
首先，需要在本地或云端创建 MySQL 数据库，并添加一个具有读写权限的用户。下面是一个创建名为 test_db 的数据库以及名为 flink_cdc_user 的用户的示例 SQL 代码：
CREATE DATABASE test_db;
CREATE USER 'flink_cdc_user'@'%' IDENTIFIED BY 'password';
GRANT ALL PRIVILEGES ON test_db.* TO 'flink_cdc_user'@'%';
步骤二：启动 Flink 集群
启动一个 Flink 集群以便运行 CDC 应用程序。可以使用 Flink 自带的 bin/start-cluster.sh 脚本启动 Flink 集群。确保 Flink 集群在运行时已经包含了 Kafka 和 MySQL 的依赖项。
步骤三：创建 MySQL 表和 CDC 表
在 MySQL 中，首先需要创建需要进行 CDC 的表和 CDC 表。CDC 表是一个系统表，它存储了需要捕获的更改数据。可以通过以下代码创建一个名为 test_table 的表以及与之关联的 CDC 表
CREATE TABLE test_db.test_table (
id INT PRIMARY KEY,
name VARCHAR(30),
age INT,
email VARCHAR(50)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci;
CREATE TABLE test_db.test_table_cdc (
`database` VARCHAR(100),
`table` VARCHAR(100),
`type` VARCHAR(10),
`ts` TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3) ON UPDATE CURRENT_TIMESTAMP(3),
`before` JSON,
`after` JSON
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci;
步骤四：编写 Flink CDC 应用程序
接下来，需要编写一个 Flink CDC 应用程序，以将 MySQL 表更改推送到 Kafka 主题中。可以使用 Flink 的 flink-connector-jdbc 库和 flink-connector-kafka 库来实现此目的。
以下是一个基本的 Flink CDC 应用程序的代码示例：
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);
Properties properties = new Properties();
properties.setProperty("bootstrap.servers", "localhost:9092");
properties.setProperty("group.id", "test-group");
JdbcSource<RowData> source = JdbcSource.<RowData>builder()
.setDrivername("com.mysql.jdbc.Driver")
.setDBUrl("jdbc:mysql://localhost:3306/test_db")
.setUsername("flink_cdc_user")
.setPassword("password")
.setQuery("SELECT id, name, age, email FROM test_table")
.setRowTypeInfo(Types.ROW(Types.INT, Types.STRING, Types.INT, Types.STRING))
.setFetchSize(1000)
.build();
DataStream<RowData> stream = env.addSource(source);
以下是一个简单的示例运行及结果：文章来源：https://www.toymoban.com/news/detail-467823.html
$ bin/flink run -c com.example.MyCDCJob ./my-cdc-job.jar --database.server=mysql.example.com --database.port=3306 --database.name=mydb --database.username=myuser --database.password=mypassword --table.name=mytable --debezium.plugin.name=mysql --debezium.plugin.property.version=1.3.1.Final
[INFO] Starting CDC process for table: mytable.
[INFO] Initializing CDC source...
[INFO] CDC source successfully initialized.
[INFO] Starting CDC source...
[INFO] CDC source successfully started.
[INFO] Adding CDC source to Flink job topology...
[INFO] CDC source successfully added to Flink job topology.
[INFO] Starting Flink job...
[INFO] Flink job started successfully.
[INFO] Change data for table: mytable.
[INFO] Record key: {"id": 1}, record value: {"id": 1, "name": "Alice", "age": 25}.
[INFO] Record key: {"id": 2}, record value: {"id": 2, "name": "Bob", "age": 30}.
[INFO] Record key: {"id": 3}, record value: {"id": 3, "name": "Charlie", "age": 35}.
[INFO] Change data for table: mytable.
[INFO] Record key: {"id": 1}, record value: {"id": 1, "name": "Alice", "age": 27}.
可以看到，当有数据变更时，Flink CDC Job 会输出变更的表名、记录的主键以及变更的数据。例如，在这个示例中，有一行记录的年龄字段从25变成了27。文章来源地址https://www.toymoban.com/news/detail-467823.html
到了这里，关于Flink CDC 最佳实践（以 MySQL 为例）的文章就介绍完了。如果您还想了解更多内容，请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章，希望大家以后多多支持TOY模板网！
