简介面向具备一定 Flink 基础、希望打通实时同步链路的研发人员这份文档围绕 Flink CDC 3.0 讲解从概念到落地的完整流程。内容先厘清 CDC 的变更捕获思想对比基于查询与基于 Binlog 两种方式的执行模式、覆盖范围、延迟和数据库压力差异再引出 flink-cdc-connectors 对 MySQL、PostgreSQL 等数据源的全量与增量读取能力。案例从 MySQL 建库建表和数据准备入手分别演示 DataStream 与 Flink SQL 写法并给出 MySQL 到 Doris 的 Streaming ETL 实操涵盖环境搭建、配置编写、任务启动与测试验证同时强调开启 Binlog、设置检查点等关键配置。资源包为 1 个 docx约 145KB已有 1789 人学习适合按章节跟做理解各配置项作用进而独立完成数据源连接、流处理与目标存储对接。1. MySQL 到 Doris 的秒级同步为什么第一次启动要先扫全表在 MySQL 里 UPDATE 一行Doris 里几秒后就能查到新值用 Flink CDC 搭这条链路核心代码不到三十行。但真正反直觉的地方不在“监听”而在第一次启动任务并没有从 binlog 开头读起它先把整张表扫了一遍扫完才无缝接进 binlog 流。全量和增量在同一条数据流里接上了中间没有锁表也没有丢一条变更。这套“快照 增量 位点对齐”的组合拳才是 MySQL 到 Doris 实时同步能做到秒级又不重不漏的基础。搞懂它你才知道为什么有的同步任务重启一次就多出几万条重复数据有的却能精确续跑也才知道binlog_formatrow、server-id这类看起来无关紧要的配置为什么能决定整条链路能不能工作。下面按“CDC 两条路线 → binlog 物理结构 → 快照位点对齐 → Source 内部实现 → 3.0 Pipeline → 踩坑排查”的顺序拆。2. CDC 的两条路线为什么 binlog 能同时拿到全量和增量CDC 这个词听起来新做数据同步的人其实早就在干这事。最朴素的办法是查定时跑一条select * from t where update_time ?把增量捞出来写进目标端。这条路线的代表是各种批同步工具几百万行的表几分钟能过一遍不算慢。问题是它有几个绕不过去的坎每一个都能在真实场景里出问题。2.1 “基于查询”的路线为什么会出问题第一个坎是删除。物理删除之后那行数据在源表里彻底消失了靠select永远查不出来。要捕获 delete要么把业务改造成软删除在表里加is_deleted字段要么定期做全量比对成本立刻上去一个量级。很多团队一开始图省事选择忽略删除跑了半年才发现目标端的“已删除用户”还在源源不断进报表。第二个坎是时间列的可信度。不少表压根没有update_time有的表有但值是应用侧传进来的应用服务器和数据库的时钟差几秒边界上的数据就漏了。这种漏数据极其隐蔽——任务日志一片正常几个月后月底对账才发现在某个时间点少了那么几十条。我一般会这么理解基于查询的 CDC 把一致性责任推给了业务表的字段设计而这个设计往往不受数据团队控制。第三个坎是轮询本身的代价。10 张表按 1 秒轮询一次就是 10 个 QPS 的索引扫描表大了索引扫不动直接退化成全表扫。业务库的 IO 本来就紧张再挂一个同步任务上去血泪经验基本都来自这里。第四个坎更根本批模式天然做不到秒级延迟下限就是你设定的轮询间隔。维度基于查询的 CDC基于 binlog 的 CDC执行模式批Batch流Streaming能否捕获删除不能物理删除后行已消失能DeleteRows 事件是否依赖业务字段依赖 update_time 之类的列不依赖直接看日志变化延迟取决于轮询间隔通常分钟级毫秒到秒级对源库压力每次轮询都是一次查询一条长连接持续读日志实现复杂度低高要解析日志并维护 schema基于 binlog 的路线把视角从“表里现在有什么”换成了“数据库刚刚发生了什么”。视角换了以后上面几个坎同时消失了删除变成一条日志事件不需要时间列也不需要轮询。代价是复杂度转移到解析端——你得能把二进制日志还原成结构化的行变更还得在源表结构变化时跟得上。2.2 binlog 的物理结构一次 UPDATE 到底写了什么binlog 是 MySQL server 层维护的逻辑日志记录的是“数据变成什么样”而不是“磁盘哪一页改了”——后者是 InnoDB 的 redo log 负责的事。这个区别很关键redo log 是物理的、循环写的、格式和存储引擎绑定binlog 是逻辑的、追加写的、可以被任何外部程序解析。CDC 能成立前提就是 binlog 是逻辑的且可读的。binlog 有三种格式CDC 场景下只能选一种STATEMENT记录 SQL 原文。问题是NOW()、UUID()、不带ORDER BY的LIMIT这类语句重放结果不确定而且 CDC 端拿到 SQL 原文也还原不出 before image下游根本不知道“改之前是什么”。ROW记录每一行变更前后的值。CDC 必须要这个格式。MIXED由 MySQL 自己决定用哪个对 CDC 来说等同于赌博。ROW格式下一次事务在 binlog 里大致长这样GTID_EVENT或QUERY_EVENT(BEGIN)事务开始标记TABLE_MAP_EVENT把“表 ID”和“这张表的列定义”建立映射WRITE_ROWS_EVENT/UPDATE_ROWS_EVENT/DELETE_ROWS_EVENT真正的行变更带列序号和值XID_EVENT事务提交。这里有个容易被忽略但后果严重的推论行事件里不含列名只有列序号。解析端必须自己能还原出“第 3 列叫什么、是什么类型”而这个还原依赖TABLE_MAP_EVENT里那份列定义快照。一旦源表ALTER了列序号整体位移解析端如果还用旧映射读出来的数据就整体错位——这就是 DDL 变更历来是 CDC 链路上最容易出问题的地方。再补一个参数binlog_row_image。取FULL默认时 before 和 after 都记全取MINIMAL时只记主键和被改的列日志体积能小一大截但 CDC 端拿不到完整的 before image遇到“只更新某一列”的语义就丢了。生产环境如果为了省空间调成MINIMAL同步链路的行为会悄悄变化。# 在源库机器上直接解析 binlog 文件 # -v 把行事件翻译成可读的伪 SQL--base64-outputDECODE-ROWS 让 ROW 事件以明文输出 mysqlbinlog -v --base64-outputDECODE-ROWS \ --start-position4 /var/lib/mysql/mysql-bin.000003 | head -n 60-v出现一次是翻译行事件出现两次-vv会额外打出列类型--start-position从指定位点开始用来复现某一段不加--base64-output会看到一堆 Base64 的行事件读不了。排查“CDC 到底有没有读到这条变更”时第一步就是在源库上确认这条变更确实进了 binlog——这一步能直接排除掉一半的“下游没数据”问题。2.3 快照与增量的衔接位点是怎么对齐的全量和增量要接上传统做法有三种演进过程本身就是一部踩坑史。第一种最老FLUSH TABLES WITH READ LOCK拿全局读锁记录SHOW MASTER STATUS的位点全量 dump再释放锁然后从那个位点读 binlog。问题是锁表期间业务完全写不进去几百 G 的表 dump 几个小时就是几个小时的写入阻塞。这个方案在稍大的库上基本不可用。第二种在 RR 隔离级别下START TRANSACTION WITH CONSISTENT SNAPSHOT不开锁靠 MVCC 拿一个一致性快照事务期间读到的都是快照时刻的数据同时记下位点。这个方案能跑但有两个隐患长事务会阻止 undo purge大表 dump 几个小时能把 undo 表空间撑爆而且它是单线程的一张大表只能一个会话慢慢读全量时间下不来。第三种是 Flink CDC 用的增量快照核心思路分三步切分。按主键把表切成若干 chunk比如id BETWEEN 1 AND 10000、10001 AND 20000chunk 大小可配。双读并行。一边用SELECT * FROM t WHERE id BETWEEN ? AND ?读 chunk一边保持 binlog 流不断两条路同时走。水位对齐。每个 chunk 开读前记下当前 binlog 位点low watermark读完再记一个high watermark。chunk 读完后把落在[low, high]区间内的 binlog 事件重放一遍如果某行主键已经在 chunk 里读过就以 binlog 事件为准覆盖如果主键不在本 chunk 范围内说明是别的 chunk 的事跳过。最终效果是一条记录在快照和增量里最多被处理两次但经过水位对齐后结果等价于“在某个时间点做了一次全量然后从那个点开始严格按顺序应用增量”不重不漏。这套机制换来的好处很实在不锁表业务无感chunk 可以并行10 个 reader 同时读 10 个 chunk大表全量时间线性下降失败恢复粒度细到 chunkcheckpoint 里记的是“哪些 chunk 已完成”重启只重跑没做完的。代价也要说清楚必须有主键或唯一非空键没有主键的表切不了 chunk只能退化成单并行度全表扫binlog 事件要在状态里缓存一段时间chunk 读得越慢缓存越大水位对齐本身有开销chunk 切得太碎反而拖慢。我一般会这么理解这套设计——它把“一致性快照”这件事从数据库侧锁、长事务挪到了计算侧水位、重放数据库只负责不停地吐日志一致性的保证交给流处理引擎的状态来兜。3. MySqlSource 的内部结构StartupOptions 与状态存储理解了上面的机制再去看 Flink CDC 暴露出来的那几个配置项就不像是在背诵参数了——每一个都对应着上面某个环节的取舍。3.1 StartupOptions 的五个模式语义差在哪MySqlSourceString mySqlSource MySqlSource.Stringbuilder() .hostname(node03) .port(3306) .databaseList(test) // 捕获的库支持正则 .tableList(test.t1) // 捕获的表格式必须是 db.table .username(root) .password(000000) .deserializer(new JsonDebeziumDeserializationSchema()) .startupOptions(StartupOptions.initial()) // 启动位点策略 .build();databaseList和tableList都支持正则但tableList的每一项必须是库名.表名的完整形式只写表名匹配不上。deserializer决定SourceRecord怎么变成下游要的类型后面单独讲。startupOptions是最容易选错的一项五种模式的语义差别很大模式首次启动行为会重跑全量吗典型场景风险initial先做一致性快照再从快照位点读 binlog有 checkpoint 时不重跑首次初始化目标端为空大表全量耗时长占带宽earliest不做快照从当前最早的 binlog 位点开始否目标端已有历史数据只补日志binlog 可能已被 purge位点前的表结构可能已变latest不做快照从 binlog 末尾开始否只要新变更历史数据已离线导入启动之前的变更全部丢失specificOffset不做快照从指定的 file pos 开始否精确复现某一段或从备份位点续位点算错就丢数据或重复timestamp不做快照从指定时间戳开始遍历 binlog 跳过更早事件否按时间点回补需要从头遍历日志日志大时启动很慢earliest是最常被误解的一个。它不做快照——语义是“我假设目标端已经有历史数据了只把日志从头补一遍”。如果目标端是空的却用earliest启动最后读到的只有 binlog 里还留着的那部分变更更早的历史数据永远补不回来。而 binlog 是有保留期的过期就被 purge 了earliest实际能拿到的起点往往比你想象的晚得多。initial之所以是默认推荐是因为它把“全量”和“增量”这两件事在语义上合并成了一次完整的同步。但要注意它只在没有可用 checkpoint 的时候才走全量。一旦有了状态重启会直接从状态里的位点续不会重来一遍。3.2 binlog 位点是怎么被 checkpoint 存下来的Flink CDC 的 Source 是按 FLIP-27 的 Source API 实现的分两个角色SplitEnumerator跑在 JobManager 上负责切分和分配SourceReader跑在 TaskManager 上负责实际读数据。这个拆分是 checkpoint 能工作的前提——枚举器知道全局还有哪些 split 没做完reader 知道自己手上这个 split 读到哪了两边各自往状态里写。split 分两类快照 split一个 chunk 一个状态是“主键上界 是否完成”和增量 split只有一个状态是 binlog 位点包含文件名、position、GTID set。checkpoint 触发时枚举器存“未分配的和已分配未完成的 split 列表”reader 存“当前 split 的进度”。恢复时枚举器从状态重建把没完成的 split 重新下发reader 从记录的位点继续——已完成的 chunk 一个都不会重读。StreamExecutionEnvironment env StreamExecutionEnvironment.getExecutionEnvironment(); env.setParallelism(1); // 关键不开 checkpointbinlog 位点就没地方存重启只能从头再来 env.enableCheckpointing(3000L, CheckpointingMode.EXACTLY_ONCE); env.getCheckpointConfig().setCheckpointTimeout(60 * 1000L); // 两次 checkpoint 之间至少隔 3 秒避免上一次没做完下一次又压上来 env.getCheckpointConfig().setMinPauseBetweenCheckpoints(3000L); // 任务被 cancel 时保留最后一次成功的 checkpoint否则状态跟着作业一起没了 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://node02:8020/flinkCDC);EXACTLY_ONCE在这里的作用不只是笼统的“数据不重复”——它保证的是 barrier 对齐也就是“状态里记的 binlog 位点”和“已经向下游发出的数据”处于同一时刻。改成AT_LEAST_ONCE后 barrier 不对齐恢复时位点可能比实际已发出的数据靠前重复就来了。RETAIN_ON_CANCELLATION是最容易漏的一项。默认行为是作业 cancel 时把 checkpoint 一起删掉。如果你是手动 cancel、改个配置再启状态没了任务只能按startupOptions重新来一遍——initial就是重跑全量。所以规范流程是先打 savepoint 再 cancel# 先触发 savepoint把当前 binlog 位点固化到指定路径 bin/flink savepoint JobId hdfs://node02:8020/flinkCDC/save # 然后从 savepoint 恢复而不是从零开始 bin/flink run -s hdfs://node02:8020/flinkCDC/save/savepoint-xxxx \ -c com.example.cdc.FlinkCDCDataStreamTest ./flink-cdc-test.jarfailureRateRestart(3, Time.days(1L), Time.minutes(1L))的语义是一天之内失败次数不超过 3 次就继续重启每次重启前等 1 分钟。这个策略适合 CDC 场景——源库短暂抖动、网络闪断这类问题重启就能好但如果是配置错误3 次之后作业会彻底失败不会无限重启把集群拖垮。3.3 反序列化SourceRecord 到 JSON 的那一层Debezium 把 binlog 事件解析成SourceRecord这是 Kafka Connect 的数据结构包含 key、value、topic、partition、sourceOffset。默认的JsonDebeziumDeserializationSchema把它转成 JSON结构大致是这样{ before: null, after: { id: 1001, name: zhangsan }, source: { db: test, table: t1, file: mysql-bin.000003, pos: 154, p a hrefhttps://download.csdn.net/download/zzl8888888/91240170 stylecolor:#ec7500;font-size:14px; 本文还有配套的精品资源点击获取 /a img altmenu-r.4af5f7ec.gif srchttps://csdnimg.cn/release/wenkucmsfe/public/img/menu-r.4af5f7ec.gif stylewidth:16px;margin-left:4px;vertical-align:text-bottom;cursor:text; /p