flink mysql cdc

flink mysqlcdc是一个用于将MySQL数据源连接到Flink的插件。它可以实时捕获MySQL数据库中的更改，并将其作为流数据进行处理和分析。引用中的代码片段展示了如何配置flink mysqlcdc作业的SQL语句。首先通过创建两个表`flink_test_order`和`flink_test_order2`来定义数据模式和字段。然后使用`mysql-cdc`连接器将`flink_test_order`表连接到MySQL数据库，并将其配置为使用指定的主机名、端口号、用户名和密码。最后，使用`print`连接器将更改的数据插入到`flink_test_order2`表中。这样，通过flink mysqlcdc，您可以实时地将MySQL数据库中的更改数据流式传输到Flink中进行进一步的处理和分析。<span class="em">1</span><span class="em">2</span><span class="em">3</span> #### 引用[.reference_title] - *1* *2* [Flink 使用之 MySQL CDC](https://blog.csdn.net/wuxintdrh/article/details/119841434)[target="_blank" data-report-click={"spm":"1018.2226.3001.9630","extra":{"utm_source":"vip_chatgpt_common_search_pc_result","utm_medium":"distribute.pc_search_result.none-task-cask-2~all~insert_cask~default-1-null.142^v93^chatsearchT3_1"}}] [.reference_item style="max-width: 50%"] - *3* [Flink mysql-cdc](https://blog.csdn.net/CarloPan/article/details/121441962)[target="_blank" data-report-click={"spm":"1018.2226.3001.9630","extra":{"utm_source":"vip_chatgpt_common_search_pc_result","utm_medium":"distribute.pc_search_result.none-task-cask-2~all~insert_cask~default-1-null.142^v93^chatsearchT3_1"}}] [.reference_item style="max-width: 50%"] [ .reference_list ]

<think>我们参考了引用[1][2][3][4]中关于Flink CDC实现MySQL数据同步的内容，并基于此整理配置指南。 核心步骤包括： 1. 环境准备：确保Flink集群版本（推荐1.18+），MySQL开启binlog，StarRocks/Doris环境可用。 2. 添加依赖：在Flink项目中引入flink-sql-connector-mysql-cdc依赖。 3. 创建CDC源表：使用Flink SQL DDL定义MySQL CDC源表。 4. 创建目标表：定义StarRocks或Doris目标表结构。 5. 执行同步任务：通过INSERT INTO语句将数据从源表同步到目标表。 下面详细说明每个步骤。</think>### Apache Flink MySQL CDC 实时数据同步配置指南 以下配置基于 Flink 1.18+ 版本，结合引用[1][2][3][4]的实践总结，实现 MySQL 到 StarRocks/Doris 的秒级同步。 --- #### **一、环境准备** 1. **基础组件** - Flink 集群（1.18.0+，[下载链接](https://flink.apache.org/downloads)） - MySQL 5.7+（需开启 `binlog` 并配置 `ROW` 模式） - StarRocks/Doris 集群（参考引用[1][2]） - Java 8+ 和 Maven 2. **MySQL 配置** 在 `my.cnf` 中添加： ```ini [mysqld] server-id=1 log_bin=mysql-bin binlog_format=ROW binlog_row_image=FULL # 必须配置 ``` --- #### **二、添加 Flink CDC 依赖** 在 Maven 项目的 `pom.xml` 中添加： ```xml <dependency> <groupId>com.ververica</groupId> <artifactId>flink-connector-mysql-cdc</artifactId> <version>2.4.0</version> <!-- 兼容 Flink 1.18 --> </dependency> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-table-planner_2.12</artifactId> <version>1.18.0</version> </dependency> ``` --- #### **三、Flink SQL 配置** 通过 SQL Client 或 Java API 定义源表和目标表（以 StarRocks 为例）： 1. **创建 MySQL CDC 源表** ```sql CREATE TABLE mysql_orders ( id INT PRIMARY KEY, order_no STRING, amount DECIMAL(10,2), update_time TIMESTAMP(3) ) WITH ( 'connector' = 'mysql-cdc', 'hostname' = 'localhost', 'port' = '3306', 'username' = 'flink_user', 'password' = 'flink_pass', 'database-name' = 'order_db', 'table-name' = 'orders', 'server-time-zone' = 'Asia/Shanghai' -- 时区设置 ); ``` 2. **创建 StarRocks 目标表** ```sql CREATE TABLE starrocks_orders ( id INT PRIMARY KEY, order_no STRING, amount DECIMAL(10,2), update_time TIMESTAMP(3) ) WITH ( 'connector' = 'starrocks', 'jdbc-url' = 'jdbc:mysql://starrocks-fe:9030', 'load-url' = 'starrocks-fe:8030', 'username' = 'admin', 'password' = 'root', 'database-name' = 'analytics_db', 'table-name' = 'orders' ); ``` 3. **启动同步任务** ```sql INSERT INTO starrocks_orders SELECT * FROM mysql_orders; ``` --- #### **四、高级配置** 1. **增量快照读取**（避免全表锁） 在源表配置中添加： ```sql 'scan.incremental.snapshot.enabled' = 'true', -- 启用增量快照 'scan.incremental.snapshot.chunk.size' = '8096' -- 分块大小 ``` 2. **处理结构变更** Flink CDC 自动捕获 `ALTER TABLE` 操作（需目标表支持 Schema 变更）[^3]。 3. **故障恢复** 启用 Flink Checkpoint（间隔 10s）： ```yaml execution.checkpointing.interval: 10s ``` --- #### **五、验证与监控** 1. **检查数据一致性** 在 StarRocks 执行： ```sql SELECT COUNT(*) FROM starrocks_orders; -- 对比 MySQL 行数 ``` 2. **监控 Binlog 延迟** 通过 Flink Web UI 的 **Source** 指标查看 `currentFetchEventTimeLag`。 --- #### **常见问题解决** - **问题：无法读取 Binlog** 检查 MySQL 用户权限： ```sql GRANT SELECT, RELOAD, SHOW DATABASES, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'flink_user'; ``` - **问题：同步延迟高** 调整 Flink 并行度： ```sql SET 'parallelism.default' = '4'; -- 根据 CPU 核数调整 ``` --- ###