Flink CDC 3.0 实时同步 MySQL 到 Doris：YAML 配置与调优实践-北京尧图网络

简介这份资源是尚硅谷研究院编写的Flink CDC 3.0技术文档面向有一定Flink基础的大数据研发工程师聚焦MySQL数据实时捕获与同步至Doris的Streaming ETL全流程。文档从CDC概念切入对比基于查询与基于Binlog两种方式的差异讲解flink-cdc-connectors组件的使用并给出数据准备、DataStream与Flink SQL两种实现路径以及MySQL到Doris的完整实操案例涵盖环境搭建、配置文件编写、任务启动与测试。资源包为1个docx文件约145KB内容紧凑、结构清晰便于按章节检索学习。目前已有1788人学习下载。读者可借此掌握CDC技术原理独立完成项目部署与数据源连接、数据流处理、目标存储对接并理解开启Binlog、设置检查点等关键配置对任务可靠性与稳定性的影响适合边实践边对照调整参数。1. Flink CDC 3.0 做 MySQL 到 Doris 的实时同步到底值不值得上MySQL 到 Doris 的实时同步是过去两年数据团队问得最多的一类需求业务库不停写分析侧要秒级可见中间还不想维护一堆 Sqoop 脚本和调度任务。Flink CDC 3.0 给出的答案是「一条 YAML 描述整条链路」把 source、sink、路由、转换全部收进声明式配置里配合 MySQL 的 binlog 和 Doris 的 Stream Load做到全量加增量的自动衔接。它解决的核心问题是不用再手写 DataStream 作业去拼 Debezium、Kafka 和 Doris Connector也不用自己维护全量快照与增量位点的切换逻辑。适合谁适合手里有 MySQL 业务库、下游用 Doris 做 OLAP 分析、团队有基本 Flink 运维能力的数据工程师。不适合谁源表没有主键、或者业务要求跨库事务严格一致的场景硬上会翻车。2. 先搞懂 Flink CDC 3.0 的管道模型与 Doris Sink 的匹配关系2.1 从 Flink CDC 2.x 到 3.0为什么多了「管道」这一层Flink CDC 2.x 的用法是MySqlSource加DataStream全量阶段靠incremental snapshot算法分片读取增量阶段切到 binlog。这套东西能跑但每个同步任务都要写一遍 Java 代码表结构一变就得改代码重新打包。3.0 引入 Pipeline 概念本质是把「源表到目标表的映射」抽象成配置运行时由框架去决定先跑全量还是先跑增量、什么时候切换、checkpoint 怎么对齐。这个抽象带来的直接好处是新增一张表只需要在 YAML 里加一段不用动代码。代价是灵活性下降比如你想在中间插一个自定义的富化逻辑就得回到 DataStream API 或者用 3.0 提供的 transform 表达式。我一般建议纯同步场景用 Pipeline带复杂计算的场景还是老老实实写 DataStream。2.2 Doris Sink 的写入模型Stream Load 与攒批Doris 的写入入口是 Stream Load本质是一个 HTTP 导入接口把数据按批次推给 BE 节点。Flink CDC 的 Doris Sink 在底下做了三件事攒批、压缩、重试。攒批由sink.buffer-flush.max-rows和sink.buffer-flush.interval控制前者是条数阈值后者是时间阈值谁先到触发谁。这里有个容易忽略的点Doris 的 Stream Load 对单批次大小是敏感的。批次太小导入频率高BE 的 compaction 压力大批次太大单次导入延迟高失败重试成本也高。经验值是把max-rows设在 5 万到 20 万之间interval设在 3 到 10 秒具体看单行宽度和 BE 的磁盘写入能力。2.3 全量与增量的衔接位点是怎么对齐的Pipeline 启动时框架先对每张表做一次全量快照读取读的过程中记录 binlog 位点快照读完切到增量。这个「读快照同时记位点」的动作依赖 MySQL 的REPEATABLE READ隔离级别和 binlog 的 GTID 或 filepos。关键参数是scan.startup.mode可选initial先全量再增量、latest只增量、earliest从最早 binlog 开始。生产上第一次同步用initial后续重启如果不想重跑全量要确保 checkpoint 或 savepoint 能恢复位点。这里踩过的坑是如果 Doris 侧表被清空但 Flink 作业的位点还在重启后不会补数据因为框架认为「已经同步过了」。3. 用 YAML 跑通 MySQL 到 Doris 的最小同步链路3.1 环境准备与依赖放置先确认三件事MySQL 开了 binlog 且格式为 ROWDoris 集群能正常 Stream LoadFlink 集群版本与 CDC 3.0 兼容。CDC 3.0 的 Pipeline 作业通过flink-cdc.sh提交需要把 MySQL Connector 和 Doris Connector 的 jar 放到 Flink 的lib目录或者用--jar参数指定。# 检查 MySQL binlog 配置确认格式为 ROW mysql -h127.0.0.1 -uroot -p -e SHOW VARIABLES LIKE binlog_format; mysql -h127.0.0.1 -uroot -p -e SHOW VARIABLES LIKE binlog_row_image; # 提交 Pipeline 作业指定 YAML 和依赖 jar ./bin/flink-cdc.sh \ --jar lib/flink-cdc-pipeline-connector-mysql-3.0.0.jar \ --jar lib/flink-cdc-pipeline-connector-doris-3.0.0.jar \ /path/to/mysql-to-doris.yamlbinlog_row_image必须是FULL否则 update 事件里拿不到完整的前镜像Doris 侧做部分列更新会出错。flink-cdc.sh的--jar可以多次指定框架会把它们加到作业的 classpath。3.2 编写 Pipeline YAMLsource、sink、route 三段################################################################################ # MySQL 到 Doris 的 Pipeline 配置 ################################################################################ source: type: mysql hostname: 127.0.0.1 port: 3306 username: cdc_user password: cdc_pass tables: app_db\.orders, app_db\.order_items # 正则匹配注意转义点号 server-id: 5400-5404 # 与已有从库错开避免冲突 scan.startup.mode: initial # 首次全量增量 sink: type: doris fenodes: 127.0.0.1:8030 # FE 的 HTTP 端口 username: root password: table.create.properties.replication_num: 1 # 测试环境单副本生产按需调 route: - source-table: app_db\.orders sink-table: doris_db.orders - source-table: app_db\.order_items sink-table: doris_db.order_items pipeline: name: mysql-to-doris-orders parallelism: 2 local-time-zone: Asia/Shanghaiserver-id是 MySQL 复制协议里的客户端标识多个 CDC 作业连同一个 MySQL 必须用不同区间否则会互相踢掉。tables用正则点号要转义成\\.不然会匹配到别的库。route段做库表名映射源库和目标库不同名时必填。parallelism建议不超过源表分片数否则有并发读同一分片的浪费。3.3 启动、观察与验证数据一致性作业提交后在 Flink Web UI 能看到Source、Route、Sink三个算子。全量阶段 Source 的numRecordsOut会快速上涨切到增量后趋于平稳。验证一致性用行数对比加抽样比对。-- MySQL 侧统计源表行数 SELECT COUNT(*) FROM app_db.orders; -- Doris 侧统计目标表行数注意 Doris 的 COUNT 是精确的 SELECT COUNT(*) FROM doris_db.orders; -- 抽样比对最近更新的记录 SELECT id, amount, updated_at FROM app_db.orders ORDER BY updated_at DESC LIMIT 10; SELECT id, amount, updated_at FROM doris_db.orders ORDER BY updated_at DESC LIMIT 10;如果行数对不上先看 Flink 作业有没有报错再看 Doris 的SHOW LOAD有没有失败的导入任务。常见情况是全量阶段某张表读了一半报连接超时作业重启后从 checkpoint 恢复但 Doris 侧已经写入了部分数据导致重复。Doris 的主键模型可以靠主键去重明细模型就会多数据所以目标表建主键模型是更稳的选择。4. 参数调优与常见故障排查4.1 攒批参数与 checkpoint 间隔的配合pipeline段可以配checkpoint.interval默认是 3 秒。这个值和 Sink 的buffer-flush.interval要配合如果 checkpoint 间隔比 flush 间隔还短每次 checkpoint 都会强制 flush攒批就失去意义。我一般把 checkpoint 设成 10 秒flush 间隔设成 5 秒让 flush 先触发。pipeline: name: mysql-to-doris-orders parallelism: 2 checkpoint.interval: 10000 # 10 秒一次 checkpoint sink: type: doris sink.buffer-flush.max-rows: 100000 sink.buffer-flush.interval: 5s sink.max-retries: 3sink.max-retries是 Stream Load 失败后的重试次数设 3 次够用。如果 BE 负载高导致频繁超时先加这个值不如先降max-rows减小单批压力。4.2 源表 DDL 变更后的同步中断MySQL 侧加列、改列类型CDC 作业会捕获 DDL 事件。如果 Doris 侧表结构没跟着变写入会报 schema 不匹配。3.0 支持 schema evolution但 Doris 侧需要开启light_schema_change且只支持加列、改列名这类操作删列和改类型不支持。处理办法是DDL 变更前先在 Doris 侧手动改表再让 CDC 作业继续。如果作业已经报错挂了改完 Doris 表后从 savepoint 重启不要从initial重跑否则全量数据会再灌一遍。4.3 排查清单现象、原因、解决现象一作业启动后全量阶段卡住不动。原因通常是 MySQL 账号没有RELOAD或REPLICATION SLAVE权限或者server-id与已有从库冲突。解决给 CDC 账号补权限GRANT SELECT, RELOAD, SHOW DATABASES, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO cdc_user%;并换一个没被占用的 server-id 区间。现象二增量阶段延迟越来越高。原因可能是 Doris 侧导入变慢BE 的 compaction 跟不上或者 Flink 的 Sink 算子反压。解决先看 Flink UI 的 backpressure 指标如果 Sink 反压降max-rows或加 BE 节点如果 Source 反压看 MySQL 的 binlog 生成速率是不是超过了单并行度的消费能力适当加parallelism。现象三Doris 侧出现重复数据。原因一般是作业失败重启后从旧 checkpoint 恢复重复消费了一段 binlog。解决目标表用主键模型Doris 会自动去重如果是明细模型只能手动按时间范围清理重跑。现象四同步一段时间后作业 OOM。原因可能是全量阶段分片太大或者 Sink 攒批的 buffer 占用过多堆内存。解决调小scan.incremental.snapshot.chunk.size默认 8096 行把单分片数据量降下来同时确认sink.buffer-flush.max-rows没有设得过大。现象五Doris 报tablet writer相关错误。原因通常是 FE 和 BE 的版本不一致或者 Stream Load 的导入事务超时。解决核对 Doris 集群各节点版本统一后再试如果是事务超时调大stream_load_default_timeout_second。5. 把同步链路做稳的几个进阶习惯5.1 用 savepoint 做可控重启Pipeline 作业支持 savepoint。每次要改配置或者升级 Connector先触发一次 savepoint停作业改完从 savepoint 恢复。这样位点不会丢也不会重跑全量。# 触发 savepoint 并停止作业 ./bin/flink stop -p hdfs:///flink/savepoints job-id # 从 savepoint 恢复指定新的 YAML ./bin/flink-cdc.sh \ --jar lib/flink-cdc-pipeline-connector-mysql-3.0.0.jar \ --jar lib/flink-cdc-pipeline-connector-doris-3.0.0.jar \ --from-savepoint hdfs:///flink/savepoints/savepoint-xxxx \ /path/to/mysql-to-doris-v2.yaml--from-savepoint是 CDC 3.0 提交脚本的参数恢复时 YAML 里的 source 配置可以改但server-id和表列表最好保持一致否则位点对不上。5.2 监控指标该盯哪几个Flink 侧盯三个Source 的currentFetchEventTimeLagbinlog 消费延迟、Sink 的numRecordsOut速率、作业的 checkpoint 成功率。Doris 侧盯SHOW LOAD的失败率和 BE 的compaction score。这几个指标正常链路基本就是稳的。5.3 一个容易忽略的细节时区MySQL 的datetime字段不带时区Doris 的DATETIME也不带。如果 Flink 作业的local-time-zone和 MySQL 服务器时区不一致写入 Doris 的时间会偏移。我一般把local-time-zone显式设成Asia/Shanghai并且在 MySQL 连接串里加serverTimezoneAsia/Shanghai两边对齐。5.4 我的习惯先小表验证再上大表每次接新的同步需求我不会一上来就把几十张表全配进去。先挑一张数据量小、更新频率低的表跑通全链路确认全量行数对得上、增量能实时进来、Doris 查询正常再把大表加进去。这个习惯帮我省过好几次「全量跑到一半发现 Doris 表建错了模型」的后悔药。同步链路这东西出问题往往不是框架的锅是配置和表结构没对齐。希望帮到你。本文还有配套的精品资源点击获取

