1. 方案选型：为什么最终选了 Flink CDC

1.1 我对比过的几个同步方案

前两年我们这边业务发展得很快，订单、库存、用户积分这些核心表的数据量开始上来，下游数仓和搜索团队隔三差五就来找我们：“能不能把 MySQL 里的数据实时同步一份过来？”一开始我们用的是凌晨批量跑任务的方式，把数据从库里面捞出来同步到数仓。这个方案最直接的问题就是时效性差，业务方早上想看前一天的数据可以接受，但后来运营要做实时大屏，搜索要更新索引，消息通知要基于用户实时行为触发，传统 T+1 的方式已经完全顶不住了。

于是我开始调研市面上能做的同步方案，当时主要看了这么几个：

• Canal 加自研消费端：Canal 伪装成 MySQL 从节点拉取 binlog，把变更解析成 JSON 放到消息队列里面，再自己写消费逻辑。思路是通的，但是自研的代价很大，你既要处理 binlog 位点管理，又要自己做全量数据和增量数据的衔接，还得分批写入下游，能跑起来是一回事，跑得稳是另一回事。
• Debezium 加 Kafka Connect：Debezium 本身是 CDC 领域的老牌框架，捕获能力很强，配合 Kafka Connect 可以很方便地把变更流发布到 Kafka。但它更偏向 Kafka 生态，如果你的下游是 Kafka，这套组合很合适。可我们当时一部分下游是 Doris 和 Elasticsearch，中间还是要写不少适配逻辑。
• DataX 这类离线同步工具：适合做批量全量，但做不到变更捕获，实时性为零。
• Flink CDC：引入连接器之后，在 Flink SQL 里直接建源表，一条 SQL 定义同步任务，下游可以是 Kafka、Doris、StarRocks、Elasticsearch 等等，最关键的是全量加增量的一体化能力是内置的，不需要你手工去维护分界线。

最后我们选的是 Flink CDC。

我聊一下当时的决策逻辑。我们的团队里既有写 Java 的，也有写 Scala 的，还有专门维护数仓的，大家都会基础的 SQL。如果选 Canal 加自研的路线，研发和运维成本都会压在少数几个人身上，后面一旦这几个核心的人忙不过来，整个同步链路就没人维护了。而 Flink CDC 的入口是 Flink SQL，意味着只要会写 SQL 的人都能上手，能降低团队协作的门槛。另外，Flink 本身是流式计算引擎，同步链路中间不只是“数据搬移”，还能顺手做数据清洗、打宽表、分流、维表关联，这些都是单纯同步工具做不了或者做起来很费劲的事情。

1.2 Flink CDC 的两大核心竞争力

第一点，也是排在最前面的，就是全量增量一体化。以前做数据同步的常规操作是：先用离线工具做一次全量导入，然后记录一个时间点位，再从那个点位起消费增量日志，两者之间要保证数据不重不漏是很头疼的。Flink CDC 的执行模式会把全量扫描和增量读取融合在一个作业里，它内部先做一个一致性快照，把存量数据读完，然后自动切到 binlog 的增量位点继续消费。这个过程你不需要关心具体切分点在哪里，作业自己管理。

第二点是分布式并行读取。老版本的 CDC 同步工具，大部分只能单并发读 binlog（毕竟 binlog 本质是单条流），一旦大表全量阶段就特别慢。Flink CDC 对存量数据做了分片，可以多个并发同时扫不同的数据段，把一张千万级的大表全量阶段从几个小时压缩到十几分钟。这种体验在老方案里是很难实现的。

除此之外，Flink CDC 天然带出 checkpoint 机制，配合下游事务型 sink，可以做到端到端的精确一次语义。这点在数据正确性要求很高的场景下非常关键，比如订单金额汇总、库存扣减同步，数据少了或者多了都会出问题。

