简介：这份文档面向具备 Flink 基础的大数据研发与实时同步工程师，围绕 Flink CDC 3.0 讲解 MySQL 变更捕获并同步至 Doris 的链路，可用于理解 CDC 原理与搭建 Streaming ETL 任务。压缩包仅含 1 个 docx 文件，约 145KB，篇幅精炼却覆盖概念到实操环节。内容对比查询式与 Binlog 式 CDC，说明 flink-cdc-connectors 读取全量与增量数据，并通过数据准备、DataStream 和 Flink SQL 示例演示 MySQL 到 Doris 的环境搭建、配置编写、任务启动与测试。文中强调开启 Binlog、设置检查点等细节，给出可参照代码与操作步骤；已有 1789 人学习，适合希望快速上手并保障数据一致性的工程师。

1. 变更捕获的真相：为什么“读日志”比“查表”更接近数据的本来面目

做实时同步的人多半踩过这样一个坑：定时任务每 5 分钟扫一次 MySQL 的更新字段，安稳跑了半年，直到业务方要一份“删除也要计入”的对账报表，这套逻辑当场失效——物理删除在表里什么都不留，扫得再勤也扫不到已经不存在的东西。Flink CDC 3.0 切入的正是这个盲区：它不去问 MySQL“你现在长什么样”，而是订阅 binlog，把每一次 insert / update / delete 当成事件流接出来，再用 Streaming ETL 送进 Doris 这类 OLAP 引擎。价值不在“快”，而在于每一次变化都变成了可重放、可追溯、可断点续传的记录，这才是实时同步的底层底气。

2. 两条 CDC 路线的分野：查询式与 Binlog 式究竟差在哪一层

2.1 查询式 CDC 的天花板藏在“时间”这个维度里

基于查询的 CDC（Sqoop、DataX 这类）本质是“拉”：给定一个时间窗口，用 where 条件把窗口内变化的数据捞出来，写到目标端，然后推进窗口。它的问题不在于性能，而在于用查询去近似表达“变更”这件事，信息量天生不足。查询只能看到某一行的当前状态，看不到这一行之前是什么、之后变成了什么，更看不到它中间被改了几次。物理删除的行在查询结果里直接消失，连一条“它曾经存在过”的痕迹都留不下，于是下游永远不可能重建出正确的删除语义。

第二个短板是粒度。多数表只有一个精度到秒的 update_time 字段，同一秒内对同一行的多次修改会被折叠成一次，窗口边界上还会出现“上一批刚扫完、下一批还没开始”的漂移区。工程上常见的补救是加自增 ID 或者把时间粒度调到毫秒，但只要源表的写入逻辑不受你控制，这个洞就永远补不严实。

第三个短板落在源库身上。每一轮全表或半表扫描都要走一遍索引、读一遍数据页，数据量涨到千万级之后，这些扫描会和线上业务抢 IO 和 Buffer Pool。调度频率调高，源库压力变大；调低，延迟又上去了，两头都不好受。查询式 CDC 的延迟下限基本就等于调度周期，这是结构性的，换工具也绕不开。

2.2 Binlog 式 CDC：把自己伪装成一个从库

Binlog 式 CDC 换了一个思路：不再从表里“问”，而是从数据库的变更日志里“接”。MySQL 在主从复制时，从库向主库发起一个复制连接、声明自己的 server-id、请求从某个位点开始的 binlog，主库的 dump 线程就把事件一条条推过来。Canal、Maxwell、Debezium 走的都是这条路——把自己伪装成一个从库，从而拿到和真实从库完全一样的字节流。

这条路上的关键前提是binlog_format=ROW。行模式下，一条 update 事件里同时带着变更前和变更后的完整行镜像，一条 delete 事件里带着被删行的全部字段。这意味着三件事同时成立了：删除能被捕获、更新前后的值都能拿到、每条变更都带着自己的位点（binlog file + position）因而可重放。语句模式下 binlog 只记 SQL 文本，拿不到行镜像，CDC 工具基本没法用，所以配置项里那个binlog_format=row不是可选项，是入场券。

