企业官网建设流程全解析

简介：面向具备一定 Flink 基础、关注实时数据同步与 Streaming ETL 的研发人员，这份尚硅谷研究院编写的 Flink CDC 3.0 文档聚焦 MySQL 变更数据实时捕获并同步至 Doris 的链路。内容从 CDC 概念切入，对比基于查询与基于 Binlog 两种方式，讲解 flink-cdc-connectors 组件，并给出数据准备、DataStream 与 Flink SQL 路径，以及 MySQL 到 Doris 的环境搭建、配置编写、任务启动与测试步骤。包内为 1 个 docx 文件，约 145KB，便于集中查阅和跟练。已有 1789 人学习，读者可理解 Binlog 开启、检查点设置等关键配置，掌握全量与增量同步思路，对照案例完成源端连接、目标存储对接与参数调整。

1. 从一条 UPDATE 说起：为什么轮询查表永远追不上日志流

业务库里一条UPDATE落地，多久之后数仓里能查到？如果你的答案是「五分钟跑一次批」，那这条语句背后的 Flink CDC、MySQL Binlog、Doris 实时同步链路基本和你无关。真正反直觉的地方在于：把变更抓出来这件事，瓶颈从来不在下游算得多快，而在上游有没有一个「变更发生时就主动告诉你」的通道。轮询查表默认没有，它只能靠update_time或者全表比对去猜，猜的代价是延迟、是数据库压力，也是删不掉的那部分数据——物理删除在表里连影子都留不下。

Flink CDC 3.0 把这条链路做成了配置：MySQL 侧读 Binlog，Flink 侧承载流与状态，Doris 侧接 Stream Load 落地，中间靠 YAML 描述一份 Streaming ETL 作业。搞懂它的价值不在于会敲几条命令，而在于你能回答：位点存在哪、快照怎么不打锁、并行度为什么常常是 1、出问题时该看哪个日志。下面按「日志从哪来 → 快照怎么拍 → 位点怎么活 → 怎么翻车 → 为什么这么设计」的顺序拆。

2. Binlog 伪装术：一条 INSERT 是怎么在毫秒内变成 Flink 里的一条流的

2.1 从库协议：CDC 客户端把自己伪装成一个 Slave

MySQL 的 Binlog 不是给外部读的，它本来是给复制用的。主库上有个dump线程，从库连上来时发一条COM_BINLOG_DUMP命令，带上自己的server-id、起始 binlog 文件名和位点，主库就开始不停地往后推事件，推到没数据了就挂着等。CDC 工具做的事情一模一样：它注册成一个伪从库，说「我是 server-id 5401 的从库，从 mysql-bin.000003 的 position 154 开始给我发」。

这个设计决定了三件事。第一，读的是磁盘上已经写好的 binlog 文件，顺序 IO，对主库几乎只有网络和内存开销，不像SELECT那样要占 buffer pool、要走执行器。第二，延迟是毫秒级的，因为 binlog 是事务提交时落盘的，主库推、客户端收，中间没有调度周期。第三，服务端必须开binlog_format=row——statement 格式只记录 SQL 文本，UPDATE t SET name='x' WHERE id>100这种语句在 CDC 侧根本还原不出每一行到底改成了什么，只有 row 格式才带 before/after 镜像。

配置部分对应到my.cnf里是这样几行：

# 每个复制实例必须唯一，CDC 客户端会用它伪装成从库 server-id = 1 # 开启 binlog 并指定文件前缀 log-bin = mysql-bin # 必须是 row，statement 格式拿不到行级前后镜像 binlog_format = row # 只对业务库开 binlog，减少无关写入 binlog-do-db = test binlog-do-db = test_route

server-id是最容易被低估的参数。它不是「随便填一个」，而是全局唯一的复制身份，一旦和已有从库或者另一个 CDC 任务撞了，主库会直接踢掉旧连接。后面讲并行读取时会看到，Flink CDC 甚至需要一段server-id区间，因为每个并行 reader 都要有自己的身份。

2.2 从行事件到 SourceRecord：Debezium 在中间做了什么

Binlog 事件本身是二进制的TableMapEvent+WriteRowsEvent/UpdateRowsEvent/DeleteRowsEvent，字段类型编码、表结构映射都得自己解析。这块轮子 Debezium 已经造好了：它维护一份表结构快照，把行事件翻译成统一的SourceRecord，里面保留了before、after两个字段镜像，加上source元信息（库名、表名、binlog 文件名、位点、gtid）和op操作类型。