不过，这里我要提醒一句：选型不能只看技术亮点，还要看自己团队的实际情况。如果你的下游只是几个固定系统，数据量也不大，业务时效要求不超过小时级，那完全没必要上这套实时链路，凌晨批处理足够了。实时代表的是成本：运行资源、运维复杂度、出问题时候的排查难度，都会上一个台阶。

2. 核心机制拆解：CDC 到底是怎么工作的

2.1 从 binlog 到变更日志：四类数据的流转

很多人在写 Flink CDC 任务的时候，容易把它当成一个“连接数据库的 JDBC 插件”来理解，其实不是。Flink CDC 读取的是 MySQL 的 binlog，不是什么 SQL 查询，它监听的是数据库的每一次提交变更，然后把变更翻译成 Flink 内部的标准格式。

MySQL 开启 binlog 之后，会记录所有数据变更操作。Flink CDC 连接器内部封装了 binlog 的解析逻辑，把变更事件转成 Flink Table 的 RowData。这个 RowData 有四种类型，对应的就是数据库里的增删改：

• +I表示插入，对应数据库的 insert；
• -U表示更新前的旧值；
• +U表示更新后的新值；
• -D表示删除，对应数据库的 delete。

这个模型是理解 Flink CDC 的关键。你在 Flink SQL 里建一张 CDC 源表，它本质不是一张“表”，而是一条“变更日志流”，流里面的每一条数据都带有上述的标记。下游算子看到+I就做插入，看到-U就用新值替换旧值，看到-D就删除。正是这种统一的 changelog 标准，让 Flink CDC 能和不同的下游系统对接，因为每种下游系统都把 Flink 的这四种事件类型映射到自己的写入语义里面。

举个实际例子，如果你要把 MySQL 的订单表同步到 Doris 里，Doris 的 sink 连接器会按照事件类型生成对应的写入请求：+I对应 insert，-U对应 update，-D对应 delete。这样听起来很简单，但当一个作业跑大了以后，比如更新事件每秒上千条、单条记录字段有几十个，对这种变更日志的传递有序性要求很高。更新操作不能乱序，否则最终结果就会对不上。Flink CDC 能保证同一个主键维度的更新事件是分到同一个下游算子实例处理的，从而保持了主键级别的有序性。

2.2 增量快照机制：全量加增量无缝衔接的关键

老版本的 Flink CDC 在做全量阶段，会被迫对整张表持有全局读锁，否则全量还没扫完，数据已经被改了，增量位点就不知道从哪里开始追。这在 MySQL 5.7 等老版本上很明显，线上大表根本不敢直接用，因为一把锁可能把整个业务拖死。

新版本的机制做了很大改进，核心思路叫做增量快照算法。我尽量用大白话讲清楚：它先把一个表的数据按主键范围切成很多个 chunk（数据块），然后多个并发线程同时去扫不同的 chunk。扫描过程中，记录 binlog 的当前位点，扫完所有 chunk 之后，再从那个位点开始拉增量的 binlog，把这些增量事件和之前扫到的快照数据按主键做合并，确保数据不丢失也不重复。

关键点在于，每一个 chunk 扫描完成之后，还会再回补一遍这个 chunk 范围内这段时间产生的增量数据，这相当于做了一次“补漏”。当所有 chunk 都完成“扫全量+补增量”之后，整个表就自然推进到了纯增量消费的状态。整个过程虽然恢复到了旧版本单并发时的逻辑复杂度，但是不同 chunk 之间是并行处理的，所以整体速度非常快。我们用一张 2000 万行的订单明细表做过测试，在 4 并发的情况下，全量加增量衔接完成大概 8 分钟左右，旧方案接近 40 分钟。

这个机制的启用有一些前提条件，我在实操部分再细说。这里先记着一个结论：做 Flink CDC 同步任务时，源表一定要有主键或者唯一键。没有主键的表无法做 chunk 切分，Flink CDC 就只能退化成单并发读取，全量阶段性能断崖式下跌。