延迟的量级也随之改变。轮询是“我多久问一次”，推送是“你一变我就收到”，中间没有调度周期的概念，正常负载下毫秒到秒级就能到下游。源库这边，写 binlog 本来就是主库提交事务的必经步骤，CDC 只是额外挂了一个消费者，不会额外触发数据页扫描——代价从“周期性压榨 IO”变成了“多占一个复制连接和一点网络带宽”。

还有一层工程细节值得说清楚：Debezium 本身是跑在 Kafka Connect 框架里的，传统用法是 MySQL → Kafka → 计算引擎，中间隔着一个 Kafka。Flink CDC 的做法是把 Debezium 的引擎内嵌进 Flink Source 里，让 binlog 事件直接变成 Flink 的SourceRecord，省掉 Kafka 这一跳的部署和运维。少一个组件就少一处故障点，也少一份端到端延迟，这是它在大数据栈里迅速铺开的重要原因。

2.3 Batch 与 Streaming 的分野，决定了延迟的量级

把两条路线摆在一起看，差异其实集中在四个维度上，而其中“执行模式”这一条是最容易被低估的：

执行模式之所以关键，是因为它决定了状态能不能被保存。批任务跑完就退出，中间进度没有地方可放，失败只能整批重来；流任务常驻，可以把 binlog 位点、快照进度这些信息做成算子状态，随 Checkpoint 一起落盘。断点续传、精确一次、增量快照，这些能力全部建立在“任务活着并且能记住自己在哪”这个前提上。所以选型时真正该问的不是“哪个快”，而是“我需不需要它在故障之后还记得自己读到哪了”。

3. Source 内部构造：全量与增量是怎么在同一根管道里接力的

3.1 一行变更在 Flink 里长什么样：SourceRecord 与反序列化

Debezium 从 binlog 解析出来的是SourceRecord对象，它是一坨结构化的键值对，直接扔给下游不好用。Flink CDC 在 Source 后面挂了一个反序列化器，JsonDebeziumDeserializationSchema是默认选项，作用是把SourceRecord摊平成 JSON 字符串。摊平之后，一条变更大致长这样：

{ "before": { "id": "1001", "name": "zhangsan" }, "after": { "id": "1001", "name": "zhangsan-updated" }, "source": { "db": "test", "table": "t1", "file": "mysql-bin.000003", "pos": 154 }, "op": "u" }

字段含义是固定的：before是变更前的行镜像，after是变更后的行镜像，source里带着这条事件来自哪个库、哪张表、哪个 binlog 文件和位点，op表示操作类型。op的取值有四种：c表示 insert，u表示 update，d表示 delete，r表示全量快照阶段读出来的存量行。这四种值合起来，就是下游重建一张表所需的全部信息——insert 用after，delete 用before，update 两边都要，快照行按 insert 处理。

我一般会把这个结构理解成“事件溯源里的那条事件”。它不描述当前状态，只描述变化本身，所以下游可以按任意顺序重放、可以丢弃、可以从任意位点重新开始，只要位点是准的，最终结果就是准的。反过来，如果你的下游逻辑只认after而不看op，delete 事件就会被当成一条字段全空的行插进去，这是很典型的一类翻车。

3.2 增量快照：不锁表还能读到一致全量的原因

任务第一次启动时，表里已经有存量数据，光读 binlog 是补不齐历史的，必须先做一次全量快照，再切到增量。朴素做法是FLUSH TABLES WITH READ LOCK锁表 → 记位点 → 全量导出 → 解锁 → 从位点读增量。锁表意味着这段时间业务写入被卡住，几千万行的表根本没法接受。

Flink CDC 用的是增量快照（incremental snapshot），思路来自 DBLog 那套“日志与快照混合”的方案，核心是用水位线把快照和 binlog 缝起来：

