Flink CDC 3.0 实时同步 MySQL 到 Doris 全攻略

简介面向具备一定Flink基础的大数据研发人员这份由尚硅谷研究院编写的技术文档围绕Flink CDC 3.0展开系统讲解CDC基本概念、基于查询与基于Binlog两类方式的差异并重点演示flink-cdc-connectors组件从MySQL等数据库读取全量与增量变更数据的用法。文档以MySQL到Doris的Streaming ETL为完整案例覆盖数据准备、DataStream与Flink SQL两种接入方式、环境搭建、配置文件编写、任务启动与测试验证等全链路步骤同时强调开启Binlog、设置检查点等关键细节便于读者理解各项参数的作用并迁移到实际生产环境。文档按章节组织包体为单个docx文档压缩包大小145KB内容包含可直接跟随操作的完整代码示例与配置说明。目前已有一千七百余人学习下载适合希望独立完成Flink CDC项目部署、实现高效稳定数据同步的工程师。1. Flink CDC 3.0 是什么从一条 MySQL 订单到 Doris 的实时链路做实时大屏或者实时数仓的同学应该都遇到这种需求业务数据在 MySQL 里要同步到 Doris 给下游查询或者做指标计算。以前我常用的办法是离线跑 T1后来业务要求延迟压到分钟级甚至秒级离线导入直接翻车。换过几种方案不是要写一堆 DataX 脚本就是全量初始化时锁表导致业务投诉。直到 Flink CDC 3.0 出现它把 MySQL 的 binlog 捕获、全量快照和增量同步整合成一套统一框架配合 Doris 的 StreamLoad 引擎几分钟就能搭出一条从 MySQL 到 Doris 的实时管道。这篇文章我就把这个方案的完整脉络讲清楚底层原理、最小可跑作业、生产级参数、以及我自己踩过的坑。内容适合正在做实时数仓、准备把 CDC 引入同步链路的数据工程师和开发同学。2. Flink CDC 3.0 的核心设计与选型理由为什么它能直接吃 MySQL binlog2.1 数据捕获链路binlog 到 Flink 的三种模式与 3.0 的取舍MySQL 的实时捕获绕不开 binlog。开启 binlog 后MySQL 会把所有数据变更以事件形式写入日志Flink CDC 要做的事情就是伪装成一个 MySQL 从库去订阅这些日志事件。常见做法有三种第一种是直接解析 binlog 文件自己维护位点灵活但工程量大第二种是接 Debezium它把 binlog 解析成结构化消息Flink 作为消费端接入第三种是 Flink CDC 自带的连接器底层嵌了 binlog 解析逻辑但对外暴露的是 SQL 表接口用户不用关心位点和反序列化细节。Flink CDC 3.0 走的是第三种路线并且把架构重新梳理了一遍——它不再是简单依赖外部组件而是把全量快照和增量日志统一进同一个框架这样在故障恢复时能做到严格一致的位点衔接。3.0 最大的变化是引入了增量快照框架替代了早期版本的全局读锁加单线程扫描。以前同步一张千万行的大表Flink CDC 2.x 会先对全表加锁然后一条条读出来这期间业务写入会被卡住很多 DBA 一听要开 CDC 就摇头。3.0 的设计思路是把表按主键切成多个 chunk每个 chunk 独立读取同时在读取过程中持续记录 binlog 当前位置。快照阶段和增量阶段不再是串联的两步而是并行推进全部 chunk 读完后再无缝切换到纯增量模式。这玩意听着玄学但实际效果非常明显大表初始化时对业务的影响被压缩到一个很小的窗口。选型时还需要考虑另一个维度目标端。Flink CDC 3.0 本身不限制目标MySQL、Kafka、Doris 都有对应连接器。但如果你是做实时数仓Doris 是比较合适的选择因为它的 StreamLoad 写入协议天然适配 Flink 这种高频微批模式并且 Doris 的 Unique Key 模型能直接承载 CDC 产生的 update 和 delete 操作。相比 Kafka 下游自己去 mergeFlink 到 Doris 的链路更短维护成本也低很多。2.2 全量增量一体化无锁快照与 chunk 分片是怎么做到的增量快照框架的核心是分片算法。Flink CDC 3.0 启动后会先查询目标表的主键范围然后按照设定的 chunk size 把主键区间拆成若干个小块。每个 source 并行子任务认领一个块对这个块执行SELECT * FROM table WHERE id BETWEEN ? AND ?执行完成后记录当前 binlog offset然后继续认领下一个块。因为每个块很小SELECT 持有普通快照读不会对 InnoDB 产生阻塞所以业务端的写操作基本不受影响。这里有几个参数决定分片的粒度。scan.incremental.snapshot.chunk.size控制每个 chunk 的最大行数默认是 8096 行。如果表数据分布不均匀比如主键不是自增而是 UUIDchunk 切分可能产生热点后续版本支持指定chunk-key-column来改变切分键。我一般会先跑通默认配置再根据源表行数和查询耗时来反推 chunk size。比如一张两千万行的表单 chunk 包含 8096 行时每个查询几毫秒就能完成整体并发度提上去后快照速度会非常快。无锁快照还有一层保障依赖 checkpoint。Flink CDC 3.0 会把当前已完成的 chunk 列表和最新 binlog offset 一起放进 checkpoint 状态里。如果作业中途挂了重启后会跳过已经读取完成的 chunk直接从未完成的部分继续同时从最近一次 checkpoint 记录的 binlog 位点恢复增量消费。这样既避免了重复读全表也保证了数据不丢不多。理解这一点对后面调参数非常重要因为 checkpoint 间隔直接影响恢复时长和同步延迟。2.3 目标端选型Doris 在实时写入场景的定位Doris 在实时链路中通常扮演明细层或 ODS 层的角色。它能接受高频写入靠的是 StreamLoad 这个类批量加载接口。Flink Doris Connector 会把一批数据攒起来然后通过 HTTP 提交给 Doris BE 节点Doris 内部再走 memtable 和 compaction 落盘。这和传统 JDBC 逐条 insert 的吞吐完全不在一个量级实测中单 Flink 作业写入 Doris 的吞吐能到每秒数万行甚至更高具体取决于并行度、批次大小和 Doris 集群配置。但 Doris 不是无脑接数据就能干活。它的表模型分三种Duplicate Key 允许重复行Aggregate Key 做预聚合Unique Key 做主键去重和更新。CDC 同步的数据往往包含更新和删除操作如果目标表建成了 Duplicate Key那么同一主键的多条变更会全部保留查询时就会出现重复记录。正确做法是同步到 Doris 的建模表必须使用 Unique Key 模型且主键必须与 MySQL 源表的主键一致。这一点在案例章节会展开写属于新手比较容易踩的坑。另外Doris 的 StreamLoad 天然基于事务Flink 可以通过两阶段提交协议把多个批次合成为一个大事务。这样当作业发生 failover 时Doris 侧能够回滚未提交的数据配合 Flink checkpoint 实现端到端的 exactly-once。所以选型时可以放心Doris 完全有能力承接 Flink CDC 的输出。3. 环境准备与最小作业让 Flink CDC 3.0 在本地先跑通3.1 版本组合与依赖清单flink-cdc-pipeline 与 connector 的搭配Flink CDC 3.0 支持两种使用方式一种是传统的 Flink SQL需要把 MySQL CDC 和 Doris 的 SQL Connector 丢进 Flink 的 lib 目录另一种是 3.0 主推的 Pipeline 方式使用独立的flink-cdc命令行工具把同步链路的 source、sink 和 transform 定义在一个 YAML 文件里然后提交到 Flink 或独立运行。我建议新项目直接用 Pipeline 方式它能省掉创建临时表、字段映射的繁琐步骤结构更清晰。你需要准备几类依赖Flink 运行环境1.17 或 1.18 以上的较新稳定版、Flink CDC 3.0 发布包、MySQL 的 CDC pipeline connector、Doris 的 pipeline connector。这里不用纠结精确版本号只要保证这三个包的主版本一致就行。我习惯先把 Flink 装起来然后把 CDC 发布包解压到lib目录再把两个 connector jar 放进去。如果你用的是 Maven 管理内部项目可以加上这样的依赖版本号用你自己环境对应的值替换dependency groupIdorg.apache.flink/groupId artifactIdflink-cdc-pipeline-connector-mysql/artifactId version${flink.cdc.version}/version /dependency dependency groupIdorg.apache.flink/groupId artifactIdflink-cdc-pipeline-connector-doris/artifactId version${flink.cdc.version}/version /dependency注意这两个 artifact 在发布版仓库里通常已经打包好依赖你不需要额外引入 Debezium 或 MySQL 驱动。排错时如果遇到ClassNotFound优先检查是不是把 pipeline connector 和 sql connector 混用了这两种 jar 的包路径不同不能互换。我最初就是把它们混在一起导致作业启动时报了一堆方法签名错误后来清理干净只保留 pipeline 版才跑通。3.2 用 Flink SQL 写一条 MySQL - Doris 的同步管道如果你是第一次接触 Flink CDC从 SQL 方式开始最容易建立直觉。SQL 方式要求你先定义源表映射 MySQL再定义目标表映射 Doris最后用一条 INSERT 语句把两者串起来。下面是一个最小可用的 SQL 作业源表是 MySQL 的app_db.orders目标表是 Doris 的ods.ordersCREATE TABLE mysql_order ( id INT PRIMARY KEY, user_id INT, amount DECIMAL(10, 2), status STRING, update_time TIMESTAMP(3) ) WITH ( connector mysql-cdc, hostname 127.0.0.1, port 3306, username cdc_user, password cdc_pass, database-name app_db, table-name orders, server-id 5400-5404 ); CREATE TABLE doris_order ( id INT, user_id INT, amount DECIMAL(10, 2), status STRING, update_time TIMESTAMP(3), PRIMARY KEY (id) NOT ENFORCED ) WITH ( connector doris, fenodes 127.0.0.1:8030, table.identifier ods.orders, username doris_user, password doris_pass, sink.label-prefix flink-cdc-order ); INSERT INTO doris_order SELECT * FROM mysql_order;这段 SQL 里有几个关键点需要解释。server-id被设置成一个范围5400-5404这不只是普通配置项它直接决定并行度。Flink 会为每个 source 子任务分配一个范围内的唯一 id如果你只写一个数字而 source 并行度大于 1任务就会因为 server-id 冲突而失败。具体取值规则是你的 source 并行度是 nserver-id 范围至少要有 n 个可用 id。至于为什么是 5400 而不是 6000没有任何特殊含义只要不跟现有从库冲突就行。Doris 表的PRIMARY KEY (id) NOT ENFORCED是必须写的它告诉 Flink 这张表的主键是什么Doris sink 才能正确生成 update 和 delete 语句。但 Flink 不会真的去校验主键唯一性真正约束在 Doris 建表时决定。sink.label-prefix是用来给每次 StreamLoad 事务打标签的同一个 Flink 作业内必须唯一否则恢复时会找到错误的事务状态。3.3 启动参数与状态目录从 YAML 到 job 提交用 SQL 方式需要进入sql-client执行上面的语句如果只是验证连通性够用。但生产环境我更推荐 Pipeline 方式因为它把 source 和 sink 的配置收敛在一个 YAML 文件里版本化管理也更方便。把下面的内容保存为mysql-to-doris.yamlsource: type: mysql hostname: 127.0.0.1 port: 3306 username: cdc_user password: cdc_pass tables: app_db.orders server-id: 5400-5404 sink: type: doris fenodes: 127.0.0.1:8030 username: doris_user password: doris_pass table.identifier: ods.orders pipeline: parallelism: 4 schema.operator.parallelism: 2然后使用 Flink CDC 提供的命令行提交bin/flink-cdc.sh mysql-to-doris.yaml如果 Flink 集群跑在远程需要在 YAML 里加上flink: { cluster: { host: flink-master, port: 8081 } }之类的配置或者先拷贝到集群提交。Pipeline 方式会隐含地把 source 表和 sink 表按照字段名一一映射字段名不一致时需要显式写route或transform规则。我先不展开因为最小作业跑通时不需要它。整个流程跑通后你可以去 MySQL 里做一条INSERT或UPDATE然后查询 Doris 对应表应该能在秒级看到数据变化。如果看不到优先检查 Flink TaskManager 日志里的 StreamLoad 报错信息大多数问题都能从那里找到答案。4. 生产级参数配置从全量快照到增量同步的每一处关键开关4.1 MySQL Source 参数server-id、锁粒度与并发分片生产环境配置 Flink CDC 的 MySQL source第一件要做的事就是固定server-id范围。我见过不少作业启动时正常跑几个小时后突然被 MySQL kill 掉原因就是作业 restart 后重新生成随机 server-id跟正在跑的另一个任务或者自建从库冲突。固定范围以后Flink 会在 checkpoint 里记录每个子任务对应的 id重启时会沿用旧 id不会产生漂移。建议每个同步作业分配一个独立的、不与生产从库重叠的号段比如5400-5449五十个号足够支持五到十个并行度。scan.startup.mode决定作业从哪个位点开始读常用值有initial、earliest-offset和latest-offset。做全量加增量同步时用initial它会在启动时先做快照再进入增量如果只想看新数据直接用latest-offset可以跳过存量扫描。这里有个容易误解的地方即使选initialFlink CDC 3.0 的快照也不是一次性全表锁住而是切块并行读所以线上可以放心用。但如果你的业务对锁极其敏感可以把scan.snapshot.fetch.size从默认的 1024 调小同时降低 source 并行度代价是快照时间变长。再一个是connect.timeout和heartbeat.interval。当源表长时间没有变更binlog 连接可能被 MySQL 或防火墙超时断开设置合理的 heartbeat 可以保活。我通常把heartbeat.interval设成 5 秒这样即使业务库静默连接也一直保持活跃。对于大表快照scan.incremental.snapshot.chunk.size是最需要感知的参数。它默认 8096 行如果你的表单行很大比如包含 text/blob 字段读取一个 chunk 可能耗时很长导致 checkpoint 之间一个 chunk 都读不完快照进度卡住。这种情况下可以调小到 2000同时增加 source 并行度让多个小 chunk 并发处理整体速度反而更快。4.2 Doris Sink 参数label 机制、批次大小与写入模型Doris sink 的参数对最终写入性能影响很大。第一个必须配置的是sink.label-prefix它用来生成 StreamLoad 的 label。Flink 的 Doris 连接器在每轮 checkpoint 结束时会把该周期内的所有批次用一个 label 提交到 Doris如果没有前缀区分两个不同的作业共用同一个 label 空间重启时有可能错误地复用未完成的提交。常见的做法是在 label 前缀里带上业务线名称比如order_cdc、user_cdc保证全局唯一。第二个参数是sink.enable.batch-mode。默认情况下Flink 每条记录对应一次 StreamLoad 调用吗其实不对连接器内部已经有 buffer 机制。打开 batch-mode 后它会按批次大小和间隔合并数据显著减少 HTTP 请求次数对吞吐提升非常明显。我通常配合设置sink.batch.size到 100000 行或更大同时把sink.batch.interval设成 5 秒。注意 batch 太大会增加内存占用需要在 TaskManager 堆内留出足够空间。这里有个权衡明显增大 batch.size 能提高写入吞吐但同步延迟会从亚秒级变成秒级。如果你的场景要求低延迟比如实时大屏的秒级刷新可以调小 batch.interval 到 1 秒保持小批量高频提交。还有一个隐藏比较深的参数是sink.properties.format。Doris 的 StreamLoad 支持 JSON 和 CSV 两种格式Flink 连接器默认用 JSON。如果你的目标表字段有很多 DECIMAL 或 DATETIME 类型建议显式设置sink.properties.format json并配合sink.properties.column_separator因为 CSV 格式在类型转换上容易出精度问题。我遇到过 DECIMAL 字段被截断成整数的坑排查到最后就是 CSV 解析把后面的小数位吃掉了换 JSON 后一切正常。写入模型方面Doris 建表时对主键模型有硬性要求。同步 CDC 数据的表必须使用 Unique Key 模型并且不能设置enable_unique_key_merge_on_write为 false否则 update 操作会退化成 insert。DDL 大致是这样CREATE TABLE ods.orders ( id INT NOT NULL, user_id INT, amount DECIMAL(10, 2), status VARCHAR(20), update_time DATETIME ) UNIQUE KEY(id) DISTRIBUTED BY HASH(id) BUCKETS 10 PROPERTIES ( replication_num 1 );4.3 checkpoint 与一致性failover 后怎么保证不丢不重Flink CDC 的端到端一致性最终是压垮在 checkpoint 和两阶段提交上的。源端 MySQL CDC 连接器会把当前读到的 binlog offset 存入状态Doris sink 则实现了一个两阶段提交的 SinkFunction每个 checkpoint 完成时它会先预提交当前批次的事务等 checkpoint 真正成功后再触发 commit。所以 checkpoint 的时间间隔和稳定性决定了整个同步管道的可靠边界。checkpoint.interval我一般设在 10 到 30 秒。太短会导致频繁生成快照占用大量 IO太长则故障恢复时重启读取的 binlog 量巨大恢复时间变长。设置 checkpoint 间隔时还要考虑 Doris 的表模型和 StreamLoad 自身的事务超时。Doris 的 StreamLoad 默认事务超时时间是 300 秒如果 checkpoint 间隔超过这个值或者某个批次数据量特别大导致上传时间超过超时时间commit 就会失败。所以当出现transaction timeout错误时第一反应应该是调大 Doris 的stream_load_default_timeout_second并把 checkpoint 间隔控制在一个合适范围。还有一个容易被忽略的配置execution.checkpointing.externalized-checkpoint.metadata.job-id和状态目录。生产环境需要让重启后的作业能够从上次的 checkpoint 恢复。如果你用 YAML 提交可以在 Flink 配置里设置state.checkpoints.dir指向 HDFS 或分布式文件系统。丢失状态目录的后果很严重作业重启后会从initial重新扫全表而 Doris 里已经有旧数据于是出现大规模重复。我吃过这个亏所以现在每次上线前都会确认状态目录是共享持久化的而不是 TaskManager 本地临时目录。5. 避坑手册Flink CDC 3.0 同步 Doris 的 5 个典型故障5.1 现象作业运行一段时间后MySQL 侧报连接被 killed日志里出现binlog dump thread错误原因最常见的是 server-id 冲突。MySQL 规定同一个 server-id 只能有一个 binlog dump 连接当 Flink CDC 作业重启后自动生成了与某个已存在连接相同的 idMySQL 就会強制断开先前连接。或者两个 CDC 作业的 server-id 范围重叠也会触发相同问题。解决在 source 配置中显式固定server-id并且为每个作业分配互不重叠的连续范围。我自己的惯例是把 server-id 范围写入配置中心的作业定义里不要依赖自动生成。同时确认作业的 source 并行度不超过 server-id 范围大小。5.2 现象Doris 写入时报错Label already exists或Duplicate label任务进入无限重试原因同一时刻多个 Flink 任务使用了相同的sink.label-prefix或者上一个失败任务残留的 label 没有被清理。Doris 的 StreamLoad label 在短时间窗口内必须全局唯一重复 label 会被拒绝。解决为每个 Flink 作业设置唯一的sink.label-prefix。如果错误已经发生可以手动到__internal_schema或通过 Doris 的show streamload查看残留 label等它超时清理或者调大 Doris 的 label 保留时间配置。重点还是防止前缀碰撞。5.3 现象同步到 Doris 后update 操作变成了重复行表中出现大量主键重复数据原因Doris 建表时用了 Duplicate Key 模型导致 Flink 的 update 语义退化成 insert。也可能是 Flink 目标表声明了主键但 Doris 表没有对应 Unique Key 约束两边的元数据不一致。解决将 Doris 目标表改为 Unique Key 模型并确保 Flink SQL 中的PRIMARY KEY字段与 Doris 表的UNIQUE KEY完全一致。修改后需要重导存量数据因为已写入的重复行不会被自动回滚。这里用 SQL 方式尤其要注意NOT ENFORCED只是告诉 Flink 这个 key 存在不会帮你修正 Doris 表结构。5.4 现象全量快照阶段业务库的 CPU 和连接数突然飙升甚至出现锁等待原因虽然 Flink CDC 3.0 采用无锁快照但如果源表没有主键增量快照框架无法切分 chunk只能退化到单线程全表扫描此时会长时间持有普通读取锁吗实际上没有主键时无法做分片直接退化为单次快照读耗时很长且每批读取都带来大量 SQL 查询压力集中。解决为源表添加主键或者使用scan.incremental.snapshot.chunk.key-column指定一个唯一列作为切分键。如果确实无法加主键降低source.parallelism和chunk.size并把快照任务安排在业务低峰期执行。更彻底的做法是先在 MySQL 里补一个自增主键很多大表其实都有潜在线索列问题不大。5.5 现象源表执行 ALTER TABLE 加列后Flink 作业报 schema 不匹配直接失败原因Flink CDC 3.0 默认不同步 DDL 变更只会把变更后的 binlog 事件里的数据尝试映射到旧 schema 上。新增一个非空列且没有默认值会导致后续 insert 事件反序列化失败。解决目前最稳妥的做法是提前在 Doris 执行对应的 ALTER 语句然后手工修改 Flink 作业中的字段定义最后从 checkpoint 重启。如果加列操作频繁可以把表按整行作为 JSON 同步到 Doris 的 String 字段里但这会牺牲查询性能属于取舍问题。我通常对核心表采用显式列映射对变更频繁的表采用 JSON 兜底。6. 应用案例与进阶验证模拟订单库同步 Doris 的完整复盘6.1 案例背景与 pipeline 配置某模拟项目X需要把 MySQL 的order_db库下订单表、订单明细表实时同步到 Doris用于运营实时看板和指标计算。源表orders约两千万行日增约两百万行主要操作是状态更新和少量退款删除。目标 Doris 表使用 Unique Key 模型主键为order_id。对应的 pipeline 配置可以写成这样source: type: mysql hostname: mysql-server port: 3306 username: cdc_user password: ****** tables: order_db.orders server-id: 5500-5509 scan.startup.mode: initial scan.incremental.snapshot.chunk.size: 4096 sink: type: doris fenodes: doris-fe:8030 username: doris_user password: ****** table.identifier: ods.orders sink.label-prefix: order_cdc sink.enable.batch-mode: true sink.batch.size: 50000 sink.batch.interval: 3000 pipeline: parallelism: 6这个配置里我刻意没有把并行度拉满是因为源库还有其它业务读取必须控制连接数。chunk size 从默认 8096 降到 4096让每个快照查询更快避免慢查询积压。sink 侧 batch.size 5 万行配合 3 秒间隔实测延迟大约在 3 到 5 秒吞吐稳定在每秒三万行左右。如果你需要更低延迟把 batch.interval 调到 1000batch.size 降到一两万即可。6.2 数据一致性校验与延迟观察作业运行一小时后需要验证数据是否真的一致。我会先对比源表和目标表的行数再对比关键字段的汇总值-- 在 MySQL 执行 SELECT COUNT(*), SUM(amount) FROM order_db.orders; -- 在 Doris 执行 SELECT COUNT(*), SUM(amount) FROM ods.orders;两边数值一致说明全量同步没有丢数据。接着验证增量更新是否生效在 MySQL 里把某个订单状态从PENDING改成PAID然后立刻去 Doris 查这个订单的状态如果 3 秒内变化链路延迟就符合预期。更严谨的做法是开启 Flink JobManager 的 Metrics观察flink-cdc/currentFetchEventTimeLag这个指标它能反映事件从源库抓取到处理的时间差。这个案例过程中我踩过最狠的坑是忘了给 Doris 表设置 Unique Key导致更新测试时出现重复行。后来重新建模、清空、全量重灌才恢复。教训是任何 CDC 写 Doris 的作业上线前的第一张目标表建表语句必须先确认主键模型再谈并行度和性能优化。希望这些踩坑记录能帮你少走弯路如果你正打算做 MySQL 到 Doris 的实时同步按照这套流程走下去至少能避开我当年翻过的一大半车。本文还有配套的精品资源点击获取