Flink CDC 3.0 实战：MySQL 到 Doris 实时同步链路搭建与避坑指南

简介这份文档由尚硅谷研究院编写面向具备一定Flink基础的大数据研发人员聚焦Flink CDC 3.0在MySQL数据实时捕获与同步至Doris场景下的全流程解析。内容从CDC概念切入对比基于查询与基于Binlog两种方式的差异重点讲解flink-cdc-connectors组件如何直接读取MySQL、PostgreSQL等数据库的全量与增量变更数据并给出DataStream与Flink SQL两种实现路径。案例部分覆盖数据准备、环境搭建、配置文件编写、任务启动与测试完整演示MySQL到Doris的Streaming ETL落地过程同时强调开启Binlog、设置检查点等关键配置对任务可靠性的影响。资源包为1个docx文档约145KB结构清晰、代码示例完整适合希望独立完成CDC项目部署、实现高效稳定数据同步的工程师参考。目前已有1788人学习。1. Flink CDC 3.0 到底解决了什么从 MySQL 到 Doris 的实时链路为什么值得重做很多团队第一次做 MySQL 到 Doris 的同步用的是「定时批量抽数 全量覆盖」的老路子。白天业务写入高峰不敢跑只能等凌晨低峰期拉一遍第二天上班看到的报表其实是昨天晚上的快照。业务方问「为什么刚下的单在实时大屏上看不到」你只能解释「T1」。这个体验在今天的实时数仓场景里已经很难交代过去了。Flink CDC 3.0 值得单独拿出来讲是因为它把过去需要自己拼装的一整条链路——MySQL binlog 捕获、schema 变更处理、数据转换、写入 Doris——收敛成了一套可以声明式配置的管道。它不再要求你写一大段 DataStream 代码去手动拼 Source 和 Sink而是用一份 YAML 描述「从哪来、到哪去、怎么映射」剩下的交给 Flink 作业去跑。对一线工程师来说这意味着从「搭框架」变成「调参数和排错」落地周期从几天压到几小时。这篇文章面向的是已经会用 Flink、也接触过 Doris但还没把两者用 CDC 串起来的后端或数据工程师。我会按「先讲清原理和选型理由再给可复现的配置和命令最后把踩过的坑摊开」的顺序写目标是让你照着能跑通一条 MySQL 到 Doris 的实时同步链路并且知道出问题时该看哪里。2. 链路拆解Flink CDC 3.0 的管道模型与 Doris Sink 选型2.1 为什么是「管道」而不是「作业」Pipeline 的声明式思路Flink CDC 3.0 最核心的变化是引入了 Pipeline 的概念。在 2.x 时代你要写一个MySqlSource再写一个DorisSink中间用stream串起来schema 变更还得自己监听和处理。3.0 把这条链路抽象成一份 YAML里面分source、sink、pipeline三段。Flink 启动时读取这份配置自动构建 Source、Sink 和中间的转换逻辑。这个抽象带来的直接好处是schema 变更加列、改类型由框架统一处理你不需要在代码里写if (column added)这种分支。代价是灵活性下降——如果你要做复杂的多表 join 再写入Pipeline 模式就不太够用还是得回到 DataStream API。所以选型上我的判断是单表或简单多表的整库同步用 Pipeline需要复杂转换的用 DataStream 自己拼。2.2 MySQL Source 的关键参数server-id、启动模式和并行度MySQL CDC Source 本质上是伪装成一个从库去拉 binlog所以server-id必须唯一不能和现有从库冲突。启动模式有三种initial先全量再增量、latest只从最新位点开始、earliest从最早位点开始。生产环境第一次同步用initial后续重启用latest否则每次重启都全量拉一遍Doris 那边会被写爆。并行度这块有个容易翻车的点MySQL CDC Source 的并行度受表数量限制。如果你只同步一张表并行度设成 4 也没用因为单表的 binlog 只能由一个 subtask 消费。整库同步时并行度可以按表数量来设一般设成表数量的一半到相等。# mysql-to-doris.yaml 的 source 段 source: type: mysql hostname: 10.0.0.10 port: 3306 username: cdc_user password: cdc_pass tables: app_db.orders, app_db.order_items server-id: 5400-5404 # 范围写法Flink 会自动分配 scan.incremental.snapshot.enabled: true scan.startup.mode: initial # 首次全量增量 server-time-zone: Asia/Shanghaiserver-id用范围写法5400-5404时Flink 会按并行度自动分配避免手动指定冲突。scan.incremental.snapshot.enabled开启后全量阶段是分片读取的不会锁表这是 3.0 相比早期版本的重要改进。server-time-zone必须和 MySQL 实例一致否则时间字段会差 8 小时这个坑后面还会提。2.3 Doris Sink 的写入模式Stream Load 与攒批参数Doris Sink 底层走的是 Stream Load也就是通过 HTTP 把数据批量推给 Doris FE再由 FE 分发到 BE。写入性能的关键在攒批sink.buffer-flush.max-rows和sink.buffer-flush.interval控制多久或多少行触发一次写入。设太小Doris 那边小文件多、compaction 压力大设太大延迟高、内存占用高。我一般会这么设max-rows设 50000 到 100000interval设 3s 到 10s。如果业务对延迟敏感比如实时风控interval 压到 1smax-rows 降到 10000。另外sink.enable-delete要打开否则 MySQL 的 delete 操作不会同步到 Doris这是很多人第一次跑完发现数据对不上的原因。# 接上面的 sink 段 sink: type: doris fenodes: 10.0.0.20:8030 username: root password: table.create.properties.light_schema_change: true table.create.properties.replication_num: 3 sink.enable-delete: true sink.buffer-flush.max-rows: 50000 sink.buffer-flush.interval: 5s sink.max-retries: 3fenodes是 Doris FE 的 HTTP 端口默认 8030不是 MySQL 协议端口9030写错了会连不上。light_schema_change打开后加列操作会更快。replication_num按你的 BE 节点数来单机测试设 1生产至少 3。2.4 整库同步的表映射与路由规则Pipeline 支持用route规则做库表映射。比如 MySQL 的app_db.orders想写到 Doris 的ods.orders_rt就在pipeline段里写 route。这个规则支持正则整库同步时特别有用——不用一张张表列出来。pipeline: name: mysql-to-doris-orders parallelism: 2 route: - source-table: app_db\.orders sink-table: ods.orders_rt - source-table: app_db\.order_items sink-table: ods.order_items_rtsource-table用正则时记得转义点号。parallelism设成 2 表示整条管道两个并行度如果 source 有两张表正好一表一个 subtask。如果表更多parallelism 可以往上加但不要超过表数量太多否则空转的 subtask 会浪费资源。3. 从零跑通环境准备、配置编写与提交命令3.1 环境清单与版本对齐跑通这条链路需要三样东西一个 MySQL开启 binlog、一个 Doris 集群、一个 Flink 集群或者本地 standalone。版本上Flink 用 1.18 或 1.19Flink CDC 用 3.0.xDoris 用 2.0 以上。CDC 3.0 对 Flink 1.17 也支持但 1.18 的兼容性更稳。MySQL 这边必须确认binlog_formatROW、binlog_row_imageFULL这两个参数不满足CDC 拿不到完整的变更数据。用下面这条命令确认SHOW VARIABLES LIKE binlog_format; SHOW VARIABLES LIKE binlog_row_image;如果binlog_format是STATEMENT或MIXED需要改配置并重启 MySQL。binlog_row_image如果是MINIMALupdate 操作只会记录被改的列CDC 拿不到完整行写入 Doris 时其他列会是 null。3.2 编写 Pipeline YAML一份可直接复用的完整配置把前面几段拼起来就是一份完整的配置。我把它整理成可以直接改地址就能用的版本# mysql-to-doris-full.yaml source: type: mysql hostname: 10.0.0.10 port: 3306 username: cdc_user password: cdc_pass tables: app_db\.orders, app_db\.order_items server-id: 5400-5404 scan.incremental.snapshot.enabled: true scan.startup.mode: initial server-time-zone: Asia/Shanghai sink: type: doris fenodes: 10.0.0.20:8030 username: root password: table.create.properties.light_schema_change: true table.create.properties.replication_num: 3 sink.enable-delete: true sink.buffer-flush.max-rows: 50000 sink.buffer-flush.interval: 5s sink.max-retries: 3 pipeline: name: mysql-to-doris-orders parallelism: 2 route: - source-table: app_db\.orders sink-table: ods.orders_rt - source-table: app_db\.order_items sink-table: ods.order_items_rt这份配置里tables用正则写法app_db\.orders和 route 里的 source-table 对应。如果只想同步部分列可以在 source 里加projection规则但一般整表同步更省事。3.3 提交作业与验证数据是否落库提交命令用 Flink 自带的flink-cdc.shCDC 3.0 提供的脚本或者用flink run指定 CDC 的 jar。我一般用前者更简洁# 提交 Pipeline 作业 ./bin/flink-cdc.sh \ --jar /opt/flink-cdc/lib/flink-cdc-3.0.0.jar \ --config /opt/flink-cdc/conf/mysql-to-doris-full.yaml提交后去 Flink Web UI 看作业状态Running 之后去 MySQL 插一条数据等 5 到 10 秒取决于攒批 interval然后去 Doris 查-- 在 Doris 里确认数据 SELECT * FROM ods.orders_rt ORDER BY id DESC LIMIT 10;如果查不到先看 Flink 的 TaskManager 日志有没有 Stream Load 报错再确认 Doris 的ods库和表是否被自动创建。CDC 3.0 默认会自动建表但前提是 Doris 用户有建表权限。3.4 全量阶段与增量阶段的切换观察initial模式下作业启动会先做全量快照再做增量。全量阶段可以在 Flink UI 的 Source 算子看到Snapshot状态增量阶段会变成Binlog。这个切换点很关键——如果全量阶段数据量大切换会花几分钟到几十分钟。切换完成后MySQL 的新写入才会开始同步。观察方法在 MySQL 里插一条数据如果 Doris 里立刻能看到说明已经进入增量阶段。如果看不到可能还在全量。全量阶段不要重启作业否则会从头再来一遍。4. 避坑与排查那些让同步链路断掉的细节4.1 时间字段差 8 小时server-time-zone 没对齐现象Doris 里的create_time比 MySQL 里少 8 小时。原因MySQL 的server-time-zone和 Flink 作业的时区不一致CDC 读取 binlog 时按 UTC 解析写入 Doris 时没转换。解决在 source 里显式写server-time-zone: Asia/Shanghai并且确认 Flink 集群的timezone配置也是Asia/Shanghai。两个地方都要对只改一个不够。4.2 全量阶段 OOM快照分片太大现象全量同步大表时 TaskManager 内存爆掉作业重启。原因scan.incremental.snapshot.chunk.size默认值偏大单表数据量大时一个分片读进内存就撑爆了。解决把这个值调小比如从默认的 8096 调到 1024 或 2048。代价是全量阶段分片变多、切换变慢但内存稳了。另外可以给 TaskManager 加内存taskmanager.memory.process.size设成 4g 以上。4.3 Doris 写入报 label already exists现象Flink 日志里反复出现Label Already Exists数据写不进去。原因Stream Load 的 label 重复了。CDC 3.0 用作业名加时间戳生成 label如果作业重启太快时间戳可能撞上。解决确认sink.max-retries设了 3 以上让 Flink 自动重试。如果还不行检查 Doris 的label_keep_max_second配置默认 24 小时可以适当调小。根本办法是保证作业名唯一不要多个作业用同一个 pipeline name。4.4 schema 变更后 Doris 表没跟着变现象MySQL 加了一列Doris 表里没有这列新数据写入报错。原因light_schema_change没开或者 Doris 用户没有 alter 权限。解决sink 里加table.create.properties.light_schema_change: true并确认 Doris 用户有ALTER权限。如果还是不行手动在 Doris 里加列然后重启 Flink 作业。注意CDC 3.0 的 schema 变更同步不是实时的有延迟加列后等一两分钟再看。4.5 增量阶段延迟越来越高现象刚开始同步很快跑几个小时后延迟从秒级涨到分钟级。原因Doris 那边 compaction 跟不上小文件太多写入变慢反压到 Flink。解决调大sink.buffer-flush.max-rows和interval减少写入频率。同时去 Doris 看SHOW BACKENDS的 compaction 状态如果 compaction score 很高说明 BE 压力大需要加节点或调 compaction 参数。这个坑的本质是写入节奏和 Doris 的消化能力不匹配不是 Flink 的问题。5. 进阶技巧让这条链路在生产环境稳下来跑通是一回事跑稳是另一回事。我在生产环境用下来有几个习惯是必须养成的。第一作业名和 pipeline name 绑定业务标识比如mysql-to-doris-orders-v2不要用默认名。这样出问题时能在 Flink UI 和 Doris 的 Stream Load 记录里快速定位是哪个作业。第二监控两个指标Flink 的currentFetchEventTimeLag增量延迟和 Doris 的compaction score。前者超过 30 秒就要看 MySQL 的写入量是不是突增了后者超过 100 就要看 BE 是不是扛不住了。这两个指标比看作业是否 Running 有用得多。第三全量阶段和增量阶段分开跑。如果表特别大上亿行第一次全量用initial跑完然后停掉作业改成latest再启动。这样后续重启不会触发全量避免重复拉数据。这个操作相当于给作业一个「后悔药」全量只做一次。第四Doris 表的分桶数要提前规划。CDC 自动建表时用的分桶数是默认值大表可能不够。我一般会在 Doris 里手动建好表设好分桶数和副本数然后让 CDC 写入已有表。这样比自动建表可控。-- 手动建 Doris 表指定分桶 CREATE TABLE ods.orders_rt ( id BIGINT, user_id BIGINT, amount DECIMAL(10,2), create_time DATETIME, update_time DATETIME ) ENGINEOLAP DUPLICATE KEY(id) DISTRIBUTED BY HASH(id) BUCKETS 16 PROPERTIES (replication_num 3);分桶数按数据量来一般单桶 1GB 到 10GB 比较合适。16 个桶对应千万级数据量上亿就加到 32 或 64。建好表后CDC 的table.create.properties就不会生效了直接写入已有表。最后说一个我自己的教训不要在生产环境用earliest模式。有一次排查问题手滑把启动模式改成了earliest结果作业从最早的 binlog 位点开始拉把三天的变更全部重放了一遍Doris 里数据翻倍清理花了两个小时。启动模式这个参数改之前一定要确认清楚latest是重启时的安全选择。希望帮到你。本文还有配套的精品资源点击获取