1. 先开始消费 binlog，但先不急着输出，记下当前位点作为低水位（low watermark）；
2. 按主键把表切成若干个 chunk，多个 reader 并行读这些 chunk，读的过程中完全不加锁，业务该写写；
3. 每个 chunk 读完之后，记录当时的位点作为高水位（high watermark）；
4. 把低水位到高水位之间的 binlog 回放一遍，凡是落在该 chunk 主键范围内的变更，用它修正快照里读到的旧值——快照读的是“读的那一刻”，而 binlog 里才有“读完之后又变了”的部分；
5. 所有 chunk 都完成后，整条流切换到纯增量阶段，之后的输出就是纯粹的 binlog 事件。

这套接力能成立的关键在于主键。chunk 是按主键区间切的，回放时也靠主键判断“这条变更属于哪个 chunk”。有主键，切分和归位都能算得清清楚楚；没有主键，切分无从下手，只能退化成加锁读全表，并行度也无从谈起。所以在设计同步链路之前，第一件事是检查源表有没有主键，这比调任何参数都重要。

3.3 断点续传的黑匣子：Checkpoint 里到底存了什么

Flink CDC 将读取 binlog 的位置信息以状态的方式保存在 CK——这句话是整个链路可靠性的支点。具体来说，Source 算子的状态里存着三类东西：当前消费到的 binlog 位点（file + position，或者 GTID）、已经完成的 chunk 列表与未完成 chunk 的进度、以及表结构（schema）的快照。Checkpoint 触发时，这些状态和下游算子的状态一起被原子地刷到持久化存储里，位点和数据因此对齐。

对齐之后才有“精确一次”可言。设想任务在第 N 条 binlog 事件处崩掉，如果位点比实际处理进度跑得快，重启后中间那段数据就永久丢了；如果位点落后于实际进度，重启后同一批数据会被重复处理。Checkpoint 的机制保证位点只在屏障对齐的那一刻才被固化，所以恢复时要么全都在，要么全都不在。下游如果是 Doris 这类支持主键模型的存储，重复写入会被主键覆盖掉，正好把“至少一次”收敛成“精确一次”。

这也解释了为什么代码里那一大段 Checkpoint 配置不能省：

// 每 3 秒做一次 CK，并要求端到端精确一次 env.enableCheckpointing(3000L, CheckpointingMode.EXACTLY_ONCE); // 单次 CK 超过 1 分钟判定失败，避免卡死的 CK 拖住整个任务 env.getCheckpointConfig().setCheckpointTimeout(60 * 1000L); // 两次 CK 之间至少间隔 3 秒，防止小间隔下 CK 堆积 env.getCheckpointConfig().setMinPauseBetweenCheckpoints(3000L); // 任务被 cancel 时保留最后一次 CK，作为下次启动的起点 env.getCheckpointConfig().enableExternalizedCheckpoints( CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION); // 状态存到 HDFS，JobManager 挂了也能从同一个位置恢复 env.setStateBackend(new HashMapStateBackend()); env.getCheckpointConfig().setCheckpointStorage("hdfs://node02:8020/flinkCDC");

几个参数值得逐个说透。enableCheckpointing的间隔是可靠性和开销的平衡点：调太短，每次 CK 都要把状态刷一遍，TaskManager 的 IO 会被打满；调太长，故障时重放的 binlog 段就长，恢复变慢，而且要求 binlog 的保留时间必须覆盖这段窗口。setCheckpointTimeout是用来兜底的，CK 卡住不报错比报错更可怕，因为你会误以为一切正常。RETAIN_ON_CANCELLATION是那颗后悔药——手工 cancel 任务做升级时，如果不保留 CK，下次启动就只能从最新的 binlog 位点或者重新全量开始，前面读的全白干。setCheckpointStorage指向分布式存储而不是本地磁盘，是因为 JobManager 一旦换机器，本地路径就没了。

4. 三层抽象的对齐：DataStream、Flink SQL 与 Pipeline YAML

4.1 DataStream 层：MySqlSource 构建器与五种启动位点

最底层也最灵活的是 DataStream API。构建一个 MySQL Source 的姿势是链式调用，把连接信息、抓取范围、反序列化器和启动模式依次挂上去：