再补充一个大家经常忽略的细节：增量快照依赖 MySQL 的 binlog 格式。必须设置binlog_format = ROW，binlog_row_image = FULL。如果你的库是混用格式或者不是 FULL，解析出来的数据就会缺字段，甚至报错。这也是我们后来排查线上问题时的经验教训，后面常见问题部分我再展开。

3. 实操落地：我用的配置与关键参数

3.1 环境准备与前置条件

这一节分享一下我们实际落地时用的版本组合和环境参数。我这里说一套经过验证的配置，基于 Flink 1.17 和对应的 CDC 连接器版本，不同版本之间 API 和参数名有一些差异，但大体思路是一致的。

第一件事，MySQL 侧要开启 binlog。修改 MySQL 配置文件，比如my.cnf，加入以下核心配置：

server-id = 223344 log_bin = mysql-bin binlog_format = ROW binlog_row_image = FULL expire_logs_days = 7

这里server-id是要单独强调的。Flink CDC 源表定义的 WITH 参数里也有server-id配置，它和 MySQL 实例的server-id是两个不同概念。Flink CDC 的server-id是用于让 Flink 客户端伪装成一个 MySQL 从节点去拉 binlog，这个值不能和集群里其他从节点冲突，否则会被 MySQL 踢掉。我习惯的做法是给每个 CDC 作业分配一段范围内的 server-id，比如5400-5404，这样多并发时每个并发都能拿到一个独立 ID。

第二件事是给同步任务创建独立的 MySQL 账号，只授予权限SELECT, RELOAD, SHOW DATABASES, REPLICATION SLAVE, REPLICATION CLIENT即可，并且限制这个账号只能访问需要同步的库。这里有一个安全考量：不要拿业务的超级账号直接给 Flink CDC 用，一旦连接器出现异常，短时间内建立大量连接，可能把线上数据库的连接池打爆。独立账号配合最小权限，即使出问题，影响面也可控。

第三件事，准备一个可用的 Flink 集群或者本地开发环境。如果是本地测试，单机模式就够了；如果用于生产，我建议至少两个 TaskManager 起步。我们在生产环境用了 3 个 TaskManager，每个分配 4 个 slot，因为一个同步集群下游往往有多个同步任务，slot 数量决定了最大并行度上限。

3.2 核心 SQL 与任务配置

我用一个真实的同步场景来解释核心 SQL 的写法。假设我们需要把 MySQL 里业务库的orders表实时同步到 Doris，同时做一个简单过滤（只同步状态为成功和关闭的订单），SQL 大致长这样：

CREATE TABLE orders_mysql ( id BIGINT, order_no VARCHAR(32), user_id BIGINT, amount DECIMAL(12, 2), status INT, create_time TIMESTAMP(3), update_time TIMESTAMP(3), PRIMARY KEY (id) NOT ENFORCED ) WITH ( 'connector' = 'mysql-cdc', 'hostname' = '192.168.1.10', 'port' = '3306', 'username' = 'cdc_user', 'password' = 'your_password', 'database-name' = 'business_db', 'table-name' = 'orders', 'server-id' = '5400-5404', 'scan.startup.mode' = 'initial' ); CREATE TABLE orders_doris ( id BIGINT, order_no VARCHAR(32), user_id BIGINT, amount DECIMAL(12, 2), status INT, create_time TIMESTAMP(3), update_time TIMESTAMP(3) ) WITH ( 'connector' = 'doris', 'fenodes' = '192.168.1.20:8030', 'table.identifier' = 'ods_db.orders', 'username' = 'doris_user', 'password' = 'doris_password', 'sink.label-prefix' = 'cdc_orders_' ); INSERT INTO orders_doris SELECT id, order_no, user_id, amount, status, create_time, update_time FROM orders_mysql WHERE status IN (1, 2);

这里有几个参数我想展开讲讲。

