1. 为什么CDC这件事值得单独拎出来聊
做数据集成的人，迟早会碰到一个绕不开的场景：业务库里的数据一直在变，下游的数仓、缓存、检索服务却只能靠定时任务一遍遍全量拉取。跑批窗口越拉越长，数据库压力越来越大，业务方还嫌数据不够实时。这个矛盾在订单、支付、库存这类高频写入的表上尤其尖锐。
Flink CDC 就是冲着这个矛盾来的。它把数据库的变更日志（binlog、WAL 这类东西）直接接进 Flink 的流处理管道，让"数据同步"从"定时搬运"变成"持续订阅"。我第一次在生产上用它替换掉一套小时级跑批任务时，下游报表的延迟从 60 分钟压到了秒级，源库的查询压力也肉眼可见地降了下来。
但真正上手之后你会发现，Flink CDC 提供了两条风格完全不同的路：一条是 DataStream API，代码里全是SourceFunction、DeserializationSchema这些偏底层的概念；另一条是 Flink SQL，几行 DDL 加一个INSERT INTO就完事。很多人卡在第一步——到底该选哪条路？两条路的坑分别在哪儿？
这篇内容就是围绕这个问题展开的。我会把 DataStream API 和 Flink SQL 两种实现方式拆开讲透，包括各自的适用场景、核心代码结构、参数配置逻辑，以及我在实际项目里踩过的那些坑。适合已经对 Flink 有基本概念、准备把 CDC 落到真实项目里的开发者，也适合正在做技术选型、想搞清楚两条路差异的架构同学。
2. Flink CDC 到底在底层做了什么
在写第一行代码之前，有必要把 Flink CDC 的工作机制讲清楚。不然你调参数的时候就是瞎调，出了问题也不知道从哪儿查。
2.1 从数据库日志到 Flink 数据流的三段式链路
Flink CDC 的本质是一条"日志捕获 → 变更解析 → 流式下发"的管道。以最典型的关系型数据库为例，整个链路可以拆成三段：
第一段是日志捕获。数据库本身会把所有数据变更写进自己的日志文件（不同数据库叫法不同，有的叫 binlog，有的叫 WAL，有的叫 redo log）。Flink CDC 通过一个独立的连接器去订阅这些日志，而不是去查表。这一点非常关键——它意味着源库上不会产生SELECT查询压力，读的是数据库自己已经在写的日志。
第二段是变更解析。日志里存的其实是二进制或者半结构化的记录，连接器需要把它解析成"哪张表、哪一行、做了什么操作（增/删/改）、变更前后的字段值分别是什么"这样的结构化事件。Flink CDC 内部用的是 Debezium 这套成熟的变更捕获框架来做解析，所以它支持的数据库类型和 Debezium 基本一致。
第三段是流式下发。解析出来的变更事件被封装成 Flink 的SourceRecord或者RowData，进入 Flink 的算子链，后续可以做过滤、聚合、join、写下游。
理解这三段之后，很多问题就顺了。比如"为什么我的任务启动后源库 CPU 没怎么涨"——因为读的是日志不是表；"为什么全量阶段和增量阶段的行为不一样"——因为全量阶段走的是另一条快照读取路径。
2.2 全量加增量：一个容易被忽略的两阶段模型
Flink CDC 最容易被低估的设计，是它的全量 + 增量一体化读取。任务刚启动时，连接器会先对目标表做一次全量快照（把当前所有数据读一遍），读完快照后无缝切换到增量日志订阅，继续消费快照期间产生的变更。
这个模型解决了一个经典难题：如果只做增量，那历史数据怎么办？如果先全量再单独起增量，那全量和增量之间的时间窗口里产生的变更就会丢。Flink CDC 通过在快照阶段记录日志位点、快照完成后从该位点继续消费，把这个问题抹平了。
但这里有个实操细节：全量阶段的并发度由scan.incremental.snapshot.chunk.size这类参数控制，它决定了快照被切成多少块并行读取。块切得太小，任务启动时会有大量小查询；切得太大，单块读取时间长，启动慢。我一般会根据表的数据量和主键分布来调，几千万行的表用默认值往往偏保守。
2.3 两条 API 路线共享的内核与分叉点
DataStream API 和 Flink SQL 在底层共享同一套连接器和解析逻辑，区别在于谁来组织数据流。
DataStream API 把连接器当作一个SourceFunction或者新的Source接口实现，变更事件以SourceRecord的形式吐出来，你需要自己写反序列化逻辑，把记录转成业务对象，再决定怎么处理。控制粒度细，但代码量大。
Flink SQL 则把连接器封装成一个表，你通过CREATE TABLE ... WITH ('connector' = 'mysql-cdc', ...)声明一张"镜像表"，然后像操作普通表一样写 SQL。Flink 在背后把 SQL 编译成算子链，连接器的细节被隐藏了。开发效率高，但可定制性弱。
一个判断原则：如果你的同步逻辑能用 SQL 表达清楚（字段映射、简单过滤、多表 join），优先用 SQL；如果涉及复杂的状态管理、自定义的旁路逻辑、或者要和非 CDC 的数据源做混合处理，DataStream API 更合适。
3. DataStream API 实现：控制力换来的复杂度
选 DataStream API 的人，通常是因为 SQL 表达不了他们的需求。我见过最多的场景是：变更事件需要先经过一段自定义的富化逻辑（比如查外部维表补字段），再写进下游；或者一条变更要同时分发到多个不同格式的下游。这些用 SQL 做起来很别扭，用 DataStream 就很自然。
3.1 依赖引入与版本对齐这件小事
DataStream 方式的第一步是引依赖。这里有个新手最容易翻车的地方：Flink 版本和 CDC 连接器版本必须对齐。Flink CDC 的版本号里通常带着它适配的 Flink 大版本，比如 2.4 系列适配 Flink 1.13 到 1.17 不等，具体要看官方兼容性矩阵。
一个典型的 Maven 依赖长这样：
<dependency> <groupId>com.ververica</groupId> <artifactId>flink-connector-mysql-cdc</artifactId> <version>2.4.2</version> </dependency>注意这个ververica的 groupId，早期版本是这样，后来部分连接器迁移到了 Apache 官方仓库。引依赖前先确认你用的 Flink 发行版和连接器来源是否匹配，否则会出现类找不到或者方法签名不兼容的问题。
还有一个隐藏坑：连接器内部依赖的 Debezium 版本可能和你项目里其他依赖冲突。我遇到过一次，项目里另一个组件引了不同版本的 Jackson，导致 CDC 解析日志时抛序列化异常。解决办法是用maven-shade-plugin把连接器依赖打进去，或者用dependencyManagement强制统一版本。
3.2 构建 Source：从 SourceFunction 到新 Source 接口的迁移
Flink 1.14 之后，老的SourceFunction接口被标记为废弃，新的Source接口（FLIP-27）成为主流。Flink CDC 也提供了两套实现。如果你还在用老接口，代码大概是这样：
MySqlSource<String> source = MySqlSource.<String>builder() .hostname("localhost") .port(3306) .databaseList("inventory") .tableList("inventory.orders") .username("reader") .password("******") .deserializer(new JsonDebeziumDeserializationSchema()) .build(); StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); env.enableCheckpointing(3000); env.fromSource(source, WatermarkStrategy.noWatermarks(), "mysql-cdc") .print(); env.execute("cdc-demo");新接口的写法在fromSource这一层是一样的，区别主要在连接器内部实现。对使用者来说，最需要关注的是deserializer这个参数——它决定了变更事件以什么格式进入你的算子。
JsonDebeziumDeserializationSchema是最省事的，直接把变更转成 JSON 字符串。但它的输出结构比较"重"，包含 before、after、source、op 等一堆元数据字段。如果你下游只需要 after 里的业务字段，用这个 schema 会带来不必要的解析开销。
3.3 自定义反序列化：把变更事件变成业务对象
真正在生产里跑，几乎没人直接用 JSON 字符串。你需要一个自定义的DebeziumDeserializationSchema，把SourceRecord转成自己的 POJO。
public class OrderDeserializer implements DebeziumDeserializationSchema<Order> { @Override public void deserialize(SourceRecord record, Collector<Order> out) { Struct value = (Struct) record.value(); String op = value.getString("op"); Struct after = value.getStruct("after"); if (after != null) { Order order = new Order(); order.setId(after.getInt64("id")); order.setAmount(after.getFloat64("amount")); order.setStatus(after.getString("status")); order.setOp(op); out.collect(order); } } @Override public TypeInformation<Order> getProducedType() { return TypeInformation.of(Order.class); } }这段代码里有几个值得说的点。op字段标识操作类型，常见取值是c（create）、u（update）、d（delete）、r（read，全量阶段）。删除操作的after是 null，只有before有值，所以如果你要处理删除，得单独判断。
另外，getProducedType返回的类型信息会影响 Flink 的序列化策略。用TypeInformation.of(Order.class)让 Flink 走 POJO 序列化，比 Kryo 效率高。前提是你的 Order 类有无参构造、字段有 getter/setter、字段类型都是 Flink 能识别的。
实操心得：自定义反序列化器里尽量不要做耗时操作（比如查数据库、调远程接口）。这个方法是逐条记录调用的，一旦变慢会直接拖垮整个 Source 的吞吐。需要富化的话，放到后面的
map或async算子里去。
3.4 变更流的下游处理与 Exactly-Once 保障
拿到DataStream<Order>之后，下游怎么处理就是常规的 Flink 开发了。可以keyBy后做窗口聚合，可以map成另一种格式写 Kafka，也可以sink到数据库。
这里必须提Exactly-Once。Flink CDC 的 Source 是支持精确一次的，但前提是你开了 checkpoint，并且下游 sink 也支持事务或者幂等写入。我见过有人开了 checkpoint 但 sink 用的是普通 JDBC 连接，结果故障恢复后数据重复写入，还以为是 CDC 的问题。
正确的做法是：Source 端开启 checkpoint（env.enableCheckpointing），sink 端用两阶段提交的 connector（比如 Kafka sink 开事务、JDBC sink 用幂等 upsert）。checkpoint 的间隔也要权衡——太短会增加源库和网络压力，太长则故障恢复时重放的数据量大。我一般从 3 到 5 秒起步，根据数据量调整。
还有一个容易忽略的点：checkpoint 超时时间。CDC 任务在快照阶段可能长时间没有数据流动，如果 checkpoint 超时设得太短，会频繁失败。建议把execution.checkpointing.timeout设得比常规任务宽松一些。
4. Flink SQL 实现：几行 DDL 背后的门道
如果你的同步逻辑能用 SQL 描述，那 Flink SQL 的体验会好很多。整个任务可能就三段：建源表、建目标表、写一条 INSERT。但"简单"的表象下，参数配置和语义细节一点不少。
4.1 用 DDL 声明一张 CDC 源表
Flink SQL 里，CDC 源表通过CREATE TABLE加WITH参数声明：
CREATE TABLE orders ( id BIGINT, amount DOUBLE, status STRING, PRIMARY KEY (id) NOT ENFORCED ) WITH ( 'connector' = 'mysql-cdc', 'hostname' = 'localhost', 'port' = '3306', 'username' = 'reader', 'password' = '******', 'database-name' = 'inventory', 'table-name' = 'orders' );几个关键点。PRIMARY KEY ... NOT ENFORCED是必须的，Flink 需要主键来做 upsert 语义和状态管理，但NOT ENFORCED表示 Flink 不会真的去校验这个约束，只是拿它当元信息用。
connector参数的值决定了用哪个连接器，mysql-cdc、postgres-cdc、mongodb-cdc等等。不同连接器的参数名有差异，比如 MySQL 用database-name和table-name，有些连接器用database和collection。写之前查一下对应连接器的文档，别凭记忆写。
还有一个隐藏参数scan.startup.mode，控制从哪儿开始读。默认是initial，即先全量再增量。如果只想读增量，设成latest-offset；如果想从指定位点开始，用specific-offset或者timestamp。这个参数在数据修复场景里特别有用。
4.2 目标表与 INSERT INTO 的语义陷阱
目标表的声明取决于你要写到哪儿。写 Kafka 是一种，写另一张数据库表是另一种。以写 MySQL 为例：
CREATE TABLE orders_sink ( id BIGINT, amount DOUBLE, status STRING, PRIMARY KEY (id) NOT ENFORCED ) WITH ( 'connector' = 'jdbc', 'url' = 'jdbc:mysql://localhost:3306/dw', 'table-name' = 'orders', 'username' = 'writer', 'password' = '******' ); INSERT INTO orders_sink SELECT id, amount, status FROM orders;这条 INSERT 看起来平平无奇，但语义上它是一条持续查询，永远不会结束。源表每来一条变更，它就会触发一次对目标表的写入。
这里有个大坑：目标表的主键必须和源表的主键语义对齐。如果源表主键是 id，目标表也声明了 id 为主键，那 Flink 会用 upsert 模式写入，更新和删除都能正确处理。但如果目标表没声明主键，Flink 只能按 append 模式写入，删除操作就丢了，更新也会变成插入新行。
注意：写 JDBC sink 时，如果目标表主键和源表不一致，或者目标表根本没有主键，务必确认你的业务能接受 append-only 的语义。我见过因为这个问题导致下游数据翻倍的案例。
4.3 多表同步与整库同步的配置方式
实际项目里很少只同步一张表。Flink SQL 支持两种多表方式。
一种是显式声明多张源表，然后写多条 INSERT。这种方式清晰可控，但表多了之后 DDL 会很长。
另一种是用正则匹配表名，让一个源表声明覆盖多张表：
CREATE TABLE all_orders ( ... ) WITH ( 'connector' = 'mysql-cdc', 'database-name' = 'inventory', 'table-name' = 'order_.*' );注意table-name支持正则，但用正则时表结构必须一致，否则字段对不上会报错。而且正则匹配到的表在主键定义上也要统一，不然 upsert 语义会乱。
整库同步是更进一步的场景，通常需要配合database-name的正则或者专门的整库同步工具。Flink CDC 本身在 SQL 层面做整库同步需要一些技巧，比如用INFORMATION_SCHEMA动态生成 DDL，或者借助外部的 schema 管理组件。这块展开会比较长，核心思路是"先探测源库的表结构，再动态拼出 Flink SQL"。
4.4 SQL 路线的性能调优参数
Flink SQL 任务的性能调优，一半在 SQL 本身，一半在连接器参数。
连接器侧最值得关注的几个参数：
| 参数 | 作用 | 调优建议 |
|---|---|---|
scan.incremental.snapshot.chunk.size | 全量快照分块大小 | 大表适当调大，减少分块数 |
scan.snapshot.fetch.size | 每次从源库拉取的行数 | 网络好可调大，源库压力大则调小 |
debezium.snapshot.mode | 快照模式 | 默认 initial，按需调整 |
server-time-zone | 服务器时区 | 必须和源库一致，否则时间字段错乱 |
SQL 侧则要注意状态大小。如果 INSERT 里带了聚合或者 join，Flink 会维护状态。CDC 源本身也会维护位点状态。状态太大时，checkpoint 会变慢甚至失败。可以通过设置状态的 TTL 来清理过期数据，或者用 RocksDB 状态后端把状态放到磁盘上。
还有一个反直觉的点：并行度不是越高越好。CDC 源在增量阶段，单个表的变更日志是串行消费的（因为要保证顺序），并行度再高也用不上。真正能并行的是全量阶段和下游算子。所以盲目调大并行度，只会增加资源浪费和 checkpoint 开销。
5. 两条路线怎么选：一张对照表和几个真实判断
讲完两种实现，回到最开始的问题：到底选哪个？
5.1 从开发效率、可控性、运维成本三个维度对比
| 维度 | DataStream API | Flink SQL |
|---|---|---|
| 开发效率 | 低，需要写反序列化和处理逻辑 | 高，DDL + INSERT 即可 |
| 可控性 | 高，任意自定义逻辑 | 中，受限于 SQL 表达能力 |
| 运维成本 | 中，代码变更需重新打包 | 低，改 SQL 即可 |
| 调试难度 | 中，可打断点 | 高，依赖日志和 explain |
| 适用场景 | 复杂富化、多路分发、混合数据源 | 字段映射、简单过滤、多表 join |
这张表不是绝对的。我见过用 DataStream 写得很优雅的简单同步，也见过用 SQL 硬凑复杂逻辑最后维护不动的。关键看你的团队更熟悉哪套。
5.2 什么情况下 SQL 会逼你回到 DataStream
有几个信号出现时，说明你该考虑 DataStream 了。
第一，变更事件需要调用外部服务。比如每条订单变更都要查一次用户中心补用户等级。SQL 里可以用 lookup join 做维表关联，但如果外部服务不是数据库、没有标准 connector，SQL 就无能为力了。
第二，一条变更要分发到多个异构下游，且每个下游的格式和处理逻辑都不同。SQL 里可以用多个 INSERT 从同一张源表出发，但如果每个下游还需要不同的富化逻辑，SQL 会变得非常臃肿。
第三，需要精细控制状态和容错。比如你想自定义 checkpoint 时的状态快照逻辑，或者想在故障恢复时做一些特殊处理，这些在 SQL 层面是黑盒。
反过来，如果你的需求就是"把 A 库的几张表同步到 B 库，字段做点映射和过滤"，那 SQL 是明显更优的选择，没必要为了"可控"去写一堆 Java 代码。
5.3 混合使用：SQL 和 DataStream 并非二选一
很多人不知道，Flink 允许在同一个任务里混用 SQL 和 DataStream。你可以用 SQL 声明 CDC 源表，然后通过toChangelogStream()把它转成 DataStream，接着用 DataStream API 做复杂处理，处理完再toChangelogStream()转回表，继续用 SQL 写下游。
这个能力在真实项目里非常实用。我做过一个任务，源表用 SQL 声明（省去写反序列化），中间用 DataStream 做了一段自定义的会话窗口聚合，最后又用 SQL 写进目标库。整个任务既有 SQL 的开发效率，又有 DataStream 的灵活性。
代价是代码结构会复杂一些，需要理解 Table API 和 DataStream API 之间的转换语义。转换过程中，changelog 的类型（insert-only 还是 retract）会影响下游处理，这块要特别小心。
6. 那些文档里不会写的踩坑记录
前面讲的都是"应该怎么做"，这一节讲"实际做的时候会出什么幺蛾子"。这些都是我在真实项目里踩过的，希望能帮你省点时间。
6.1 时区问题导致的时间字段错乱
这是最高频的坑，没有之一。CDC 从源库读出来的时间字段，到了 Flink 里可能差 8 小时，也可能差别的数。原因通常是源库时区、连接器时区、Flink 任务时区三者不一致。
排查思路：先确认源库的time_zone设置，再检查连接器的server-time-zone参数，最后看 Flink 任务的table.local-time-zone配置。三者对齐之后，时间字段才不会漂。
我一般的做法是：连接器里显式写'server-time-zone' = 'Asia/Shanghai'，Flink 配置里也显式指定时区，不依赖默认值。默认值在不同部署环境下可能不一样，显式指定最稳。
6.2 大表全量阶段的 OOM 与超时
几亿行的表做全量快照，很容易遇到内存溢出或者快照超时。根因通常是分块太大，单块数据一次性加载到内存。
解决方向有三个：调小scan.incremental.snapshot.chunk.size，让每块数据量可控；调大任务内存，给快照阶段留足空间；或者用scan.startup.mode = latest-offset跳过全量，只做增量（前提是历史数据已经通过其他方式同步过了）。
还有一个隐蔽的问题：全量阶段如果源库有大量写入，快照读和增量日志之间可能产生锁竞争。有些数据库在快照读时会加锁，影响业务。这种情况要考虑在业务低峰期启动任务，或者用无锁快照模式（如果连接器支持）。
6.3 主键变更引发的下游数据异常
CDC 捕获的是行级变更，如果源表的主键被更新了，Debezium 会把它解析成一条删除加一条插入。如果你的下游是按主键做 upsert 的，这条"删除+插入"会先删掉旧主键的行，再插入新主键的行，看起来是对的。
但如果下游没有正确处理删除，或者主键变更涉及多行（比如批量更新主键），就可能出现数据不一致。更麻烦的是，有些业务表的主键是自增的，理论上不该变，但实际中因为数据迁移等原因确实会变。
我的建议是：在源表层面尽量避免更新主键。如果无法避免，确保下游的 sink 能正确处理 delete 事件，并且在同步任务里对主键变更做监控告警。
6.4 任务重启后的位点恢复行为
CDC 任务重启时，从哪个位点恢复，取决于 checkpoint 和 savepoint 的状态。如果 checkpoint 正常，任务会从最后一次成功的 checkpoint 位点继续，保证不丢不重。
但有两种情况会导致行为异常。一是 checkpoint 一直失败，任务重启后可能从头开始全量，这在生产上是灾难。二是源库的日志被清理了，checkpoint 里记录的位点已经不存在，任务会报错或者跳过一段数据。
防范措施：监控 checkpoint 的成功率，设置合理的日志保留时间（确保大于最长故障恢复时间），以及定期做 savepoint 备份。我一般会把日志保留时间设成至少 7 天，给故障排查留足窗口。
7. 把 CDC 任务跑稳的几个工程习惯
技术细节讲完了，最后分享几个让 CDC 任务长期稳定运行的工程习惯。这些不是某个参数能解决的，而是日常运维中积累的经验。
第一，给每个 CDC 任务配独立的监控。重点盯三个指标：checkpoint 成功率、端到端延迟、源库日志消费位点。延迟突然上涨往往意味着下游处理变慢或者源库写入激增，位点停滞则可能是连接器出了问题。
第二，变更前先在预发环境验证。CDC 任务的变更（改 SQL、调参数、升级连接器版本）影响面很大，直接上生产风险高。预发环境跑一遍，观察至少一个完整的 checkpoint 周期，确认没有异常再上。
第三，保留回滚能力。每次变更前记录当前的任务配置和 savepoint 路径，出问题时能快速回滚。我见过因为升级连接器版本导致数据格式变化、下游解析失败的案例，有 savepoint 就能几分钟内恢复。
第四，对源库保持敬畏。CDC 虽然读的是日志，但全量阶段仍然会对源库产生读压力，连接器建立连接也会占用资源。在源库负载高的时候启动全量任务，可能成为压垮骆驼的最后一根稻草。选好启动时机，比调任何参数都重要。
这套东西我在多个项目里反复验证过，DataStream 和 SQL 两条路都跑过，核心结论就一句：先想清楚你的同步逻辑能不能用 SQL 表达，能就用 SQL，不能再用 DataStream，别为了技术上的"掌控感"去写不必要的代码。真正难的不是写代码，而是让任务在几个月甚至几年里稳定地跑下去。