MySqlSource<String> mySqlSource = MySqlSource.<String>builder() .hostname("node03") // MySQL 地址 .port(3306) .databaseList("test") // 抓取的库，可写多个 .tableList("test.t1") // 抓取的表，库名.表名，支持正则 .username("root") .password("000000") // 把 Debezium 的 SourceRecord 转成 JSON 字符串 .deserializer(new JsonDebeziumDeserializationSchema()) // 启动位点策略：首次启动先做全量快照，再接着读增量 .startupOptions(StartupOptions.initial()) .build(); DataStreamSource<String> mysqlDS = env.fromSource(mySqlSource, WatermarkStrategy.noWatermarks(), "MySqlSource"); mysqlDS.print();

databaseList和tableList决定了抓取范围，后者支持test\.*这种正则写法，整库同步时可以直接把库下所有表圈进来。deserializer决定了流里跑的是什么类型，用 JSON 方便调试，用自定义的DebeziumDeserializationSchema可以直出 POJO，省掉下游解析。

真正需要想清楚的是startupOptions，它决定了任务第一次启动时从哪儿开始读：

这张表最容易踩的坑是earliest。它的语义是“从 binlog 里能翻到的最早位置开始”，如果 binlog 只保留三天，那你拿到的就是三天的变更，三天以前的历史数据一条都没有。带earliest上生产之前，务必确认 binlog 的保留策略真的覆盖了你要的历史区间，这是个没有后悔药的决定。

4.2 Flink SQL 层：DDL 参数怎么落到同一个 Source 上

Flink SQL 把这套东西包装成了一张带 connector 的虚拟表。写起来只有一段 DDL，背后生成的 Source 和 DataStream 里手工构建的那个是同一个：

CREATE TABLE t1 ( id STRING, name STRING, PRIMARY KEY (id) NOT ENFORCED -- 声明主键，用于生成 changelog 语义 ) WITH ( 'connector' = 'mysql-cdc', 'hostname' = 'node03', 'port' = '3306', 'username' = 'root', 'password' = '000000', 'database-name' = 'test', 'table-name' = 't1' );

PRIMARY KEY ... NOT ENFORCED这一行不是装饰。Flink SQL 的 changelog 语义要求它知道哪一列是主键：update 事件要靠主键生成一条-U（撤回旧值）加一条+U（插入新值），delete 事件要靠主键生成-D。没有主键声明，select * from t1出来的流在 upsert 场景下就没法正确收敛，下游聚合会越算越多。NOT ENFORCED只是告诉 Flink“别在运行时真的去校验唯一性”，校验的责任交给 MySQL 自己。

SQL 层的好处是把同步链路变成了可以 join、可以聚合的普通表。MySQL 的t1和t3在同一个 Flink SQL 作业里直接JOIN，拿到的就是实时的宽表，不需要额外落一个中间存储。代价是调试信息不如 DataStream 直观，出错时得看 JobManager 的日志才能定位到底卡在反序列化还是连不上库。

4.3 Pipeline 层：一个 YAML 把 MySQL 到 Doris 的整条链路收敛掉

到了 3.0，多出来一层 Pipeline 抽象，把“从 A 库同步到 B 库”这种整库同步写成了一个 YAML 文件。启动方式也随之变化，不再是flink run一个 jar，而是bin/flink-cdc.sh job/mysql-to-doris.yaml：

source: type: mysql hostname: node01 port: 3306 username: root password: "123" tables: test.\.* # 正则圈定整库，也可写单表 test.t1 server-id: 5400-5404 # 范围要 >= 并行度，每个 reader 占一个 server-time-zone: UTC+8 # 与 MySQL 侧时区保持一致 sink: type: doris fenodes: node01:8030 # FE 的 HTTP 端口 username: root password: "000000" table.create.properties.light_schema_change: true table.create.properties.replication_num: 1 pipeline: name: Sync MySQL Database to Doris parallelism: 1

