简介：面向具备一定 Flink 基础的大数据研发与实时同步工程师，这份文档围绕 Flink CDC 3.0 展开，重点解决 MySQL 变更数据实时捕获并同步至 Doris 的落地问题，可作为流式 ETL 场景的入门与实操参考。资源包为单个 docx 文件，约 145KB，内容集中，便于按章节查阅。已有 1789 人学习，反映其在实时数据集成方向具有一定关注度。文档先说明 CDC 通过监测数据库插入、更新、删除等变动并按顺序记录供下游订阅的核心思想，对比基于查询与基于 Binlog 两种方式在延迟、完整性和数据库压力上的差异，再介绍 Flink CDC 连接器直接读取 MySQL 等全量与增量数据的能力。案例部分覆盖数据准备、DataStream 与 Flink SQL 应用、MySQL 到 Doris 的流式 ETL 环境搭建、配置文件编写、任务启动及测试，读者可据此理解 Binlog 开启、检查点设置等关键环节，掌握从数据源接入、流处理到目标存储对接的完整链路，并用于实际项目中的参数调整与稳定性验证。
1. 一条 UPDATE 到 Doris 只要几秒，代价却不在「快」上
很多人第一次接触 Flink CDC 做 MySQL 到 Doris 的实时同步，注意力都放在延迟有多低上。真正跑起来才发现，难的从来不是把变更读出来，而是让「全量 + 增量」这条 StreamingETL 链路在中途崩掉之后，还能站在原来的位置上继续走。基于查询的同步工具每隔几分钟扫一遍表，靠 WHERE 条件捞增量，延迟高、压力大，还要求业务表有可靠的更新时间字段；基于 Binlog 的 CDC 换了思路，从日志层面拿变更，顺序完整、对源库几乎零侵入。Flink CDC 3.0 又往前推了一步：用一份 YAML 把 source、sink、路由描述清楚，交给 pipeline 去跑，不必再手写一整套 DataStream 作业。下面拆的是它凭什么 work。
2. 增量快照：全量和增量是怎么缝成一条流的
2.1 基于查询的 CDC，延迟和压力究竟出在哪
早期的同步方案基本都是批处理思维：Sqoop 或 DataX 拉一次全量，之后用定时任务按update_time > 上次时间捞增量。这套模式有三个绕不过去的硬伤，而且一个比一个隐蔽。
第一是捕获不全。物理删除在源表里不留痕迹，按时间戳查询永远查不到「被删掉的那一行」，目标端就会一直挂着一份幽灵数据。第二是延迟由调度周期决定，跑五分钟一次就是五分钟延迟，跑一分钟一次又意味着源库每分钟被全表扫一遍。第三是压力在线性叠加，分库分表一多，几百个定时任务在同一个时间窗口涌进来，数据库的监控曲线会非常难看。
更麻烦的是时间戳本身不可靠。业务代码用now()写库、主从之间存在毫秒级延迟、批量更新时一批行共享同一个时间戳——这些在业务上都没问题，但会让增量条件出现漏读或重复读。血泪经验往往是：同步跑了大半年，某天对账发现少了几万行，回溯才发现是某次跨库时间不同步。这类问题不报错、不告警，是个纯粹的黑匣子。
Binlog 路线把视角换了一层：不再问「表里哪些行变了」，而是问「数据库引擎自己记录了什么」。MySQL 的 row 模式 Binlog 会逐行记录 before/after 镜像，INSERT、UPDATE、DELETE 都在里面，也有精确的位点。变更的完整性由数据库自身保证，同步端只是订阅者。
2.2 增量快照：把大表切成能并行的块
只读 Binlog 是不够的，因为历史数据不在日志里。所以任何 CDC 工具的第一件事都是「先快照、再追增量」。传统做法是加全局读锁（FTWRL）拿到一致性位点，然后全表 SELECT，期间业务写入被阻塞。一张几亿行的表扫几个小时，锁就持几个小时，这在生产上基本不可接受。
Flink CDC 的核心贡献就在这一层，它用的是无锁并行增量快照。常见的实现思路大致是这样：
- 连接器先读取表的当前 Binlog 位点，作为「起点水位」，此时不加锁；
- 按主键把表切成若干 chunk（分片），每个 chunk 记录自己的主键上下界；
- 多个并发任务同时读取各自的 chunk，这期间 Binlog 也在被消费，读到的事件先在内存里缓冲；
- chunk 读完时，把缓冲区内落在该 chunk 主键范围内的变更事件重放一遍，覆盖掉快照里的旧值；
- 全部 chunk 完成后，缓冲区的消费位置接上正式的下游消费，全量与增量无缝衔接。
这个设计的巧妙之处在于：它把「一致性」这个全局问题，降解成了「每个 chunk 局部一致 + 边界重放」的局部问题。不需要锁表，也不需要把整张表当成一个不可分割的事务。代价是连接器必须自己维护一堆进度元数据——哪些 chunk 读完了、每个 chunk 的边界事件补偿到哪了、当前 Binlog 位点在哪。这些元数据就是后面要讲的状态。
提示：chunk 切分的粒度由连接器内部根据表大小和主键分布自动计算，并行度越高、chunk 被摊得越细，但 chunk 数量远大于并行度时，剩余 chunk 会变成木桶的短板，需要看任务里的 split 分配情况来判断是不是切歪了。
2.3 位点即状态：checkpoint 里存的到底是什么
理解了「位点即状态」，Flink CDC 的很多配置就顺理成章了。Source 在运行时会往 Flink 的 state 里写两类东西：一类是已完成的 chunk 集合与未完成 chunk 的读取进度，另一类是当前消费到的 MySQL Binlog 位点（binlog 文件名 + offset，开启 GTID 时还包括 GTID 集合）。这两部分随 checkpoint 一起落盘。
所以捕获任务的可靠性，本质上是 checkpoint 的可靠性。任务挂掉后从最近的 checkpoint 恢复，连接器读出状态里的位点，直接从那里续读 Binlog，不会重跑全量。反过来，如果没开 checkpoint，或者把外部化 checkpoint 关掉（RETAIN_ON_CANCELLATION没配），任务一 cancel 状态就没了，下次启动只能从initial重新全量扫一遍——这就是很多人口中「重启一次同步两小时」的真正原因。
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); env.setParallelism(4); // 关键：开启 checkpoint，位点信息才有地方落脚 env.enableCheckpointing(3000L, CheckpointingMode.EXACTLY_ONCE); env.getCheckpointConfig().setCheckpointTimeout(60 * 1000L); // 两次 CK 之间留出间隔，避免大状态表一直处于 checkpoint 中 env.getCheckpointConfig().setMinPauseBetweenCheckpoints(3000L); // 任务被 cancel 时保留最后一次 CK，否则状态随作业一起消失 env.getCheckpointConfig().enableExternalizedCheckpoints( CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION); // 失败后按频率重启：1 天内最多 3 次，每次间隔至少 1 分钟 env.setRestartStrategy(RestartStrategies.failureRateRestart( 3, Time.days(1L), Time.minutes(1L))); env.setStateBackend(new HashMapStateBackend()); env.getCheckpointConfig().setCheckpointStorage("hdfs://node01:8020/flink-cdc");这段配置里每一行都对应一个失败场景。setCheckpointTimeout防止某个算子卡住导致整个 CK 永远不完成；setMinPauseBetweenCheckpoints在状态较大时给系统喘息时间，否则 CK 会一个接一个排队；failureRateRestart比固定延迟重启更安全，因为如果是配置性错误（比如密码错了），无限重启只会把日志刷爆。状态后端选HashMapStateBackend把状态放内存、快照写 HDFS，是这类无大窗口作业的常规选择。
真正需要留心的是 checkpoint 间隔和端到端延迟的关系。CK 间隔 3 秒，意味着故障恢复时最多回退 3 秒的数据，下游如果没做幂等，这 3 秒的数据会被重放。对 Doris 这种主键模型表来说重放是覆盖，无害；对只追加的日志表来说就是重复行。这个取舍后面还会再碰一次。
3. Source 的三种写法：一套 connector，三种抽象层级
3.1 五个启动档位，决定的是「从哪儿开始读」
MySqlSource的 builder 是 Flink CDC 最原始也最完整的入口。它的参数分两类：连接信息（hostname、port、username、password、databaseList、tableList）和读取策略（startupOptions、deserializer、server-id、server-time-zone 等）。
MySqlSource<String> mySqlSource = MySqlSource.<String>builder() .hostname("node02") .port(3306) .databaseList("test") // 库级过滤，支持正则 .tableList("test.t1") // 表级过滤，必须写 库名.表名 .username("root") .password("******") // SourceRecord 反序列化成 JSON，便于直接 print 或写下游 .deserializer(new JsonDebeziumDeserializationSchema()) // 首次启动先做全量快照，之后续读 binlog .startupOptions(StartupOptions.initial()) .build(); DataStreamSource<String> mysqlDS = env.fromSource( mySqlSource, WatermarkStrategy.noWatermarks(), "MySQLSource"); mysqlDS.print();databaseList和tableList是取交集的关系，最终捕获范围是两者的交集；tableList不带库名前缀会直接报错，这是个很常见的低级翻车点。server-id不显式指定时连接器会自动生成，但如果你在同一套 MySQL 上跑多个同步任务，就必须手动划分区间，否则会互相踢下线。
StartupOptions的五个档位决定了「第一次启动的行为」，也决定了运维时能不能吃后悔药：
| 档位 | 是否做全量快照 | 起始读取位置 | 典型场景 |
|---|---|---|---|
initial | 是 | 快照完成后从当前位点续读 | 默认选择，新任务接管存量数据 |
earliest | 否 | 从 binlog 最早位点开始 | 日志保留期内可完整重放，做数据修复 |
latest | 否 | 从 binlog 末尾开始 | 只关心启动之后的变更 |
specificOffset | 否 | 指定 binlog 文件与 offset | 精确重置到某个已知位点 |
timestamp | 否 | 从指定时间戳之后开始 | 按时间点补数据 |
earliest和timestamp都把「一致性」的责任交给了 Binlog 保留策略。如果 binlog 已经被清理，连接器会在启动时报位点不存在的错误，而不是悄悄从一个错误的起点开始——这个「快速失败」的设计比静默降级好得多。
3.2 FlinkSQL DDL：同一套 Source 的另一层皮
FlinkSQL 方式把 Source 的构建从 Java 代码挪到了 DDL 里，底下调用的还是同一个MySqlSource，只是把参数翻译成了 with 子句。它的价值在于让同步任务可以用纯 SQL 描述，和下游的SELECT、JOIN、聚合写在同一个环境里。
CREATE TABLE t1 ( id STRING, name STRING, PRIMARY KEY (id) NOT ENFORCED -- 声明主键，决定 changelog 的 upsert 语义 ) WITH ( 'connector' = 'mysql-cdc', 'hostname' = 'node02', 'port' = '3306', 'username' = 'root', 'password' = '******', 'database-name' = 'test', 'table-name' = 't1' ); SELECT * FROM t1;PRIMARY KEY ... NOT ENFORCED这行是整段 DDL 里最容易被忽略、又最关键的一行。它告诉 planner：这张表有主键，但别去数据库里校验。有了主键声明，变更流才能被解释成+I / -U / +U / -D四种标准 changelog 消息，下游做聚合或写入支持 upsert 的 Sink 时才能正确合并。不写主键，流会被当成纯 append，UPDATE 会变成两条互不相干的插入，结果就是目标端数据翻倍。
NOT ENFORCED本身也说明了一个事实：CDC 场景下的主键约束是由源库保证的，Flink 只做语义标注，不做运行时校验。这个分工很合理，如果 Flink 在每条变更上都去校验一次主键唯一性，那条链路就没法要了。
3.3 Pipeline：3.0 把「搭作业」变成了「写配置」
DataStream 和 FlinkSQL 都需要你懂 Flink 的运行时概念：环境、并行度、checkpoint、状态后端。而实际做数据同步的人，很多时候只关心三件事：从哪读、写到哪、库表怎么对应。3.0 引入的 pipeline 框架就是冲着这个来的——它把 source 和 sink 抽象成两个可插拔的 connector，中间的数据流转、schema 演进、checkpoint 由框架统一接管。
source: type: mysql hostname: node01 port: 3306 username: root password: "******" tables: test\.\.* # 正则匹配库表 server-id: 5400-5404 # 手动划分区间，避免多任务冲突 server-time-zone: UTC+8 # 与 MySQL 会话时区对齐 sink: type: doris fenodes: node01:7030 username: root password: "******" table.create.properties.light_schema_change: true table.create.properties.replication_num: 1 pipeline: name: Sync MySQL Database to Doris parallelism: 1这份 YAML 的信息密度很高。server-id: 5400-5404划了一个区间而不是单个值，是因为并行快照会为每个读取任务分配一个独立的 server-id，用区间才能容纳并发连接。light_schema_change: true让 Doris 在源表加列时能用轻量 schema change 完成变更，而不是走昂贵的整表 rewrite。parallelism在 pipeline 层统一设定，source 的分片读取和 sink 的写入并发都受它影响。
框架化的收益是启动方式变得极简，一条命令拉起整个同步作业：
# 将 mysql 与 doris 的 pipeline connector 放入 lib 目录后执行 bin/flink-cdc.sh job/mysql-to-doris.yaml代价是灵活性收窄。你没法在 pipeline 中间插入一段自定义的过滤逻辑，也没法像 DataStream 那样精细控制每个算子的并行度。所以选型上有个朴素的原则：字段级转换和复杂清洗走 FlinkSQL 或 DataStream，纯搬运走 pipeline。
4. Doris Sink：变更流落到 OLAP 表时发生了什么
4.1 写入路径：攒批、Stream Load 与主键覆盖
Doris 不是消息队列，没有「订阅」这个概念，所有写入最终都归结为导入。CDC Sink 的做法是把上游的 changelog 在内存里攒成批，通过 Doris 的 Stream Load 接口一次性推给 FE，再由 FE 分发到 BE 落盘。
这中间有一个语义转换：Binlog 里的-U（更新前）和+U（更新后）两条记录，在 Doris 侧要合并成一次主键覆盖。这依赖 Doris 表的模型选择——必须是 Unique Key 模型，才能让相同主键的新行覆盖旧行。如果用 Aggregate 模型或 Duplicate 模型，UPDATE 会变成插入新行，DELETE 更是完全无法表达。这是选型阶段就要定下来的事，事后改模型要重建表。
攒批的大小直接影响两个指标：批次太小，每次 Stream Load 都有 RPC 和事务开销，导入频率高但吞吐上不去，还会在 Doris 侧积累大量小版本，触发后台 compaction 压力；批次太大，端到端延迟被拉长，故障时重放的窗口也变大。常见的做法是把攒批时间和攒批条数做成双阈值，谁先到就触发哪一路。
4.2 建表参数：light_schema_change 与 replication_num
同步链路刚开始跑的时候最容易被忽略的是「目标表从哪来」。pipeline 支持自动建表，源端有什么结构就在 Doris 里建一张对应的表，这就引出了两个参数。
light_schema_change打开后，加列减列走的是轻量级元数据变更，毫秒级完成。如果关掉，Doris 会走 schema change 的老路径，对大表来说可能是一次小时级的任务，期间表不可写。CDC 场景下源表结构变化是常态，这个开关基本属于必开项。
replication_num是副本数。单机部署环境必须设成 1，否则建表会因为「BE 数量不足以满足副本数」直接失败。这个错误信息在日志里不算显眼，任务启动几秒后退出，很容易被误判成连接问题。
注意：自动建表只覆盖结构，不覆盖分区分桶策略。业务表如果是按时间分区的，Doris 侧默认建出来的表很可能没有分区，数据量上来之后查询会很难受。生产上一般是先手工建好目标表，再让 pipeline 往里写。
4.3 路由变更：一份 source 拆成多路 sink
真实场景里很少是「一个库对一张表」。更常见的是源端有test和test_route两个库，目标端希望它们落在不同的 Doris 库里，或者把多张分表合并成一张目标表。路由规则就是干这个的。
route: - source-table: test_route\.t1 sink-table: doris_test_route.t1 - source-table: test_route\.t2 sink-table: doris_test_route.t2路由在 pipeline 内部是一次数据流的重新分配：匹配到不同sink-table的记录被分到不同的下游写入分支，每个分支独立攒批、独立建表。这意味着目标表的数量会放大写入端的并发压力，路由规则写得越碎，Doris 侧同时活跃的导入任务越多。做分库分表合并的时候要特别注意这一点——上游 64 张分表路由到 1 张目标表是收敛，压力可控；反过来的扇出就要算一下 BE 能不能扛住。
5. 排查清单：这几处配置错一次就得重跑全量
5.1 采集端：binlog、server-id 与时区
现象：任务启动后报「Cannot read binlog」或直接读到空数据。原因是 MySQL 没开 Binlog，或者binlog_format不是row。statement 模式下 Binlog 记录的是 SQL 语句而不是行镜像，连接器拿不到 before/after 值，反序列化会失败。解决方式是修改my.cnf，把log-bin、binlog_format=row、binlog-do-db配好，然后重启 MySQL——这几个参数都是静态参数，改完不重启不生效，配置文件改了没重启是最常见的自欺欺人。
现象：两个同步任务同时跑，其中一个频繁断连重连。原因是server-id冲突。MySQL 主从协议里每个消费者必须用唯一 id 向主库注册，重复注册会互相挤掉。解决方式是给每个任务显式划分不同区间，比如任务 A 用5400-5404，任务 B 用5405-5409。区间宽度至少要不小于该任务的 source 并行度。
现象：同步过去的时间字段普遍差 8 小时。原因是server-time-zone没配，或者配的值和 MySQL 会话时区不一致。连接器在解析 datetime 字段时依赖这个参数做换算，缺少它就会按 UTC 处理。解决方式是在 source 配置里显式写出server-time-zone: UTC+8，并且确认 Doris 侧的字段类型和时区语义与源端对齐。
5.2 运行期：checkpoint、并行度与目标表
现象：任务重启后从头开始扫全表，几小时的快照重跑一遍。原因是没有开启 checkpoint，或者虽然开了但没配置外部化保留策略。前面讲过，位点存在 state 里，state 随 checkpoint 存活，checkpoint 不落盘就等于没有进度。解决方式是开启 checkpoint 并配置RETAIN_ON_CANCELLATION，同时确认状态存储路径（HDFS 或对象存储）在作业重启后依然可访问。从 Savepoint 恢复启动也是同理：flink run -s <savepoint路径>带上位点，才能跳过全量。
现象：Doris 建表失败，任务启动几秒后退出，日志里只有一行 BE 相关提示。原因是replication_num大于可用 BE 数量。单节点环境默认副本数是 3，只有 1 个 BE 时建表必然失败。解决方式是在 sink 的table.create.properties里显式指定replication_num: 1，或者先把 BE 扩容到满足副本数。
现象：全量阶段跑得很慢，但 CPU 和网络都没打满。原因是 chunk 切分粒度和并行度不匹配。并行度设成 1 时，无论表有多大都是一个线程顺序读 chunk；并行度设得远大于 chunk 数量时，多出来的 subtask 会空转。解决方式是先看任务里的 split 分配情况，把并行度调整到与 chunk 数量同一量级，同时留意源库的连接数上限——每个并发读取任务都会占一个 MySQL 连接。
6. 设计取舍：一致性、并行度与运维成本没法同时拉满
这套链路跑稳之后，回头看的其实是几个不可兼得的选择。
第一个取舍：checkpoint 间隔。调小到 1 秒，故障回退的数据少，但每次 CK 都要把分片进度和位点刷到 HDFS，作业的稳定吞吐会被拖下来；调到 30 秒，吞吐好看，故障时最多重放 30 秒。多数场景 3 到 5 秒是个甜点，前提是下游对重复数据幂等。Doris 的 Unique Key 模型天然幂等，这条就变得很便宜——这也是为什么「MySQL 到 Doris」这个组合在实践中比「MySQL 到只追加的日志表」省心得多。
第二个取舍：并行度。并行快照的收益来自把大表切开，但并行的上限由源库的连接数和 BE 的写入能力共同决定。我见过把并行度从 1 调到 8 之后全量时间缩短 5 倍、却在增量阶段把源库连接池打满的案例。并行度不是越大越好，它有一个由外部资源决定的隐性天花板。
第三个取舍：pipeline 与手写作业。pipeline 省掉了环境搭建和状态管理，但把中间处理逻辑挡在了外面。凡是需要字段脱敏、维度补全、多流 join 的场景，还是得回到 FlinkSQL 或 DataStream，自己承担 checkpoint 和状态后端的配置责任。
| 维度 | DataStream | FlinkSQL | Pipeline（3.0） |
|---|---|---|---|
| 上手成本 | 高，需懂运行时 | 中，需懂 DDL 语义 | 低，写 YAML |
| 中间处理能力 | 完全自由 | 支持 SQL 表达的逻辑 | 仅路由与结构调整 |
| 状态与 CK 管理 | 手动配置 | 手动配置 | 框架托管 |
| 适合的场景 | 复杂清洗、自定义算子 | 单流转换、轻量聚合 | 库表级纯搬运 |
验证一套配置是否真的可靠，我的习惯是做一次「断电演练」：全量跑到一半手动 kill TaskManager，观察恢复后是从断点续跑还是从头再来；增量阶段改一条记录、删一条记录、给源表加一列，然后去 Doris 里逐条核对结果和表结构。这三步做完，链路的真实行为就暴露得差不多了，比看十遍文档管用。
从那以后我每次配 CDC 任务，都强制先跑一遍这个演练再上生产——因为位点和状态这类东西，不出事的时候完全感觉不到它的存在，出事的时候又没有后悔药可吃。希望帮到你。
本文还有配套的精品资源，点击获取