Flink CDC 的 MySQL Source 就是在 Debezium 之上包了一层 Flink 的Source接口。JsonDebeziumDeserializationSchema负责把SourceRecord序列化成 JSON，所以你在下游看到的每条数据长这样：

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

op是理解整条链路语义的钥匙：c是插入、u是更新、d是删除、r是快照阶段读出来的存量数据。很多下游同步出错，根源就在于把r和c当成了一回事——快照数据并不是「刚刚发生的插入」，它的ts_ms是读取时刻而不是业务发生时刻，拿它做时间窗口统计会直接错位。

2.3 用 DataStream 把 Source 拼起来

先看第一代写法，也就是直接用 DataStream API 构建 Source。虽然 3.0 主推 YAML，但这条路径是理解内部结构的唯一入口：

// 1. 环境与并行度：单表 binlog 天然有序，先设 1 避免乱序 StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); env.setParallelism(1); // 2. 检查点：CDC 的位点是以状态形式存的，不开 CK 就没有断点续传 env.enableCheckpointing(3000L, CheckpointingMode.EXACTLY_ONCE); env.getCheckpointConfig().setCheckpointTimeout(60 * 1000L); env.getCheckpointConfig().setMinPauseBetweenCheckpoints(3000L); env.getCheckpointConfig().enableExternalizedCheckpoints( CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION); // 3. 构建 Source：库表、账号、反序列化器、启动模式 MySqlSource<String> mySqlSource = MySqlSource.<String>builder() .hostname("node01") .port(3306) .databaseList("test") .tableList("test.t1") .username("root") .password("******") .deserializer(new JsonDebeziumDeserializationSchema()) .startupOptions(StartupOptions.initial()) .build(); // 4. 接入流并输出 DataStreamSource<String> mysqlDS = env.fromSource(mySqlSource, WatermarkStrategy.noWatermarks(), "MySQLSource"); mysqlDS.print(); env.execute();

几个参数值得单独说。tableList("test.t1")里必须带库名前缀，只写表名在 3.0 会直接报错，这是从旧版本升级时最常见的翻车点。WatermarkStrategy.noWatermarks()不是省略，是明确声明这条流不产生事件时间水位线——CDC 数据本身的时间语义由ts_ms承载，强行加 watermark 反而会引入乱序等待。

enableCheckpointing(3000L, EXACTLY_ONCE)这一行是整个作业的生命线。Flink CDC 把「读到哪个 binlog 文件、哪个位点」当作算子状态管理，只有 checkpoint 成功落盘，这份状态才存在；作业重启时从最近一次成功的 checkpoint 恢复，位点跟着回来，数据既不重也不丢。不开 checkpoint 的作业在容错上等同于一次性脚本。

注意：RETAIN_ON_CANCELLATION决定了手动 cancel 作业时保留最后一次 checkpoint。没有它，取消即清空，后面想从 Savepoint 之外的路径恢复就没机会了。

3. 快照与增量的夹缝：无锁分片读取为什么是 3.0 的分水岭

3.1 传统快照的代价：一把全局读锁

先说老办法。Debezium 早期做初始快照的逻辑很朴素：开一个可重复读事务，先FLUSH TABLES WITH READ LOCK把表锁住，记录当前 binlog 位点，然后释放锁、在事务里慢慢SELECT全表。锁的作用是保证「位点」和「数据」对得上，代价是锁期间主库写入全部阻塞。

小表无所谓，几千万行的表就不一样了。锁可能只持有一瞬间（因为一致性读靠的是 MVCC 快照而不是持续持锁），但长事务带来的 undo 膨胀、大表全扫带来的 IO 冲击是实打实的。更麻烦的是，一旦快照阶段失败，整个流程要重来，对线上库的扰动是周期性出现的。

这里还有个隐性的结构性问题：全表SELECT只能单线程跑，因为要保证顺序性和位点对齐。这就意味着无论你 Flink 集群有多少 slot，初始阶段就是一个线程在拉数据，几亿行能拉几个小时。

3.2 分片读 + 位点水位线：把「一致性」拆成可并行的小块

3.0 的核心改动叫增量快照（incremental snapshot）。思路是把一张表按主键切成若干 chunk，每个 chunk 记录一个[low, high]主键区间，读取时不是简单地SELECT * WHERE id BETWEEN low AND high，而是在读这个区间的同时监听 binlog，把落在区间内的变更事件缓存下来；chunk 读完时，再拿缓存的变更去修正快照数据，最后把 chunk 标记为完成。