三段结构各司其职。source段等价于前面那串 builder 调用，tables用正则圈范围，server-id写成范围而不是单值——这一条和并行度是绑死的，MySQL 的复制协议要求每个模拟从库的连接都有唯一的 server-id，并行度是 4 就得给 4 个号，范围写小了任务一起来就报冲突。server-time-zone决定时间字段怎么解释，MySQL 侧是东八区而这里不写，同步到 Doris 的时间列会整体差 8 小时，且不会报任何错。

sink段里table.create.properties.*这一族参数，会在目标端自动建表时透传成 Doris 的建表属性。light_schema_change打开之后，源端加列时下游能走轻量 schema change，不用整表重写；replication_num是副本数，单机环境必须设成 1，因为默认的 3 副本在只有一个 BE 的集群上根本建不出表。这两个参数不配，表现是任务起来了但目标库里一张表都没有。

pipeline段只有名字和并行度，但并行度这里有个容易被忽略的对应关系：它同时受 source 和 sink 两侧能力的约束，MySQL 侧的上限是 server-id 有多少个号，Doris 侧的上限是 BE 的写入能力。调大之前先想清楚瓶颈在哪一侧。

整库同步之外，Pipeline 还支持路由改写，把源库改个名字落到目标端：

route: - source-table: test_route.\.* sink-table: doris_test_route.\.*

source-table和sink-table都用“库名.表名”的正则，匹配上的表会被改写成目标名称。典型用途是多套环境的库名隔离，或者把源端的多个分库归并到目标端的一个库里。这里用的是正则捕获的一致性匹配，源端圈了几个表，目标端就得能接住几个表，两边的正则写得不一致，表现是部分表同步过去了、部分表静默消失，排查起来很费劲。

5. 避坑与排查：Binlog、server-id、Checkpoint 上最容易翻车的五处

5.1 编译期报找不到 MySqlSource：包名停留在旧版本

现象：依赖里明明写了 CDC 的 connector，代码里import com.ververica.cdc.connectors.mysql.source.MySqlSource却报红，或者打包时提示类不存在。

原因：这套 connector 的包名在版本演进中迁移过。2.x 时代在com.ververica.cdc下，后续版本整体搬到了org.apache.flink.cdc，类和 API 形态基本没变，但 import 路径全变了。拿着旧版示例代码配新版依赖，编译期就会卡住。

解决：先确认依赖的实际坐标和版本，再决定 import 用哪个前缀。两套包名不要混着出现，MySqlSource和StartupOptions必须来自同一个前缀，混用会在运行时报类型转换异常，比编译报错更难查。升级依赖时把 import 一次性全改掉，别留一半。

5.2 任务起来了但一条数据都没有：binlog 没开或格式不对

现象：作业状态是 RUNNING，日志里没有异常，但下游一条数据都收不到；或者首次启动时全量能读到，之后的增量变更完全没反应。

原因：MySQL 侧没开 binlog，或者binlog_format是STATEMENT/MIXED。行镜像拿不到，Debezium 在解析阶段就会卡住或者直接跳过。另一种情况是开了binlog-do-db过滤，但只写了部分库名，被同步的库没在名单里。

解决：在 MySQL 上执行show variables like 'log_bin'确认为 ON，show variables like 'binlog_format'确认为 ROW。binlog-do-db要按实际同步范围逐条列全，加完之后必须重启 MySQL 才生效，改完不重启是最常见的一类“改了但没用”。另外server-id也得配，只开 binlog 不配 server-id，复制连接一样建不起来。

5.3 并行度一调大就报 server-id 冲突

现象：并行度是 1 时一切正常，调到 4 之后任务启动失败，日志里出现类似“A slave with the same server_uuid/server_id as this slave has connected to the master”的报错。

原因：Flink CDC 的每个并行 reader 都会向 MySQL 注册一个模拟从库连接，每个连接要求独立的 server-id。如果配置里只写了一个固定值，或者写的范围小于并行度，多个 reader 就会抢同一个号。

解决：把 server-id 写成范围，上界减下界加一要不小于并行度，比如并行度 4 就写5400-5403或更宽的5400-5404。同时确认这个号段没有被别的 CDC 任务或者真实的从库占用，同一套 MySQL 上跑多个同步任务时尤其要注意错开号段。