scan.startup.mode有三个常用值：initial表示作业启动时先做全量扫描再衔接增量；latest-offset表示只同步作业启动后新产生的数据，历史数据不管；timestamp表示从指定时间点开始消费。首次跑一个表，一定要用initial，它能把存量数据补齐，后续重启作业才考虑latest-offset或者基于保存点恢复。

sink.label-prefix是 Doris 幂等写入的标签前缀。Flink CDC 默认和 checkpoint 绑定，每完成一个 checkpoint 提交一次事务，Doris sink 会用自己的事务机制保证一批一批写入不重复。标签前缀相当于给每批事务打了一个唯一标识，如果作业重启，Doris 能识别出当前这批事务是否已经提交过了，避免重复写入。

这里还要补充一个非常容易被忽略的事情：源表字段和下游表字段的映射要保持一致，特别是主键字段的声明。在 CDC 源表 DDL 里，PRIMARY KEY (id) NOT ENFORCED这句并不是真的去 MySQL 建立主键约束，而是告诉 Flink 这张 changelog 流的“主键”是什么，用于保证更新事件的有序性以及下游 upsert 语义。如果没有声明主键，下游有些系统（比如 Doris 的 Unique 模型）无法正确执行更新和删除。

3.3 参数调优与资源规划

代码能跑通只是入门，生产环境还要看你敢不敢把并行度调到 8 甚至 16。分享一下我在实践中的配置经验。

Flink SQL 作业中，源表并行度并不是直接在 DDL 里配置的，而是通过作业级别的并行度参数或者Source的并行度 hint 来控制。MySQL CDC 源默认并行度是 1，即使你的作业并行度设置了 8，读取 binlog 的增量阶段也只能有一个并发在跑。这倒不是因为 Flink 限制，而是 binlog 本身就是单线程的字节流，多个并发去消费同一个 binlog 流是无法保证顺序的。所以，增量的实时同步瓶颈往往是在下游写入和序列化环节，而不是在源读取环节。

全量阶段就不一样了，增量快照算法会把表按主键切成多个 chunk，这时候高并行度能显著加速存量数据读取。我一般这样设置：如果一个表有 500 万行以下，并行度 2 就够了；500 万到 5000 万行，并行度 4 到 8；超过 5000 万行，可以调到 16 甚至更高，同时配合调整scan.incremental.snapshot.chunk.size。

scan.incremental.snapshot.chunk.size这个参数决定每个 chunk 包含多少行数据，默认是 8096。如果单行字段很宽（比如几十个字段，还有长文本），建议把 chunk size 调小一点，比如 4096，避免单个 chunk 在内存中积压太多数据导致 GC 压力。如果表行数非常大、字段又不宽，可以适当调大 chunk size 到 16000 左右，减少 chunk 数量，降低调度开销。

内存方面，Flink CDC 作业的内存使用大头在状态后端。增量同步的状态主要是 binlog 位点、schema 信息，以及正在处理中的变更数据。我建议状态后端使用增量 RocksDB，配置类似：

state.backend: rocksdb state.checkpoint-storage: filesystem state.checkpoints.dir: hdfs:///flink/checkpoints execution.checkpointing.interval: 30s execution.checkpointing.min-pause: 10s execution.checkpointing.tolerable-failed-checkpoints: 2

checkpoint 间隔设置 30 秒是一个比较平衡的值。如果你把间隔压到 10 秒、5 秒，数据延迟会降低，但是对上游 MySQL 的压力、对下游的写入压力都会增加。对于大多数业务同步场景，30 秒的延迟完全够用，真到了秒级同步需求，你会先遇到成本瓶颈而不是配置瓶颈。

还有一点就是重启策略。生产上绝对不能因为一次反序列化异常就让整个作业挂掉。我会在提交 SQL 的时候显式指定重启策略：

SET 'restart-strategy' = 'fixed-delay'; SET 'restart-strategy.fixed-delay.attempts' = '10'; SET 'restart-strategy.fixed-delay.delay' = '60s';