这样做的好处是把「全局一致」降级成「分片内一致」。每个 chunk 独立推进、独立记录进度，天然可以并行；不需要长时间持锁，只依赖主键索引的范围扫描；某一段失败只重跑那一段，不用从头再来。

代价是对表结构有硬要求：必须有主键或者非空的唯一索引，否则切不了 chunk，任务启动就会报找不到分片键。这一点在接入历史遗留表时要提前确认，用SHOW KEYS FROM t1 WHERE Key_name='PRIMARY'看一眼就能排掉大半问题。

并行带来的第二个约束就是前文提到的server-id区间。每个 chunk reader 都要以独立的从库身份连 MySQL，所以配置里写的是server-id: 5400-5404这样的范围，而不是单值。范围太小、并行度又调得大，就会出现「抢不到身份」的启动失败。

3.3 StartupOptions 的五种模式与适用边界

启动模式决定了「首次启动时到底做不做快照、从哪开始读」，选错了要么全量重刷，要么丢掉历史变更：

initial是默认选择，也是绝大多数生产接入的起点。earliest要格外小心：它要求 binlog 里还完整保留着库的全部历史，一旦 binlog 被 purge 过，读到一半就会断，而且补不回来。latest则相反，它主动放弃了存量数据，适合「存量我已经用批处理灌过了，现在只接增量」的场景——这恰恰是大表接入时更稳妥的做法，后文会再提。

代码上就是一行切换：

// 首次接入：先快照再追增量，位点从快照结束点开始 .startupOptions(StartupOptions.initial()) // 存量已用批任务灌完：跳过快照，只消费启动后的变更 // .startupOptions(StartupOptions.latest()) // 指定时间点补数据：从该时间戳之后的 binlog 开始 // .startupOptions(StartupOptions.timestamp(1700000000000L)) // 指定精确位点：file + position，适合定点修复 // .startupOptions(StartupOptions.specificOffset("mysql-bin.000003", 1542L))

同一份代码换一个启动模式，语义完全不同。这个设计其实反映了一个现实：CDC 工具没法替业务判断「你现在处于哪个阶段」，首次接入、补数、追增量是三种不同诉求，只能把选择权交出去，让使用者显式声明。

4. 位点即状态：Checkpoint 如何把「读到哪了」钉死在 Flink 里

4.1 位点为什么必须进状态

如果位点存在外部（比如写一张 MySQL 表记录进度），那么「处理数据」和「记录位点」就是两次独立写入，中间崩了要么重复、要么丢。Flink CDC 的选择是：位点就是算子状态的一部分，和 checkpoint 一起原子落盘。

具体来说，Source 算子在快照阶段记录每个 chunk 的完成情况，在增量阶段记录当前的 binlog 文件名和位点。checkpoint 触发时，这些信息被序列化进状态后端（示例里用的是HashMapStateBackend+ HDFS 存储），checkpoint 完成时才算数。作业失败重启，从最近一次完成的 checkpoint 恢复，Source 从那个位点重新开始读。

这也解释了一个常见疑问：为什么不用外部存储记录位点？因为外部存储没法参与 Flink 的分布式快照，做不到「数据流处理到哪」和「位点到哪」严格对齐。位点一旦脱离状态，exactly-once 就无从谈起。

4.2 端到端一致性的两段：Flink 侧与 Sink 侧

Flink 的 checkpoint 保证的是「状态一致」，也就是算子内部看到的输入不重不丢。但要端到端一致，Sink 端必须配合——这正是两阶段提交（2PC）协议存在的原因：checkpoint 期间数据先写但不提交，等 checkpoint 完成再统一提交，checkpoint 失败就回滚。

Doris 这一侧的落地方式是基于 Stream Load 的两阶段提交，配合 label 的幂等去重：同一个 label 重复导入，Doris 只认第一次。所以即使作业在「数据已写入、checkpoint 未完成」之间崩溃重来，重放的数据带着同样的 label 也不会变成重复行。这个机制是 CDC 到 Doris 这条路能称得上「同步」而不是「广播」的根本原因。

配置层面需要关注的参数如下：

// 检查点间隔：太短会增加 IO，太长意味着恢复时要重放更多 binlog env.enableCheckpointing(3000L, CheckpointingMode.EXACTLY_ONCE); // 单次超时：HDFS 抖动时的保护，超时即失败重来 env.getCheckpointConfig().setCheckpointTimeout(60 * 1000L); // 两次之间的最小间隔：避免上一个还没完下一个又来了 env.getCheckpointConfig().setMinPauseBetweenCheckpoints(3000L); // 取消时保留最后一次 CK，给手动恢复留后路 env.getCheckpointConfig().enableExternalizedCheckpoints( CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION); // 失败率重启：1 天内允许 3 次，间隔至少 1 分钟 env.setRestartStrategy(RestartStrategies.failureRateRestart( 3, Time.days(1L), Time.minutes(1L)));

