做数据同步这个事，圈子里绕不开 Flink CDC 已经很久了。标题说“十分钟掌握”，听起来有点唬人，但实操过的人都知道，核心链路确实就是几条 SQL 的事，真正花时间的反而是环境准备和参数调优。这篇文章我就把自己从零开始把 MySQL 增量数据同步到 ClickHouse 的完整过程拆开讲，包含踩过的坑、排查思路、以及最后跑通的配置，全程干货，没有废话。适合刚接触 CDC、被数据同步折磨过、或者准备在生产环境上 Flink CDC 的开发者参考。
1. 内容整体设计与思路拆解
1.1 先搞清楚 Flink CDC 到底在做什么
Flink CDC 是 Flink 生态里基于变更数据捕获（Change Data Capture）技术实现的连接器。它的核心能力是直接解析 MySQL 的 binlog，把增删改操作实时转换成 Flink 的变更流，再通过 Flink SQL 或 DataStream API 写入下游存储。
用生活化类比来理解：MySQL 的 binlog 就像收银台的购物小票，每笔交易都有记录。Flink CDC 就是那个一直坐在收银台旁边、盯着小票打印、把每笔交易念给仓库管理员听的人。仓库管理员（ClickHouse）按听到的内容更新货架。全程不需要 MySQL 主动推数据，而是 CDC 主动去“盯”。
这个机制带来的最大好处是：对业务库几乎零侵入。不需要在业务代码里写双写逻辑，不需要改表结构，只要 MySQL 开了 binlog，给一个只读账号，剩下的就是 Flink 的活。
1.2 为什么是 Flink CDC，而不是 Canal + Kafka
很多人第一反应是用 Canal 监听 binlog 发到 Kafka，再由 Flink 消费写入 ClickHouse。这条路完全走得通，但引入的组件太多，维护成本高。对比一下就清楚了：
| 方案 | 组件数量 | 全量+增量一体 | 断点续传 | 数据转换能力 | 维护成本 |
|---|---|---|---|---|---|
| Canal + Kafka + Flink | 3 套系统 | 需要手动拼全量+增量 | 依赖 Kafka offset | 需要写 UDF 或额外处理 | 高 |
| Flink CDC 直连 | 1 套系统 | 天然支持 | 依赖 checkpoint | Flink SQL 直接处理 | 低 |
Flink CDC 最突出的优势是“全量+增量一体化”。它启动时会先做一次全量快照，快照结束后不中断，自动切换到增量模式继续消费 binlog。这个切换过程对使用者完全透明，你不需要关心数据从哪来的，只需要定义好 source 和 sink。
另外，Flink CDC 2.x 版本引入了增量快照框架，支持并行读取全量数据，处理千万级大表比老版本快很多，这个后面会细说。
1.3 整体链路设计：全量+增量一体化的思路
我最终跑通的架构链路是这样的：
MySQL binlog → Flink CDC connector → Flink 任务（checkpoint 机制）→ ClickHouse 表关键设计点有三个：
第一，全量与增量统一在同一个 Flink 任务里。任务启动后 Flink CDC 先扫描 MySQL 当前的全量数据，等到扫描完成再无缝切换到 binlog 增量解析。因为状态存在 Flink 的 checkpoint 里，所以即使任务重启，也能从上次消费的位置继续，不会丢数据。
第二，ClickHouse 这边的表引擎选择决定了数据一致性。MySQL 的 update 操作在 CDC 流里表现为先删除后插入或者直接更新，ClickHouse 不原生支持单行更新，所以目标表必须选 ReplacingMergeTree，靠版本号或时间戳来保证最终一致性。
第三，整条链路的延迟在秒级，Flink 默认的 checkpoint 间隔和 source 的 poll 频率决定了实时性。实测下来，MySQL 那边一条 update 提交，到 ClickHouse 可查询，延迟大约 2 到 3 秒，满足绝大多数分析场景。
2. 核心细节解析与实操要点
2.1 MySQL 端三个前置条件，一个都不能少
很多人任务跑不起来，问题都出在 MySQL 端。三个前置条件必须全部满足：
第一个是开启 binlog。MySQL 配置文件里需要设置：
[mysqld] server-id=1 log_bin=mysql-bin binlog_format=ROW binlog_row_image=FULL expire_logs_days=7其中binlog_format=ROW是硬性要求。Flink CDC 需要解析每一行变更前后的值，只有 ROW 格式的记录里才包含完整的行数据。STATEMENT 格式只记录 SQL 语句，无法还原数据内容。
第二个是创建专用的同步账号，开发环境用 root 图省事可以，生产环境千万不能这么干：
CREATE USER 'cdc_user'@'%' IDENTIFIED BY 'your_password'; GRANT SELECT, RELOAD, SHOW DATABASES, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'cdc_user'@'%'; FLUSH PRIVILEGES;注意REPLICATION SLAVE和REPLICATION CLIENT这两个权限必须有。Flink CDC 内部是模拟一个从库去主库拉取 binlog，没有复制权限会被 MySQL 拒绝。
第三个容易被忽略的点：如果同一个 MySQL 实例会被多个 Flink CDC 任务监听，每个任务的 source 都要配置不同的server-id，否则会互相干扰。因为 Flink CDC 内部实现依赖 MySQL 的主从复制协议，相同 server-id 的连接会被 MySQL 错乱处理。
2.2 ClickHouse 端建表与引擎选择的讲究
ClickHouse 不是传统 OLTP 数据库，它的核心竞争力在列式存储和分析查询。把 MySQL 数据同步过来，首先要接受一个现实：ClickHouse 对单行更新的支持很弱。因此目标表一般选用 ReplacingMergeTree 系列引擎，用版本号来标记新数据，合并时保留最新版本。
我实际使用的建表语句：
CREATE TABLE default.orders_sync ( id UInt64, order_no String, customer_name String, amount Decimal(18, 2), status Int8, create_time DateTime64(3), update_time DateTime64(3), _version UInt64 ) ENGINE = ReplacingMergeTree(_version) PARTITION BY toYYYYMMDD(create_time) ORDER BY id_version这个字段是手动加的，用来承载 Flink CDC 同步过来的数据版本。我的做法是用 MySQL 侧的update_time转成时间戳，配合 MySQL binlog 里的自增序号来保证同一个主键下，后写入的就是较新的数据。
如果不同步带版本号的字段，直接用 ReplacingMergeTree 也不是不行，但如果 MySQL 侧短时间内对同一条记录做了多次 update，而分区合并恰好只保留其中一个版本，就会出现数据不是最终值的情况。这是实战中最容易踩的坑，后面常见问题里还会提到。
2.3 Flink 环境与连接器版本匹配
版本匹配问题是 Flink CDC 项目里最折磨人的环节。我用的是 Flink 1.17.2 + Flink CDC 2.4.x，配套的 MySQL 连接器是内置的，不需要单独下载 jar，直接依赖 flink-sql-connector-mysql-cdc 即可。
如果你用的是 Flink 1.14 或更早版本，就必须用对应的 CDC 1.x 版本，两者的 SQL 语法和 source 定义方式有差异，不能直接套用。建议新项目直接上 Flink 1.17 以上的版本，CDC 2.4 的增量快照框架成熟很多，并行度配置也灵活。
另外，要注意 ClickHouse 连接器目前没有官方维护的 Flink SQL connector，社区版是flink-connector-clickhouse。我使用的是 2.4.0 版本，基本稳定，但写 ClickHouse 时如果遇到格式问题，优先级最高的手段是先把流打印出来排查，而不是猜。
3. 实操过程与核心环节实现
3.1 环境准备清单
先看版本，不完全固定，但建议参照这套已验证过的组合：
| 组件 | 版本 |
|---|---|
| Flink | 1.17.2（Standalone 模式） |
| Flink CDC | 2.4.0 |
| MySQL | 8.0.28 |
| ClickHouse | 23.8 |
| JDK | 1.8 或 11 |
Flink 我用的 Standalone 集群，生产环境可以考虑 YARN 或 K8s，但模式不影响下面的 SQL 写法。
启动 Flink 集群以后，需要把 ClickHouse 连接器的 jar 包放到 Flink 的 lib 目录，然后重启集群。这一步说多了都是泪，不重启的话 Flink SQL 客户端加载不到新的 jar，任务提交会直接报ClassNotFoundException。
3.2 MySQL 开启 binlog 并创建同步账号
MySQL 开启 binlog 的操作已经在上文写到了。这里补充一个验证操作：
SHOW VARIABLES LIKE 'log_bin'; SHOW VARIABLES LIKE 'binlog_format';确认log_bin是 ON，binlog_format是 ROW。然后执行一条修改操作，再执行：
SHOW BINLOG EVENTS IN 'mysql-bin.000001' LIMIT 5;能看到Write_rows或Update_rows事件，说明 binlog 是正常记录的，CDC 就有数据可读。
创建账号后记得授权并验证：
mysql -ucdc_user -p -h127.0.0.1 -e "SELECT 1"如果带mysql_native_password报错，需要在 MySQL 8.0 里给用户设置兼容的加密方式：
ALTER USER 'cdc_user'@'%' IDENTIFIED WITH mysql_native_password BY 'your_password';这个坑在 MySQL 8.0 默认 caching_sha2_password 下很常见。
3.3 ClickHouse 里建好目标表
前面已经给出了建表 SQL。这里补充一个注意点：字段类型的映射关系。MySQL 的bigint对应 ClickHouse 的Int64或UInt64，MySQL 的decimal(10,2)对应Decimal(10,2)，MySQL 的datetime对应DateTime。但 MySQL 的datetime(3)如果有毫秒精度，ClickHouse 侧就要用DateTime64(3)，否则毫秒部分会被截断，导致数据对不上。
另外，MySQL 的字段如果允许NULL，ClickHouse 侧也要用Nullable(...)包裹，否则任务运行时遇到 NULL 值会直接报错。ClickHouse 默认字段是非 NULL 的，这一点和 MySQL 的习惯很不一样。
3.4 用 Flink SQL 把任务跑起来
核心 SQL 三件事：建 source 表、建 sink 表、提交插入任务。
先建 source 表：
CREATE TABLE orders_source ( id BIGINT, order_no STRING, customer_name STRING, amount DECIMAL(18, 2), status INT, create_time TIMESTAMP(3), update_time TIMESTAMP(3), PRIMARY KEY (id) NOT ENFORCED ) WITH ( 'connector' = 'mysql-cdc', 'hostname' = '127.0.0.1', 'port' = '3306', 'username' = 'cdc_user', 'password' = 'your_password', 'database-list' = 'test_db', 'table-list' = 'orders', 'server-id' = '5401-5404', 'scan.incremental.snapshot.enabled' = 'true', 'scan.startup.mode' = 'initial' );scan.startup.mode有三个值：initial表示先做全量再增量，earliest-offset表示从最早的 binlog 位置开始，latest-offset表示只消费任务启动后新产生的变更。需要跑存量数据就用initial，只关心增量就选latest-offset。
server-id配置成了5401-5404，这是给并行度用的。并行度是 4 的话，每个 subtask 各占一个 server-id，避免冲突。
再建 sink 表：
CREATE TABLE orders_sink ( id BIGINT, order_no STRING, customer_name STRING, amount DECIMAL(18, 2), status INT, create_time TIMESTAMP(3), update_time TIMESTAMP(3), _version BIGINT ) WITH ( 'connector' = 'clickhouse', 'url' = 'clickhouse://127.0.0.1:8123', 'username' = 'default', 'password' = '', 'database-name' = 'default', 'table-name' = 'orders_sync', 'sink.batch-size' = '1000', 'sink.flush-interval' = '1000', 'sink.max-retries' = '3' );sink.batch-size和sink.flush-interval是写入吞吐和实时性的权衡。batch-size 越大，ClickHouse 写入越高效，但数据落库的延迟越高。
最后提交任务：
INSERT INTO orders_sink SELECT id, order_no, customer_name, amount, status, create_time, update_time, UNIX_TIMESTAMP(update_time) * 1000 AS _version FROM orders_source;这里_version用的是update_time的时间戳毫秒值，用于 ReplacingMergeTree 的版本比较。
3.5 验证同步效果与数据一致性
任务提交成功以后，先在 MySQL 侧插入一条测试数据：
INSERT INTO test_db.orders(order_no, customer_name, amount, status, create_time, update_time) VALUES ('TEST001', '张三', 99.90, 1, NOW(), NOW());然后在 ClickHouse 查询：
SELECT * FROM default.orders_sync WHERE order_no = 'TEST001';正常情况下几秒内就能查到。再执行一次 update：
UPDATE test_db.orders SET status = 2, update_time = NOW() WHERE order_no = 'TEST001';ClickHouse 里再查一次，status 应该变成 2。因为 ReplacingMergeTree 是异步合并，可能短时间内查到两条记录，用SELECT ... FINAL能强制合并读取到最终状态。
顺便验证删除操作：MySQL 里 delete 一条记录，默认情况下 ClickHouse 里不会自动删除对应行。这是 Flink CDC 同步到 ClickHouse 最需要注意的行为差异，下面常见问题里细说。
4. 常见问题与排查技巧实录
4.1 任务启动报权限和连接问题
报错Access denied for user或者Connection refused是最常见的。连接问题用 MySQL 客户端先验证账号能不能远程连上，排除网络和安全组因素。权限问题重点检查REPLICATION SLAVE和REPLICATION CLIENT两个权限是否授予成功。
还有一种情况是 MySQL 8.0 默认认证插件不兼容，需要在创建用户时显式指定：
CREATE USER 'cdc_user'@'%' IDENTIFIED WITH mysql_native_password BY 'your_password';4.2 时间字段差 8 小时，时区坑
同步过来的create_time比 MySQL 里的值少 8 小时，这是典型的时区问题。MySQL 的 server 时区、JDBC 连接时区、Flink 的local-time-zone三者不一致就会产生偏移。
解决办法有两个。一是在 Flink 提交任务前设置会话时区：
SET 'table.local-time-zone' = 'Asia/Shanghai';二是在 source 表连接参数里加：
'server-time-zone' = 'Asia/Shanghai'两个都配上更保险。这个坑在跨时区部署的集群上尤其隐蔽，本地测不出来，到了服务器上才发现。
4.3 重复数据处理，ReplacingMergeTree 不是银弹
前面提到过，同一主键的数据可能同时存在多版本，ReplacingMergeTree 只在后台合并时才真正去重。如果在合并之前查询，不强制加 FINAL 就会看到重复数据。
解决思路有两个角度。一是查询时加 FINAL：
SELECT * FROM default.orders_sync FINAL WHERE order_no = 'TEST001';二是从源头优化，让写入的数据尽量唯一。比如用 CDC 的op字段做过滤，只写入更新后的最终状态。不过最简单的还是接受 ReplacingMergeTree 异步合并的特性，查询时统一走 FINAL 或者加去重逻辑。
如果业务对精确去重要求特别高，可以考虑在 Flink 侧做 KeyedProcessFunction 按主键去重后再写入，但会牺牲一部分吞吐。
4.4 DDL 变更和表结构不一致
Flink CDC 默认行为是：MySQL 侧新增一列后，sink 侧如果没有对应字段，任务会直接报错。这是因为 source 解析 schema 变化后，生成的 RowData 结构变了，而 sink 端表结构没同步更新。
这个问题的正规解法是同步执行 ClickHouse 的 ALTER TABLE，把列补齐。生产环境建议先把 DDL 变更的流程纳入变更管理，不要直接在业务库执行 DDL，否则 Flink 任务随时可能挂。
有些团队用 Flink CDC 的schema-change策略自动把 DDL 转发到下游，但 ClickHouse 连接器对 DDL 同步的支持并不完整，目前最稳妥的做法还是人工介入。
4.5 大表初始化性能优化
全量阶段如果是千万级大表，直接默认配置跑会很慢，甚至出现 checkpoint 超时。2.x 的增量快照框架支持并行读取，我实践下来的调优手段基本这三板斧：
- 把
scan.incremental.snapshot.chunk.size调小，默认 8096 行，可以改成 4096，小 chunk 能让 checkpoint 更频繁地完成，降低超时风险。 - 调大 source 的并行度，注意 server-id 范围也要对应扩大，比如并行度 4 就配 4 个 server-id。
- 给 Flink 任务分配更多内存，特别是 taskmanager 的堆外内存，防止大表快照阶段 GC 压力过大导致任务卡住。
4.6 存量历史数据处理策略
任务启动时如果已经选了initial模式，全量数据会自动同步，但业务上有时只想同步最近 N 天的数据，不想把整张历史表全部搬过去。
常见做法是先在 MySQL 侧做视图或临时表，只取需要的数据范围，然后 Flink CDC 监听这个视图。但注意 MySQL 的 binlog 不会记录视图的变更，视图在 Flink CDC 里只能做一次快照，后续增量数据不会自动更新视图，所以这种方式只适合一次性全量初始化。
另一种更通用的思路：先在 MySQL 建一张新表，把需要的数据 INSERT INTO 进去，然后 CDC 直接同步这张表。这张表只承载同步任务，业务上定时刷新即可。这样既控制了数据范围，又保留了增量能力。
实操总结
Flink CDC 同步 MySQL 到 ClickHouse 的核心链路不复杂，一旦跑通以后，维护成本很低。我踩过的坑主要集中在三块：MySQL 端权限和 binlog 配置、ClickHouse 的版本去重与类型映射、Flink 的 checkpoint 和连接器版本匹配。
个人建议是先用数据量小的表跑通全流程，确认同步延迟和去重逻辑都符合预期，再扩展到核心大表。生产环境强烈建议把全量初始化阶段放到业务低峰期，同时给每个同步任务都配上监控告警，Flink 任务挂掉不可怕，可怕的是挂了你不知道，等业务方来投诉数据对不上。最后再分享一个小技巧：给 source 表和 sink 表的字段都加上WITH注解写清楚注释，团队协作时能省很多沟通成本，这个细节常规文档不会告诉你。