遇到脏数据或者下游临时不可用，Flink 会先重启几次，如果连续失败超过 10 次才会把作业标记为失败，触发告警。这里的 10 次和 60 秒不是拍脑袋定的，60 秒的延迟可以给下游系统一些恢复时间，如果 10 分钟之内都没缓过来，作业再挂着也没意义，不如让它失败，让值班的人介入。

4. 常见问题与排查：实测中踩过的坑

4.1 全量阶段慢、卡住甚至失败

这个问题基本是每个第一次在生产落地 Flink CDC 的人都会遇到的。表现是：全量同步阶段执行到一半，进度条不动了，或者长时间没有数据写入下游，再严重点直接 OOM 挂掉。

先检查有没有主键。我之前调过一张没有主键的大表，不管并行度调配到多少，全量阶段都是单并发慢如蜗牛。解决方案是评估这张表能否在满足业务语义的前提下添加唯一键。如果表本身确实没有业务主键，另一条路是拆表：把一个大表按业务分区字段拆成多个小任务同步，至少可以让不同分区并行。

接着检查 chunk size 和内存配置。默认 8096 的行数，如果每行数据特别宽，内存压力就会非常大。我遇到过一次每行 30 多个字段、还有几个TEXT类型的表，全量阶段 TaskManager 频繁 Full GC，最后通过把 chunk size 调整为 3000 左右解决。

还有一种容易忽略的情况：MySQL 实例的max_allowed_packet设置过小。binlog 里单条变更事件超过该值后，Flink 拉取 binlog 会报错或者截断。建议把 MySQL 的max_allowed_packet调到 64MB 以上，特别是在有 JSON、长文本字段的场景下。

4.2 时区问题导致数据差 8 小时

这个坑我印象太深了。当时同步一个交易流水表，下游数仓反映时间字段和 MySQL 里的原始值对不上，总是相差 8 个小时。

原因很简单：Flink CDC 在解析 MySQL 的TIMESTAMP和DATETIME类型时，默认使用的是 JVM 时区。如果你的 Flink 集群时区是 UTC，而 MySQL 连接时区是东八区，字段值就会被转换一遍。解决方法是统一时区配置，包括三个层面：

• Flink 的env.java.opts里增加-Duser.timezone=Asia/Shanghai；
• Flink SQL 的 table 配置里设置table.local-time-zone为Asia/Shanghai；
• 建表语句中时间字段的类型要匹配，比如TIMESTAMP(3)这种写法，不要简写成TIMESTAMP。

另外提一个更隐蔽的点：TIMESTAMP和TIMESTAMP_LTZ在 Flink CDC 链接 MySQL 时的语义是不一样的。TIMESTAMP受时区影响，TIMESTAMP_LTZ是一个绝对时间。如果下游需要按业务本地时间展示，建议源表声明为TIMESTAMP，并在 Flink SQL 里加上CONVERT_TZ或者确保所有环节时区统一。我们最终统一了所有环境的时区为东八区，之后再也没有出现过时间漂移问题。

4.3 DDL 变更导致作业失败或数据错乱

这是 Flink CDC 在使用中最麻烦的一个问题，比时区问题麻烦得多。业务方上线新功能，给订单表加了一个字段，如果你的同步作业正好在这期间运行，很可能出现 schema 不匹配的报错，作业直接失败。

Flink CDC 对新增字段的处理策略取决于连接器版本。新一点的版本会自动兼容新增字段，把新字段同步到下游。但前提是，下游表也是同样的 schema，如果你的 Doris 表和 MySQL 表字段不一致，同步过去要么报错要么漏字段。

我的做法是建立一套字段变更规范：业务侧的 DDL 变更提前在群里告知数据团队，数据团队先评估下游表结构是否需要同步调整，再允许业务改动源表。听起来多了一个沟通环节，但实际落地下来非常省心。因为实时同步链路一旦因为字段变更挂了，恢复流程才是真正的麻烦：要么重新做全量初始化，要么手工定位 binlog 位点，这两种做法的恢复时间都是以小时计的。

