Flink CDC 同步 MySQL 到 Doris：增量快照与断点续传

简介面向具备一定 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 等。MySqlSourceString mySqlSource MySqlSource.Stringbuilder() .hostname(node02) .port(3306) .databaseList(test) // 库级过滤支持正则 .tableList(test.t1) // 表级过滤必须写 库名.表名 .username(root) .password(******) // SourceRecord 反序列化成 JSON便于直接 print 或写下游 .deserializer(new JsonDebeziumDeserializationSchema()) // 首次启动先做全量快照之后续读 binlog .startupOptions(StartupOptions.initial()) .build(); DataStreamSourceString 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 时才能正确合并。不写主键流会被当成纯 appendUPDATE 会变成两条互不相干的插入结果就是目标端数据翻倍。NOT ENFORCED本身也说明了一个事实CDC 场景下的主键约束是由源库保证的Flink 只做语义标注不做运行时校验。这个分工很合理如果 Flink 在每条变更上都去校验一次主键唯一性那条链路就没法要了。3.3 Pipeline3.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: UTC8 # 与 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_formatrow、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: UTC8并且确认 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 和状态后端的配置责任。维度DataStreamFlinkSQLPipeline3.0上手成本高需懂运行时中需懂 DDL 语义低写 YAML中间处理能力完全自由支持 SQL 表达的逻辑仅路由与结构调整状态与 CK 管理手动配置手动配置框架托管适合的场景复杂清洗、自定义算子单流转换、轻量聚合库表级纯搬运验证一套配置是否真的可靠我的习惯是做一次「断电演练」全量跑到一半手动 kill TaskManager观察恢复后是从断点续跑还是从头再来增量阶段改一条记录、删一条记录、给源表加一列然后去 Doris 里逐条核对结果和表结构。这三步做完链路的真实行为就暴露得差不多了比看十遍文档管用。从那以后我每次配 CDC 任务都强制先跑一遍这个演练再上生产——因为位点和状态这类东西不出事的时候完全感觉不到它的存在出事的时候又没有后悔药可吃。希望帮到你。本文还有配套的精品资源点击获取