Flink CDC 3.0 实时同步 MySQL 到 Doris：YAML 配置与调优实践

相关推荐

PV-RCNN 3D目标检测实战：从体素到点的两阶段融合与复现指南

KNN鸢尾花分类实战：从数据加载到参数调优的完整拆解

基于 WebAssembly 的轻量级高安全 Agent 代码沙箱：WasmEdge 与 WASI 系统调用级隔离

更多资讯

AI 时代，技术人靠什么赢？从“写好代码”到“做成事情”的范式转移

华为题：LRU 缓存

华为交换机VLAN批量配置实战：端口组与批量绑定核心技巧

数据泄漏风险偏倚评估：一篇给初学者的元分析入门指南

华为题：零钱兑换

Codex CLI配置指南：从环境变量到代理网络的完整实战方案

今日更新

小波包能量谱原理与MATLAB实现：从频带分解到故障特征提取

基于LSTM的网易云音乐评论情感分析：数据清洗与模型调优实战

Alpha-Beta剪枝与极大极小值算法：博弈树搜索原理与实战优化

本周精选

交通拥堵预测毕设：基于Python+LSTM的车辆流量预测系统全流程解析

MATLAB/Simulink搭建10机39节点电力系统暂态稳定仿真实战

无人机航拍三维重建全流程：从SfM到网格生成的避坑指南

本月热门

我发现了一个新思路：用 Remotion + Claude Code 像写代码一样自动化生成短视频

Windows下 Codex 中 Chrome 和 Computer Use 插件不可用问题排查及解决参考方式：TaoToken 统一 Key 配置与验证

2026 大模型集体涨价：用 Python 做企业 Token 成本测算与选型避坑（附配置）

读完文章，想聊聊您的网站？