Flink CDC与Paimon实现MySQL新表自动同步配置详解

1. 问题现象与背景分析

最近在使用Paimon进行MySQL整库同步时，遇到了一个让人头疼的问题：当源端MySQL数据库新增表时，Paimon端并没有自动同步这些新表。这个问题在数据仓库实时同步场景中尤为关键，因为业务系统经常会新增表来适应需求变化。

Paimon作为Apache Flink生态中的流式数据湖存储，通常与Flink CDC配合使用实现数据库实时同步。根据官方文档，Flink CDC 3.x版本确实支持Schema Evolution（表结构变更自动同步）功能，理论上应该能处理新表同步。但实际测试发现，在默认配置下，这个功能并没有如预期般工作。

2. 排查过程与根因定位

2.1 初始环境验证

首先确认了基础环境配置：

• Flink 1.17.0
• Flink CDC 2.4.0
• Paimon 0.4.0
• MySQL 8.0.32

使用的同步SQL如下：

CREATE TABLE paimon_sink WITH ( 'connector' = 'paimon', 'path' = 'file:///tmp/paimon/mysql_catalog', 'auto-create' = 'true' ) AS TABLE mysql_source /*+ OPTIONS('scan.incremental.snapshot.enabled' = 'true') */;

2.2 关键配置检查

经过反复测试，发现以下配置项对自动同步新表有决定性影响：

1. scan.incremental.snapshot.enabled：必须设为true以启用增量快照
2. scan.newly-added-table.enabled：默认false，这是问题的关键
3. paimon.auto-create：必须为true才能自动创建新表

2.3 根本原因

Flink CDC出于性能考虑，默认不会监控数据库中的新表创建事件。这是通过scan.newly-added-table.enabled参数控制的。同时，Paimon端的auto-create也需要配合启用。

3. 完整解决方案

3.1 正确配置示例

以下是经过验证可用的完整配置：

CREATE TABLE paimon_sink WITH ( 'connector' = 'paimon', 'path' = 'file:///tmp/paimon/mysql_catalog', 'auto-create' = 'true', 'table-default-options' = 'bucket=4', 'primary-key' = 'id' ) AS TABLE mysql_source /*+ OPTIONS( 'scan.incremental.snapshot.enabled' = 'true', 'scan.newly-added-table.enabled' = 'true', 'database-name' = 'your_db', 'table-name' = '.*' ) */;

3.2 各参数详解

1. scan.newly-added-table.enabled=true：启用新表检测
2. table-name='.*'：使用正则匹配所有表
3. paimon.auto-create=true：允许Paimon自动创建表
4. primary-key：必须指定，通常使用源表主键

3.3 监控与验证方法

可以通过以下方式验证新表同步是否正常工作：

-- 在MySQL端创建新表 CREATE TABLE new_table (id INT PRIMARY KEY, name VARCHAR(100)); -- 在Paimon端查询 SELECT * FROM paimon_sink.new_table;

同时可以监控Flink作业日志，搜索"New table found"关键字确认检测到新表。

4. 生产环境注意事项

4.1 权限要求

MySQL账号需要额外授予以下权限：

GRANT SELECT, RELOAD, SHOW DATABASES, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'user'@'%'; GRANT LOCK TABLES ON `your_db`.* TO 'user'@'%';

4.2 性能考量

1. 新表检测间隔由scan.incremental.snapshot.interval.ms控制（默认30s）
2. 大量表情况下建议增加scan.incremental.snapshot.chunk.size（默认8096）
3. 监控Flink作业的currentFetchEventTimeLag指标

4.3 异常处理

常见问题及解决方案：

1. 新表同步延迟：检查Flink作业是否背压，适当增加并行度
2. 权限错误：确保账号有LOCK TABLES权限
3. 主键冲突：检查Paimon表的primary-key配置是否与源表一致

5. 实现原理深度解析

5.1 Flink CDC的新表检测机制

Flink CDC通过定期执行SHOW TABLES命令并与缓存对比发现新表。检测到新表后会：

1. 获取表结构元数据
2. 创建新的数据源分片
3. 发起初始快照读取
4. 进入增量binlog跟踪阶段

5.2 Paimon的自动建表流程

当收到新表数据时：

1. 根据Flink Schema自动推断表结构
2. 应用WITH参数中的表配置
3. 在指定path下创建新的表目录
4. 初始化LSM树存储结构

5.3 端到端数据一致性保障

通过以下机制保证数据不丢失：

1. Flink CDC的checkpoint保存offset信息
2. Paimon的commit标识确保数据持久化
3. 两阶段提交协议保证原子性

6. 进阶配置建议

对于生产环境，建议增加以下配置：

WITH ( ... 'snapshot.mode' = 'initial', 'debezium.snapshot.locking.mode' = 'minimal', 'chunk-key.even-distribution.factor.upper-bound' = '1000', 'chunk-key.even-distribution.factor.lower-bound' = '0.05', 'scan.incremental.snapshot.interval.ms' = '10000', 'server-time-zone' = 'Asia/Shanghai' )

这些配置可以优化：

• 初始快照性能
• 锁争用情况
• 分片均衡性
• 时区处理

7. 监控与运维方案

建议配置以下监控指标：

1. Flink作业指标：

   • currentFetchEventTimeLag
   • sourceIdleTime
   • numRecordsIn

2. Paimon指标：

   • commitDuration
   • fileCount
   • fileSize

3. 自定义告警规则：

   • 新表检测延迟 > 5分钟
   • 快照持续时间 > 1小时
   • commit失败次数 > 0

可以通过Prometheus + Grafana搭建监控看板，关键指标示例：

SELECT table_name, COUNT(*) as file_count, SUM(file_size) as total_size FROM `paimon`.`information_schema`.`files` GROUP BY table_name;

8. 同类问题扩展

类似的同步问题还可能出现在：

1. 表结构变更不同步： 需要确保scan.incremental.schema.changes.enabled=true

2. 分库分表合并同步： 使用table-name='order_\\d+'形式正则匹配

3. 批量表初始同步： 配置snapshot.mode=initial_only先完成历史数据同步

对于这些场景，核心思路都是充分理解Flink CDC的监控机制，并合理配置相关参数。