failureRateRestart(3, Time.days(1L), Time.minutes(1L))的语义容易被读反：它表示在一个自然日窗口内最多允许 3 次失败重启，超过就彻底失败不再拉起。设得太宽松，作业会陷入无限重启的死循环而没人发现；设得太紧，一次网络抖动就把生产任务打挂。经验值是让它在几分钟内能自愈常见故障，又不至于掩盖结构性问题。

4.3 Savepoint 续跑：一条可验证的容错链路

checkpoint 是自动的，Savepoint 是手动的，两者都存位点，区别在于后者由人主动触发、用来做有计划的升级或迁移。验证位点是否真的生效，最直接的方式就是走一遍完整的续跑流程：

# 1. 给运行中的作业打一个 savepoint，注意目标路径要提前规划 bin/flink savepoint <JobId> hdfs://node01:8020/flink-cdc/save # 2. 在控制台取消作业（不要 kill 进程），确保 savepoint 被完整保留 bin/flink cancel <JobId> # 3. 此时在 MySQL 里对 t1 做几条 INSERT / UPDATE / DELETE # 这些变更会落进 binlog，但不会被任何消费者处理 # 4. 从 savepoint 恢复，观察是否只补上了第 3 步的变更、没有重复历史数据 bin/flink run -s hdfs://node01:8020/flink-cdc/save/savepoint-xxxx \ -c com.example.cdc.FlinkCDCDataStreamTest ./flink-cdc-test.jar

如果第 4 步之后看到的输出里，历史数据一条没重、第 3 步的变更一条没漏，位点状态就是真的在工作。这个方法我建议在接入任何新表时都跑一遍——它是唯一能在上线前证明「容错配置不是摆设」的手段。

4.4 FlinkSQL 的声明式写法：主键决定流语义

DataStream 之外的另一条路是 FlinkSQL，DDL 里一个关键细节是PRIMARY KEY NOT ENFORCED：

CREATE TABLE t1 ( id STRING PRIMARY KEY NOT ENFORCED, -- Flink 不做约束校验，但据此推断 upsert 语义 name STRING ) WITH ( 'connector' = 'mysql-cdc', 'hostname' = 'node01', 'port' = '3306', 'username' = 'root', 'password' = '******', 'database-name' = 'test', 'table-name' = 't1' );

NOT ENFORCED不是说主键没用，而是「Flink 不去校验它，但请按它来理解这条流」。有主键声明，Flink 会把变更流当成 upsert 流（+I / -U / +U / -D），下游接 Doris 这类支持主键更新的存储才能正确合并；没声明主键，流会被当成 append-only，同一行的多次更新会变成多条独立记录堆在那里，最终表里全是脏数据。

5. 排查清单：MySQL 到 Doris 链路上最容易翻车的五个地方

5.1 现象：任务启动即报「A slave with the same server_id」

原因几乎总是server-id撞车。它可能是和历史从库撞，也可能是同一份配置起了两个任务，还可能是 YAML 里写死了单值却在 3.0 pipeline 里被默认并行度拉起多个 reader。报错信息里通常会带上当前占用者的 uuid，但要定位到自己配置哪一行冲突，还是得把server-id范围、并行度、已有复制拓扑一起看。

解决办法是把server-id配成区间，比如5400-5404覆盖 5 个 reader；同时核对pipeline.parallelism，让范围至少不小于并行度。改完重启后如果还报，去主库SHOW SLAVE HOSTS看一眼有没有僵死的连接占着号。

5.2 现象：任务重启后目标端出现大量重复行

先查 checkpoint 有没有真的开起来。不开 checkpoint 的作业在恢复时只能从startupOptions指定的位置重新开始，initial会重拍一次快照并再次灌入存量数据，目标端如果没有主键去重，就是一片重复。这个问题在测试环境常常被忽略，因为测试时任务很少重启，一上生产遇到一次 OOM 就全暴露了。

解决思路是双保险：Flink 侧强制开启 checkpoint 并保留外部化 checkpoint；Doris 侧建表用 Unique Key 模型，让重复数据能被覆盖掉。只靠一侧都不够，因为 checkpoint 只是概率上减少重放，Unique Key 才是最后一道兜底。

5.3 现象：目标表里同一主键的最终值和源库对不上