5.4 重启后全量重跑一遍：Checkpoint 根本没生效

现象：任务失败重启，或者手工 cancel 后重新提交，日志显示又在从头做全量快照，几千万行重新扫一遍，恢复时间以小时计。

原因：三种可能。一是代码里压根没调enableCheckpointing，状态从来没落过盘；二是 checkpoint storage 配的是本地路径，换机器或者容器重建之后路径没了；三是 cancel 时没保留外部化 Checkpoint，重启时找不到起点，只能按startupOptions重新来一遍。

解决：确认enableCheckpointing被调用且间隔合理，setCheckpointStorage指向 HDFS 这类共享存储，enableExternalizedCheckpoints设成RETAIN_ON_CANCELLATION。恢复时用flink run -s <savepoint路径>显式指定起点，而不是直接重新提交。Pipeline 模式同理，Flink 配置里那个execution.checkpointing.interval必须打开，不打开等于整条链路没有断点续传能力。

5.5 Doris 端建表失败：端口与副本数

现象：同步任务日志显示 source 侧正常读到数据，但 Doris 里查不到表；或者任务直接报建表失败。

原因：fenodes填错了端口。Doris 的 FE 有多个端口，fenodes需要的是 FE 的 HTTP 端口（默认 8030），填成查询端口或者元数据端口都连不通。另一个高频原因是replication_num默认是 3，而测试环境只有一个 BE，副本数大于 BE 数量时建表直接被拒。还有一种情况是目标端的库压根没建，sink 不会替你把库创建出来。

解决：先用 MySQL 客户端连上 Doris（默认 9030 端口）手工create database，确认 FE 通、库存在。fenodes用 HTTP 端口。单 BE 环境把table.create.properties.replication_num显式设成 1。建表失败时优先看 sink 侧的日志，source 侧日志是干净的，问题通常不在读取端。

6. 设计取舍：为什么 Flink CDC 把复杂度藏进了配置里

从 2.x 到 3.0，最明显的变化是同步一条链路需要写的 Java 代码变少了，YAML 变多了。这个取舍值得琢磨：DataStream API 把每个细节都摊在你面前，灵活但啰嗦；Pipeline 模式用一份 YAML 把 source、sink、route、transform 收敛掉，DBA 改个表名正则就能调同步范围，不用重新打包发版。代价是出问题时少了一层可编程的抓手，很多东西只能在配置层面调，调不动就得退回 DataStream。

增量快照这个机制也是同样的取舍。它用 chunk 分片 + 双水位线回放换来了不锁表的一致性读，代价是对主键的强依赖——无主键表退化成锁表读，大表上等于不可用；chunk 切得越细并行度越高，但回放时要处理的 binlog 区间也越长。默认 chunk size 是几千行这个量级，我一般会在千万级表上先按这个默认值跑一轮，看 binlog 回放阶段的耗时占比，再决定要不要调。

验证这套链路有没有丢数据，光看任务状态是不够的。我的习惯是三段对账：跑完之后比一次两边的count(*)；挑几个主键区间比sum或校验和；再做一轮“制造变更”的验证——在源端插一条、改一条、删一条，看目标端是不是三条语义都对上了。delete 那条最容易露馅，因为只认after不认op的写法会把删除变成插入空行。

有一次我在测试环境上把startupOptions写成了latest，跑了一晚上数据量对不上，查了半天以为是丢包，最后发现是启动位点选错了，历史数据一条都没同步。从那以后我每次配新的同步链路，都强制先把 startup 模式和 binlog 保留时长这两项抄在纸上核对一遍，再动手写 YAML。这份资料里从建库建表、开 binlog、DataStream 与 Flink SQL 两套写法，到 MySQL 到 Doris 的 Pipeline 配置和路由改写，整条路径都走到了，按着顺序过一遍能把上面这些坑提前踩完。希望帮到你。

本文还有配套的精品资源，点击获取