简介由尚硅谷研究院编写的Flink CDC 3.0技术文档面向有一定Flink基础的大数据研发人员聚焦MySQL数据库变更数据的实时捕获、与Doris目标存储同步的完整链路可应用于实时数仓、流式ETL等场景。内容先从CDC概念界定出发对比基于查询与基于Binlog两种模式的差异明确后者能捕获全部数据变更且延迟更低继而介绍flink-cdc-connectors组件的用法并通过DataStream和Flink SQL两种API给出MySQL到Doris的Streaming ETL实操案例覆盖数据准备、环境搭建、配置编写、任务启动与测试验证等完整步骤。资源为单个docx文档约145KB虽体积小巧但包含可运行代码示例并重点提示开启Binlog、设置检查点等关键配置以保障任务可靠性。目前已有1788人学习下载适合希望自主部署Flink CDC项目并实现高效稳定数据同步的工程师参考收藏。1. CDC 不是轮询为什么 Flink CDC 3.0 是 MySQL 实时同步的另一种解法拿到一个 MySQL 到 Doris 的实时同步需求第一反应通常是写定时任务或全量抽取。但业务要求秒级延迟时全量轮询不仅慢还会把源库拖垮。Flink CDC 3.0 换了一条路直接监听 MySQL Binlog把插入、更新、删除按顺序解析成流再做 ETL 写入 Doris。和 2.x 相比3.0 多了一层 Pipeline 能力一个 YAML 文件就能把整库同步任务跑起来不用单独写 DataStream 代码。这篇笔记适合有一定 Flink 基础、想直接落地 MySQL 到 Doris 同步的工程师。我会把原理、配置、命令和常见坑一起拆开按“是什么 → 怎么用 → 坑在哪”逐步推进。2. Flink CDC 原理与选型先搞清 Binlog 和 Pipeline再动手写任务2.1 基于查询 vs 基于 Binlog延迟、压力与数据完整性CDC 的中文名是变更数据获取核心不是“拉数据”而是“监听变化”。同样是拿到数据库变更实现方式不同后面的同步逻辑和运维成本会差很多。基于查询的方式靠定时任务扫描源表比如 Sqoop、DataX 这类离线工具基于 Binlog 的方式则直接解析 MySQL 的主从复制日志比如 Canal、Maxwell、Debezium。两者对比如下比较项基于查询的 CDC基于 Binlog 的 CDC开源产品Sqoop、DataXCanal、Maxwell、Debezium执行模式Batch 批量Streaming 流式能否捕获所有数据否是变化延迟性高延迟低延迟是否增加数据库压力是否能否捕获删除操作困难天然支持基于查询的方式拿不到删除动作因为删除发生后源表里已经找不到这一行只能靠目标端全量对比来补齐逻辑复杂且滞后。基于 Binlog 的方式能拿到完整的前后镜像插入、更新、删除都是独立事件延迟可以做到秒级对源库几乎没有额外压力。选型时不要为省配置跳过 Binlog否则后面同步删除会非常痛苦。2.2 Flink-CDC 组件与 3.0 的 Streaming ETL 分层Flink 社区维护了一套 flink-cdc-connectors底层封装了 Debezium可以直接从 MySQL、PostgreSQL 等数据库读取全量快照和增量变更。早期版本基本要写 DataStream 代码声明 source、自定义 deserializer、再把数据发到下游。到了 3.0官方把整条链路明确拆成 source、sink、pipeline 三层最直观的变化是可以用 YAML 声明一个同步任务。下面是最小可用的配置骨架source: type: mysql hostname: node02 port: 3306 username: root password: 000000 tables: test.\\.* sink: type: doris fenodes: node01:7030 username: root password: 000000这段配置只做了两件事source 指定从 node02 的 MySQL 读取 test 库所有表sink 指定写入 node01 的 Doris。tables 里的正则写法是为了匹配库下所有表YAML 中写成两个反斜杠最终传给正则引擎才是\.。pipeline 层会负责表结构映射、建表、增量位点管理这些过去需要手写 source 和 sink 完成的脏活3.0 直接内建了。2.3 什么时候用 DataStream、SQL什么时候用 Pipeline三种方式不是替代关系而是按任务复杂度划分。DataStream 适合需要复杂加工的场景多流 join、自定义状态、窗口聚合或者要写入 Kafka、Elasticsearch 等非内置目标。Flink SQL 适合做实时 ETL用 CREATE TABLE 声明源表再用 SQL 做过滤、关联、聚合调试起来非常快。Pipeline 则适合“把 MySQL 原样搬到 Doris”这类同步任务整库几十张表不需要写代码一个 YAML 文件就够了。我的建议是同步优先 Pipeline加工优先 SQL特殊逻辑才用 DataStream。这样既能快速上线也方便以后交给运维同学维护。Flink CDC 3.0 的迭代方向也是把 Pipeline 作为默认入口所以新项目可以直接从 YAML 起步。3. 环境准备MySQL Binlog、Maven 依赖与 Flink 集群三项硬前提3.1 开启 MySQL Binlog参数、重启与验证Flink CDC 读的是 Binlog所以源 MySQL 必须提前开启。这里最容易翻车的是配置改了但没重启或重启后参数被覆盖。先把配置写进/etc/my.cnfserver-id 1 log-binmysql-bin binlog_formatrow binlog-do-dbtest binlog-do-dbtest_routeserver-id 用来标识当前 MySQL 实例不能和集群内其他实例重复。log-bin 指定 Binlog 文件名前缀binlog_formatrow是硬性要求只有 ROW 格式才会记录每行变更的前后镜像Flink CDC 才能解析出完整的 before 和 after。binlog-do-db是白名单只记录指定库的变更生产环境建议加上避免系统库和无业务意义的库产生大量日志。修改后执行sudo systemctl restart mysqld mysql -uroot -p000000 SHOW VARIABLES LIKE log_bin; SHOW VARIABLES LIKE binlog_format;确认log_bin为 ONbinlog_format为 ROW。注意一点修改配置前的历史变更不会补进 Binlog所以第一次启动 Flink CDC 任务时要用全量快照兜底这也正是启动参数里 initial 模式存在的意义。3.2 建库建表与数据准备test 和 test_route 双库双份为了验证单库同步和整库匹配原文准备了两个库test 和 test_route每个库里有 t1、t2、t3 三张表。执行以下 SQLCREATE DATABASE test; CREATE DATABASE test_route; USE test; CREATE TABLE t1( id VARCHAR(255) PRIMARY KEY, name VARCHAR(255) ); CREATE TABLE t2( id VARCHAR(255) PRIMARY KEY, name VARCHAR(255) ); CREATE TABLE t3( id VARCHAR(255) PRIMARY KEY, sex VARCHAR(255) ); USE test_route; CREATE TABLE t1( id VARCHAR(255) PRIMARY KEY, name VARCHAR(255) ); CREATE TABLE t2( id VARCHAR(255) PRIMARY KEY, name VARCHAR(255) ); CREATE TABLE t3( id VARCHAR(255) PRIMARY KEY, sex VARCHAR(255) );两库结构保持一致是为了测试 YAML 里的库级正则能否同时命中两个库。t3 的字段故意不是 name 而是 sex用来验证 DDL 和 schema 同步能力。插入数据时让 test_route.t2 用 1004-1006数据不一致才能区分来源库INSERT INTO test.t1 VALUES(1001,zhangsan),(1002,lisi),(1003,wangwu); INSERT INTO test_route.t1 VALUES(1001,zhangsan),(1002,lisi),(1003,wangwu); INSERT INTO test_route.t2 VALUES(1004,zhangsan),(1005,lisi),(1006,wangwu);每张表都建议带主键。Doris 侧做 UPDATE 和 DELETE 时依赖主键定位行源表没有主键会让同步链路很难处理后面避坑部分会单独说。3.3 Maven 依赖、Checkpoint 配置与集群启动DataStream 方式需要建一个 Maven 工程关键依赖如下。版本以原文为准Flink 1.18.0flink-connector-mysql-cdc 3.0.0MySQL 驱动 8.0.31。properties flink-version1.18.0/flink-version maven.compiler.source8/maven.compiler.source maven.compiler.target8/maven.compiler.target /properties dependencies dependency groupIdorg.apache.flink/groupId artifactIdflink-java/artifactId version${flink-version}/version /dependency dependency groupIdorg.apache.flink/groupId artifactIdflink-streaming-java/artifactId version${flink-version}/version /dependency dependency groupIdorg.apache.flink/groupId artifactIdflink-clients/artifactId version${flink-version}/version /dependency dependency groupIdorg.apache.flink/groupId artifactIdflink-table-planner_2.12/artifactId version${flink-version}/version /dependency dependency groupIdorg.apache.flink/groupId artifactIdflink-table-runtime/artifactId version${flink-version}/version /dependency dependency groupIdorg.apache.flink/groupId artifactIdflink-table-api-java-bridge/artifactId version${flink-version}/version /dependency dependency groupIdorg.apache.flink/groupId artifactIdflink-connector-base/artifactId version${flink-version}/version /dependency dependency groupIdcom.ververica/groupId artifactIdflink-connector-mysql-cdc/artifactId version3.0.0/version /dependency dependency groupIdmysql/groupId artifactIdmysql-connector-java/artifactId version8.0.31/version /dependency /dependenciesmysql-cdc 是读取 Binlog 的 source 组件mysql-connector-java 是 JDBC 驱动。打包时用 maven-assembly-plugin 打带依赖的 jar避免运行时缺类。Flink 集群这边需要预先开启 Checkpoint因为 Flink CDC 会把 Binlog 位点保存到状态里靠 Checkpoint 实现断点续传。在flink-conf.yaml里加execution.checkpointing.interval: 5000这里每 5 秒做一次 Checkpoint频率可以根据业务容忍度调整。如果不开启任务一重启就会重新全量读增量同步的意义就没有了。随后启动集群start-dfs.sh bin/start-cluster.sh cd fe bin/start_fe.sh cd be bin/start_be.sh启动后先确认 FE 和 BE 进程都在再检查端口是否被防火墙拦截。Flink 集群、HDFS、Doris 三者连通后环境部分才算就绪。4. 三种落地姿势DataStream、SQL 与 PipelineMySQL 到 Doris 都吃这一套4.1 DataStreamMySqlSource 全量 增量与启动参数先看完整代码这是 Flink CDC 2.x 之后推荐的 MySqlSource 写法import com.ververica.cdc.connectors.mysql.source.MySqlSource; import com.ververica.cdc.connectors.mysql.table.StartupOptions; import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema; import org.apache.flink.api.common.eventtime.WatermarkStrategy; import org.apache.flink.api.common.restartstrategy.RestartStrategies; import org.apache.flink.api.common.time.Time; import org.apache.flink.runtime.state.hashmap.HashMapStateBackend; import org.apache.flink.streaming.api.CheckpointingMode; import org.apache.flink.streaming.api.datastream.DataStreamSource; import org.apache.flink.streaming.api.environment.CheckpointConfig; import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; import java.util.concurrent.TimeUnit; public class FlinkCDCDataStreamTest { public static void main(String[] args) throws Exception { StreamExecutionEnvironment env StreamExecutionEnvironment.getExecutionEnvironment(); env.setParallelism(1); // 开启 Checkpoint每 3 秒一次EXACTLY_ONCE 保证 Binlog 位点不丢 env.enableCheckpointing(3000L, CheckpointingMode.EXACTLY_ONCE); env.getCheckpointConfig().setCheckpointTimeout(60 * 1000L); env.getCheckpointConfig().setMinPauseBetweenCheckpoints(3000L); env.getCheckpointConfig().enableExternalizedCheckpoints( CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION); env.setRestartStrategy(RestartStrategies.failureRateRestart( 3, Time.of(1, TimeUnit.DAYS), Time.of(1, TimeUnit.MINUTES))); env.setStateBackend(new HashMapStateBackend()); env.getCheckpointConfig().setCheckpointStorage(hdfs://node01:8020/flinkCDC); MySqlSourceString mySqlSource MySqlSource.Stringbuilder() .hostname(node02) .port(3306) .databaseList(test) .tableList(test.t1) .username(root) .password(000000) .deserializer(new JsonDebeziumDeserializationSchema()) .startupOptions(StartupOptions.initial()) .build(); DataStreamSourceString mysqlDS env.fromSource( mySqlSource, WatermarkStrategy.noWatermarks(), MysqlSource); mysqlDS.print(); env.execute(); } }这段代码最核心的是MySqlSource.builder()的链路。databaseList(test)指定捕获的库tableList(test.t1)指定捕获的表格式是库名.表名。startupOptions(initial())表示第一次启动先做全量快照再继续读最新的 Binlog。如果改用latest()任务启动前的历史数据不会同步改用earliest()则会从 Binlog 最早位置开始读要求 Binlog 完整保留。JsonDebeziumDeserializationSchema把 Debezium 的 SourceRecord 转成 JSON 字符串打印出来的数据里包含 before、after、op 字段op 为 c 表示插入u 表示更新d 表示删除。Checkpoint 部分用了 3 秒间隔任务关闭时保留最后一次 Checkpoint这样取消后再启动可以从上次状态恢复。RETAIN_ON_CANCELLATION的意思是 Cancel 任务时不要把 Checkpoint 清掉这是断点续传的后悔药。打包后提交到集群mvn clean package -DskipTests bin/flink run -m node01:8081 -c com.example.FlinkCDCDataStreamTest ./flink-cdc-test.jar观察 TaskManager 日志会看到先读历史数据然后进入增量监听状态。此时在 MySQL 里执行一条 INSERT日志会立刻出现对应 JSON。4.2 Savepoint 断点续传从指定偏移恢复任务Flink CDC 将 Binlog 读取位置保存在状态里所以恢复任务的关键是恢复状态。先给运行中的任务创建 Savepointbin/flink savepoint JobId hdfs://node01:8020/flinkCDC/saveWeb UI 中 Cancel 掉当前任务再带 Savepoint 路径重启bin/flink run -s hdfs://node01:8020/flinkCDC/save/savepoint-5dadae-02c69ee54885 \ -c com.example.FlinkCDCDataStreamTest ./flink-cdc-test.jar从 Savepoint 启动后任务不会再执行全量快照而是直接从上次记录的 Binlog 位点继续读。这里有个容易忽略的点如果作业代码里改了算子名称、并行度或算子拓扑结构反序列化状态会失败。遇到这种情况优先保证 Savepoint 时的作业结构不变才谈得上断点续传。4.3 Flink SQLCREATE TABLE 声明 CDC 源如果不想写 DataStreamFlink SQL 是更轻量的方式。在 Java 工程里创建 StreamTableEnvironment然后用一段建表 SQL 声明 MySQL CDC 源import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; import org.apache.flink.table.api.bridge.java.StreamTableEnvironment; public class FlinkSQLTest { public static void main(String[] args) { StreamExecutionEnvironment env StreamExecutionEnvironment.getExecutionEnvironment(); env.setParallelism(1); StreamTableEnvironment tableEnv StreamTableEnvironment.create(env); tableEnv.executeSql( CREATE TABLE t1 (\n id STRING PRIMARY KEY NOT ENFORCED,\n name STRING\n ) WITH (\n connector mysql-cdc,\n hostname node02,\n port 3306,\n username root,\n password 000000,\n database-name test,\n table-name t1\n )); tableEnv.executeSql(select * from t1).print(); } }PRIMARY KEY NOT ENFORCED是 CDC 建表的固定写法它不要求 Flink 强制维护主键约束只是告诉优化器这张表的变更流以哪个字段作为 Key。WITH 里配置的是 MySQL 连接信息和要读取的库表。这种方式最大的好处是免打包直接在 SQL 客户端里改表名、改库名就能试一条链路适合快速验证连接是否通。4.4 YAML PipelineMySQL 整库同步到 Doris到了 3.0最推荐的生产姿势是用独立分发包里的 flink-cdc.sh 跑 YAML。先把 Flink CDC 包解压并把对应的 pipeline connector 放进 lib 目录tar -zxvf flink-cdc-3.0.0-bin.tar.gz -C /opt/module/ cd flink-cdc-3.0.0 cp flink-cdc-pipeline-connector-doris-3.0.0.jar lib/ cp flink-cdc-pipeline-connector-mysql-3.0.0.jar lib/在 flink-cdc 目录下建 job 文件夹写一个 mysql-to-doris.yamlsource: type: mysql hostname: node02 port: 3306 username: root password: 000000 tables: test.\\.* server-id: 5400-5404 server-time-zone: UTC8 sink: type: doris fenodes: node01:7030 username: root password: 000000 table.create.properties.light_schema_change: true table.create.properties.replication_num: 1 pipeline: name: Sync MySQL Database to Doris parallelism: 1source 里的tables用正则匹配 test 库全部表server-id给了一个范围当前并行度 1 时只会用到 5400范围留出来是为了以后调大并行度。server-time-zone必须写清楚否则时间字段按服务器默认时区解析容易差 8 小时。sink 里light_schema_change: true让 Doris 支持轻量 schema 变更加列这类 DDL 能快速完成replication_num: 1表示单副本测试环境够用生产要按集群副本数调整。先在 Doris 里创建目标数据库再启动任务mysql -uroot -p000000 -P9030 -hnode01 CREATE DATABASE test;bin/flink-cdc.sh job/mysql-to-doris.yaml启动后刷新 Doristest 库里的表应该已经自动建好。然后在 MySQL 的 test 库做一轮变更INSERT INTO test.t1 VALUES(1004,zhaoliu); UPDATE test.t1 SET namezhangsan_v2 WHERE id1001; DELETE FROM test.t1 WHERE id1003; ALTER TABLE test.t1 ADD COLUMN age INT;再去 Doris 里查 test.t1能看到插入的行、更新的姓名、删除的行消失新增的 age 列也出现了。这就是整库同步加 schema 变更同步的完整效果。如果还要同步另一个库比如 test_route只需要把 source.tables 改成test_route.\\.*并在 Doris 预建 test_route 库其他配置不变。5. 避坑与常见问题Binlog、位点恢复与 Doris 同步的五个常见故障5.1 现象-原因-解决五条踩坑记录下面几条都是实际跑任务时容易翻车的地方按现象、原因、解决三步整理。第一任务启动后一直停在全量快照阶段增量变更不生效。现象是日志里反复读表数据MySQL 里执行 update 没有任何输出。原因多半是binlog_format没有真正改成 row或者改了配置后没重启 MySQLFlink CDC 解析不到 ROW 格式的 Binlog。解决方法是先执行SHOW VARIABLES LIKE binlog_format确认结果为 ROW再重启 MySQL任务取消后重新以 initial 模式启动让它重新走一遍快照加监听流程。第二只配置了binlog-do-dbtest结果 test_route 库的变更一条都收不到。现象是 test 库同步正常同步脚本换了 test_route 库后没有反应。原因是 MySQL 只把白名单内库的变更写进 Binlog白名单之外的库根本没有日志下游自然读不到。解决方法是把需要同步的库都写进binlog-do-db一项一行或者干脆去掉这个参数让所有库都记录 Binlog然后用 Flink 侧的tables正则控制范围。第三从 Savepoint 恢复时报错提示无法恢复状态。现象是带-s路径启动任务后直接失败TaskManager 日志里出现算子状态不匹配。原因通常是作业代码改了算子结构、并行度变化或者 Checkpoint 存储路径换了。解决方法是恢复前保持作业拓扑不变、并行度不变、setCheckpointStorage路径不变如果确实有算子被删除可以在启动参数里加--allowNonRestoredState跳过缺失状态但这只能作为临时手段。第四MySQL 到 Doris 同步启动了但新表没自动建出来。现象是 Doris 里只有手动建的表有数据新表完全不出现。原因是 sink 侧缺少建表配置、Doris 账号没有建表权限或者 source 里的tables正则写错。解决方法是确认table.create.properties.light_schema_change存在确认 Doris 用户有建库建表权限再用test.\\.*这种写法匹配整个库。注意 YAML 里如果写成test.*正则会匹配 test 开头的所有库包括 test_route同步范围会意外扩大。第五同步进去的时间字段和 MySQL 不一致固定差 8 小时。现象是其他字段都对只有 timestamp/datetime 字段偏早或偏晚。原因是源连接、Flink JVM、Doris 三处时区不一致。解决方法是在 source 里显式配置server-time-zone: UTC8同时把 Doris FE 的时区也设置为 Asia/Shanghai并在 MySQL JDBC 连接串上追加connectionTimeZoneAsia/Shanghai。5.2 排查位点与日志用三个命令定位同步卡点同步任务跑久了最常遇到的是“任务活着但数据不动”。这时先到源库确认当前 Binlog 位置SHOW MASTER STATUS;返回结果里的 File 和 Position 是 MySQL 当前写到哪里的标志。再对比 TaskManager 日志里 Flink CDC 记录的 BinlogOffset如果源库 position 已经很大Flink 还停在旧位点说明消费卡住了优先看 Checkpoint 是否连续失败。如果 Checkpoint 失败日志里通常会出现长时间 GC 或者写 HDFS 超时。另一个常见排查点是并行度。YAML 里parallelism调大后每个并行子任务需要独立的 server-id所以server-id要配置成范围。如果范围不够日志里会出现 Binlog 客户端连接被拒绝。遇到这类问题先不要查 Doris先把 Flink Checkpoint 和 BinlogOffset 对齐再往下游排查。6. 进阶把增量一致性验证做成固定工序6.1 用手工变更清单验证 DML 和 DDL 同步同步任务不是启动成功就算完。我的习惯是每次新建一个 MySQL 到 Doris 的 Pipeline都在测试环境先跑一遍固定四类操作INSERT、UPDATE、DELETE、ALTER TABLE ADD COLUMN。INSERT 验证新增行UPDATE 验证主键定位和字段更新DELETE 验证删除事件能正确落到 DorisALTER TABLE 验证 schema 变更同步。验证时先把源表和目标表各自导入临时表做一次全量对比再执行上面的变更5 秒后到 Doris 查询。不要只在 Web UI 里看任务状态数据一致才是唯一标准。6.2 无主键表与生产前置检查源表没有主键时Pipeline 能读但 Doris 更新和删除时无法确定目标行同步结果会不可控。我一般会要求源表补主键实在补不了就在同步链路之前用 SQL 生成一个唯一键比如把 row_number 加工成 id 字段。这个逻辑不是 Flink CDC 内置能力需要接到一个加工层去处理所以能补主键就尽量补。生产任务上线前我会强制检查五件事binlog_format 是不是 row、server-id 范围是否足够、Checkpoint 间隔是否合理、Doris 账号是否有建表权限、时区是否对齐。这几项都确认后再从 Savepoint 恢复一次确保断点续传路径没有断裂。从那以后我每次新建同步任务都强制走一遍流程先跑四类变更做一致性验证再看 TaskManager 日志里的 BinlogOffset确认无误后才切生产。这套工序帮我挡掉了不少隐藏问题希望帮到你。本文还有配套的精品资源点击获取