另外，如果用的是latest-offset模式启动的作业，DDL 变更带来的影响会直接体现在解析 binlog 阶段，报错信息通常是“无法找到某个字段”或者“字段类型不匹配”。排查时先确认 binlog 里记录的原始事件，找到变更前后的 schema 记录，就能定位是哪条 DDL 触发的异常。

4.4 数据乱序、重复与精确一次

有些场景下，下游看到的数据出现短暂的回退，比如一条订单先看到状态是“已完成”，转头又变成“处理中”，过几秒又变回“已完成”。这种情况通常不是 Flink CDC 本身的问题，而是源表没有主键，或者下游写入模型不支持更新语义。

我之前遇到过一张没有主键的日志表，Flink CDC 的输出流里没有主键定义，Doris 侧使用的是 Duplicate 模型，结果每次都把变更数据当成新数据插入，最终产生了大量重复记录。处理方式也很直接：在 Flink SQL 里对日志表声明一个合成的唯一键（比如时间戳加随机 ID），下游表改成 Unique 模型，用这个合成键做去重。不过要说明的是，这种方式只能缓解问题，更根本的解决思路还是推动业务侧为每张表定义真正的主键。

端到端精确一次依赖 checkpoint 和下游 sink 的事务支持。如果下游是不支持事务的存储系统，最多只能做到“至少一次”。我们做过一个测试：模拟下游 ES 写入过程中断网，恢复后确实出现了少量重复索引覆盖，但最终结果一致，因为 ES 是按文档 ID 覆盖的。所以我对下游的要求是：一定要支持按主键 upsert 或者文档覆盖更新，否则精确一次只是一句空话。

5. 一阶段实践后的几点体会与扩展方向

如果非要用一句话总结这段 Flink CDC 实践，我会说：实时同步的能力极大提升了业务数据时效性，同时带来了全新的系统治理要求。你可能以前不需要关心 MySQL 的 binlog 保留天数、不需要给下游系统设计幂等标签、不需要把 DDL 变更纳入流程管理，但上了 CDC 之后，这些全都变成了日常工作的一部分。

我个人实际操作中有几个一直坚持的经验，在这里分享给刚开始接触 Flink CDC 的读者：

第一，同步任务上线前，先执行一次完整的全量初始化，并且把这段时间的延迟指标记录下来，作为后续优化的基线数据。没有基线数据，后面排查性能问题只能靠猜。

第二，每个同步任务都用 savepoint 管理，升级连接器版本或者调整 SQL 之前，手动创建一个 savepoint，出了问题可以立即回滚。我自己因为贪图方便跳过 savepoint 吃过一次亏，升级版本后作业状态对不上，被迫全量重新同步了一次。

第三，不要在一个同步作业里堆太多表。语法上你可以把几十张表都装进一个作业里，但运维上这绝对是一个灾难。一张表结构出问题，整个作业就会挂起，所有表的同步都中断。我们后期把大作业按业务域拆分了，一个作业最多负责 5 张核心表，故障影响范围可控，排查起来也快很多。

后续我们在规划把 Flink CDC 的消费链路进一步延伸到 Iceberg 数据湖和 Kafka 消息总线，让同一个变更流可以同时服务于实时报表、离线数仓和业务系统的消息通知。有了前面的基础，这些事情做起来就顺畅很多了。

最后再补充一个小建议：网上看 Flink CDC 的文档和示例时，一定要注意版本号。Flink 版本、CDC 连接器版本、下游连接器版本，三者之间的兼容关系直接决定了很多参数能不能用、某些语法是否支持。我们最早踩的坑，就是因为集群里 Flink 是 1.15，但 CDC 连接器用了 2.4 的版本，两者的 REST API 和 SQL 校验逻辑对不上，任务一直提交失败。把版本统一到官方兼容矩阵推荐的组合之后，问题就消失了。这个看似简单的小事，其实能帮你省下很多排查时间。