这种「顺序错乱」通常出在并行度上。binlog 对单表的变更是有严格顺序的，但如果 Flink 里把同一张表的数据分发到多个并行子任务，而下游写入又不保证按主键路由，更新就可能乱序落库，最终值取决于哪条先到。典型场景是上游parallelism调成 4，下游 sink 用轮询分发。

稳妥的做法是让同一主键始终走同一个并行实例——按主键 hash 分发，或者干脆把单表链路的并行度设成 1。牺牲一点吞吐换正确性，在 CDC 这种场景里通常是划算的，因为瓶颈一般不在 Flink 的计算上，而在源库的 binlog 读取和目标端的写入批次上。

5.4 现象：Doris 建表失败，提示副本数超过可用 BE 数量

3.0 pipeline 会自动在 Doris 侧建表，建表属性来自 YAML 里的table.create.properties.*。如果replication_num设成 3，而集群里只有 1 个 BE，建表就会失败，任务卡在初始化阶段。测试环境里这个坑几乎人人都踩过，因为默认值是按生产规模设定的。

解决方式是在 YAML 里显式写table.create.properties.replication_num: 1，生产环境再按 BE 数量调整。同一段配置里还有light_schema_change: true，它关系到源库加列时 Doris 侧能不能快速跟进——Doris 的轻量 schema change 只改元数据，加列是毫秒级的，但对历史数据多的表来说，不开启这个选项加列可能引发长时间的数据重写。

5.5 现象：数据都在，但ts_ms整体偏了 8 小时

时区问题几乎不会报错，只会让下游的时间维度统计悄悄错位。根源是 MySQL 服务端时区、CDC 配置里的server-time-zone、Flink 作业运行时的 JVM 时区三者不一致。YAML 里写server-time-zone: UTC+8，表示「按东八区解释源库的时间字段」，但如果 MySQL 实际是 UTC，解析出来的时间戳就整体偏移。

排查方法是拿一条已知写入时间的数据，对比源库的UPDATE_TIME、CDC 输出里的ts_ms换算结果、Doris 里落地的时间列，三者对齐才算过关。固定时区配置、不用默认值，是唯一可靠的规避手段。

提示：排查顺序建议固定为「源库 Binlog 配置 → CDC 位点状态 → Flink checkpoint 记录 → Sink 写入批次」，从上游往下游走。反过来查往往会先怀疑下游，浪费大量时间在没问题的环节上。

6. 取舍：把作业写成 YAML 之后，哪些能力被让渡出去了

3.0 最大的形态变化是把「写代码」换成了「写配置」。DataStream 里那一整段MySqlSource.builder()、反序列化器、sink 构建，全部收进了一份 YAML：

source: type: mysql hostname: node01 port: 3306 username: root password: "******" tables: test.\.* # 正则匹配整库，路由规则另配 server-id: 5400-5404 server-time-zone: UTC+8 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

这份配置解决的问题很清楚：整库同步这种高度模板化的需求，写代码是纯负担——每接一个库要 copy 一遍 builder 链，改一个参数要重新编译打包。配置化的收益是作业变成可版本管理、可复制、可 diff 的文本。

让渡出去的能力同样清楚。一旦需要复杂清洗、多流 join、自定义 UDF，YAML 就无能为力了，只能退回 DataStream。所以真实的选型往往不是二选一，而是分层：整库同步、纯搬运用 pipeline；带清洗逻辑的链路用 DataStream 手写，把 CDC Source 当成一个普通的流式 Source 接进去。

parallelism: 1这个默认值也值得多想一层。它不是一个性能妥协，而是一个正确性默认值——单表 binlog 有序，默认开并行就等于默认给了你一个可能乱序的作业。想调大，前提是你已经想清楚了主键路由和下游合并语义。

还有一类场景根本不该用 CDC 做全量初始化。几亿行的历史数据用 CDC 快照去拉，耗时以小时计，还会长时间占用 binlog 连接和主库 IO；更稳的做法是用批处理工具把存量一次性灌进 Doris，然后 CDC 从latest模式接增量，两条链路在时间上衔接。判断标准很简单：如果存量数据的「一致性时点」要求不高，就别让 CDC 去干批处理的活。

我自己的习惯是，每次配置好一条新链路，先拿一张小表把 savepoint 续跑、加列、删行、时区这四个动作各走一遍，确认位点、schema、时间戳都对得上，再切到业务表。那次漏掉 savepoint 验证、结果上线后重启把整库重灌了一遍的经历，让我把这条检查固化了。希望帮到你。

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