简介：面向大数据实时同步场景的 Flink CDC 3.0 实战资料，由尚硅谷研究院编写，适合有一定 Flink 基础的研发工程师，目标是解决 MySQL 到 Doris 的高效稳定数据同步问题。资料先厘清 CDC 定义，对比基于查询与基于 Binlog 两种方式的差异，突出 Binlog 方案可捕获全部变更、延迟低且不增加数据库压力，并介绍 flink-cdc-connectors 组件能力。核心部分给出完整案例实操，包含 MySQL 数据准备、DataStream 与 Flink SQL 两种接入方式，以及 MySQL 到 Doris 的 Streaming ETL 实现，覆盖环境搭建、配置编写、任务启动与结果测试，附详细代码示例和操作指南。资源为 1 个 docx 文档，大小 145KB，内容紧凑、步骤清晰，重点提示开启 Binlog、设置检查点等可靠性配置，便于读者按需调参并直接应用于项目。目前已有 1788 人学习下载，适合需要快速上手 Flink CDC 同步链路的工程师。
1. Flink CDC 3.0：这份资料把 MySQL 到 Doris 的实时同步讲透了
做大数据开发的人应该都有过这种体验：业务库的 MySQL 数据要同步到数仓，以前要么用 Sqoop 跑离线批量，要么用 DataX 定时拉，延迟高不说，还总给业务库增加压力。等你想上实时同步的时候，又得在 Canal、Maxwell 之间纠结，搭一套 Debezium + Kafka 的链路，光维护就够喝一壶的。这份 Flink CDC 3.0 的资料，目标就是解决这个痛点——它把 MySQL 的 Binlog 捕获、解析、分发，到最终写入 Doris 的整条链路完整演示了一遍，既有 DataStream API 的底层写法，也有 FlinkSQL 的声明式用法，还包含了 3.0 版本主推的 YAML 配置化同步方案。我拆完这份文档的结论是：对于想从零搭建一套 MySQL 实时数仓同步管道、或者正在调研 CDC 技术的工程师来说，这里面所有的代码和配置都是可以照着敲的，踩坑点也写得比较明确。
2. CDC 技术选型与项目架构：为什么最终落在 Flink 上
2.1 两种 CDC 实现方式的本质差异
文档第一章花了不小篇幅对比基于查询和基于 Binlog 两种 CDC 方案的差异。这个对比值得细看，因为它决定了同步管道的整体架构选型。
| 对比维度 | 基于查询的 CDC | 基于 Binlog 的 CDC |
|---|---|---|
| 代表产品 | Sqoop、DataX | Canal、Maxwell、Debezium |
| 执行模式 | Batch 批量 | Streaming 流式 |
| 能否捕获所有数据变更 | 否，只能捕获查询时刻的快照 | 是，能捕获插入、更新、删除全部操作 |
| 延迟性 | 高延迟，取决于调度周期 | 低延迟，毫秒级 |
| 对数据库的压力 | 增加压力，频繁查询 | 几乎无压力，基于 Binlog 复制协议 |
基于查询的方式本质上是定期对源库做全量或增量查询，然后比对数据差异。这种方式有两个先天缺陷：一是无法感知删除操作，二是性能瓶颈明显。基于 Binlog 的方式则是让 MySQL 自己把变更记录写到二进制日志里，CDC 工具伪装成从库去拉取 Binlog，然后按事件类型解析。这种方式拿到的是一条完整的、带事务语义的变更流，可以精确还原每一次插入、更新和删除。
这个差异直接影响了架构决策：如果你只需要每天凌晨同步一次全量数据，用 DataX 就够了；但如果业务需要分钟级甚至秒级的实时数仓，Binlog 方案就是唯一选择。Flink 社区正是因为看中了这个趋势，才在 flink-cdc-connectors 组件里直接内置了 MySQL、PostgreSQL 等数据库的 CDC source，让这批数据能以流的方式进入 Flink 计算引擎，再通过丰富的 sink 组件落到 Doris、StarRocks、Kafka 等目标端。
2.2 Flink 在 CDC 生态里的位置与版本演进
Flink CDC 组件之所以能流行，是因为它把「读取 Binlog 变更流」这件事封装成了一个标准的 Flink source connector。你在 DataStream API 里把它当成一个数据源接入，在 FlinkSQL 里注册一张映射表，它就能持续不断地往 Flink 作业里推送数据变更事件。其中每条事件记录都包含操作类型（op 字段）、变更前后的数据内容，以及 Binlog 的位点信息。
版本演进上，Flink CDC 的核心能力正在从「主库连接器」向「同步框架」转变。2.x 版本你需要自己维护整条链路，先建表、写代码、再配置 sink，每一层都需要自己完善。3.0 版本最大的变化是提供了一套 YAML 配置化同步方案：你只需要写一个描述 source、sink、pipeline 的文件，就能启动一条完整的同步管道。更关键的是引入了路由机制，可以灵活地把源端表映射到目标端表，这在多库多表同步场景下意义重大。
架构上可以分为三层去理解：最底层是 Debezium 嵌入式引擎，负责真正的 Binlog 解析和变更事件捕获；中间层是 Flink CDC connector 对 Debezium 事件的封装和适配；最上层就是用户直接使用的 API 层，包括 DataStream 模式、SQL 模式以及 YAML pipeline 模式。资料提供的案例恰好把这三层都覆盖到了，从最底层的 API 用法到最高层的配置化同步，一应俱全。
2.3 项目目录结构与前置环境需求
按资料中的实践场景梳理，这套案例涉及的核心环境组件如下：
- MySQL（8.0.x）：源数据库，测试表包含 t1、t2、t3 三张表，分布在 test 和 test_route 两个库
- Flink（1.18.0）：流式计算引擎，同时承担状态存储与 Checkpoint 管理
- Flink CDC（3.0.0）：提供 MySQL source connector 和 Doris pipeline connector
- Doris（1.2.x）：目标数据库，作为实时数仓的分析引擎
- HDFS：用于存储 Flink Checkpoint 和 Savepoint 的状态数据
整个项目的文件组织也围绕这些组件展开：Maven 工程里包含 flink-java、flink-streaming-java、flink-clients、flink-table-planner、flink-table-runtime 等 Flink 基础依赖，以及 flink-connector-mysql-cdc 核心连接器和 mysql-connector-java 驱动；独立部署的 flink-cdc-3.0.0 目录下需要手动放置 flink-cdc-pipeline-connector-doris-3.0.0.jar 和 flink-cdc-pipeline-connector-mysql-3.0.0.jar，用于支撑 YAML 配置化同步任务。
3. DataStream 模式实操：从环境准备到断点续传完整跑通
3.1 开启 MySQL Binlog 与初始数据准备
在写任何代码之前，先把 MySQL 的 Binlog 打开。文档给了一个很清晰的配置方案，在 /etc/my.cnf 里追加如下配置：
# 数据库id server-id = 1 # 启动binlog，mysql-bin是binlog的文件名前缀 log-bin=mysql-bin # binlog类型，maxwell要求为row类型 binlog_format=row # 启用binlog的数据库，需根据实际情况作出修改 binlog-do-db=test binlog-do-db=test_route注意几个关键点：server-id 必须设置，这是 MySQL 主从复制协议中每个实例的唯一标识；binlog_format 必须为 row，因为只有 row 格式的 Binlog 才记录了每一行数据变更前后的完整镜像，而 statement 格式只记录 SQL 语句，无法还原精确的变更数据；binlog-do-db 建议按需指定，只对需要的库开启 Binlog 可以减少日志量，避免对数据库性能造成不必要的额外影响。
binlog-do-db 这个参数容易踩坑。它只能精确到数据库级别，如果你后续新增了一个也需要同步的库，必须记得改配置并重启 MySQL。另外还有个隐含条件——数据表本身必须存在主键，如果表没有主键，CDC 解析出的落库操作无法准确匹配变更行，容易造成数据错乱。
数据准备阶段，需要在 test 数据库里建 t1、t2、t3 三张表并插入测试数据，然后在 test_route 数据库里建同样结构的表但插入不同的数据。这个设计是为了后面验证「同步变更」和「路由变更」两种场景，区分哪些表是直接映射同步、哪些表需要做库名或表名的路由转换。
3.2 Maven 工程搭建与 Flink CDC 依赖引入
使用 Maven 管理工程，参考资料中的依赖结构如下：
<properties> <maven.compiler.source>8</maven.compiler.source> <maven.compiler.target>8</maven.compiler.target> <flink-version>1.18.0</flink-version> <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding> </properties> <dependencies> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-java</artifactId> <version>${flink-version}</version> </dependency> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-streaming-java</artifactId> <version>${flink-version}</version> </dependency> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-clients</artifactId> <version>${flink-version}</version> </dependency> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-table-planner_2.12</artifactId> <version>${flink-version}</version> </dependency> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-table-runtime</artifactId> <version>${flink-version}</version> </dependency> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-table-api-java-bridge</artifactId> <version>${flink-version}</version> </dependency> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-connector-base</artifactId> <version>${flink-version}</version> </dependency> <dependency> <groupId>com.ververica</groupId> <artifactId>flink-connector-mysql-cdc</artifactId> <version>3.0.0</version> </dependency> <dependency> <groupId>mysql</groupId> <artifactId>mysql-connector-java</artifactId> <version>8.0.31</version> </dependency> </dependencies> <build> <plugins> <plugin> <groupId>org.apache.maven.plugins</groupId> <artifactId>maven-assembly-plugin</artifactId> <version>3.0.0</version> <configuration> <descriptorRefs> <descriptorRef>jar-with-dependencies</descriptorRef> </descriptorRefs> </configuration> <executions> <execution> <id>make-assembly</id> <phase>package</phase> <goals> <goal>single</goal> </goals> </execution> </executions> </plugin> </plugins> </build>这里的依赖选择有讲究。flink-java 和 flink-streaming-java 是流处理的基础，flink-table-planner 和 flink-table-api-java-bridge 是后续 FlinkSQL 模式必需的，即使当前只用 DataStream API，也建议提前引入。打包插件用 maven-assembly-plugin 的 jar-with-dependencies 模式，能把所有依赖打进一个可执行的 fat jar，这样提交到 Flink 集群时不需要额外分发依赖包。flink-connector-mysql-cdc 3.0.0 的 groupId 是 com.ververica，注意不要写成 org.apache.flink。mysql-connector-java 用于底层连接 MySQL 实例。
3.3 核心代码逐段拆解：MySqlSource 构建与执行
DataStream 模式的核心代码可以按逻辑拆成几个独立模块。先看环境准备和 Checkpoint 配置：
// TODO 1. 准备流处理环境 StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); env.setParallelism(1); // TODO 2. 开启检查点，Flink-CDC将读取binlog的位置信息以状态的方式保存在CK // 如果要做到断点续传，需要从Checkpoint或者Savepoint启动程序 // 2.1 开启Checkpoint，每隔3秒钟做一次CK，并指定CK的一致性语义 env.enableCheckpointing(3000L, CheckpointingMode.EXACTLY_ONCE); // 2.2 设置超时时间为1分钟 env.getCheckpointConfig().setCheckpointTimeout(60 * 1000L); // 2.3 设置两次重启的最小时间间隔 env.getCheckpointConfig().setMinPauseBetweenCheckpoints(3000L); // 2.4 设置任务关闭的时候保留最后一次CK数据 env.getCheckpointConfig().enableExternalizedCheckpoints( CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION); // 2.5 指定从CK自动重启策略 env.setRestartStrategy(RestartStrategies.failureRateRestart( 3, Time.days(1L), Time.minutes(1L))); // 2.6 设置状态后端 env.setStateBackend(new HashMapStateBackend()); env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop102:8020/flinkCDC"); // 2.7 设置访问HDFS的用户名 System.setProperty("HADOOP_USER_NAME", "atguigu");Checkpoint 对 Flink CDC 来说不是可选配置，而是必选配置。原因在于 Flink CDC source 读取 Binlog 时，会把当前消费到的 Binlog 位点信息存放在 Flink 的状态（State）中。如果没有开启 Checkpoint，状态不持久化，作业一重启就要重新读全量数据；开启后作业崩溃时能从最近一次 Checkpoint 恢复位点，真正做到断点续传。建议保持每 3 到 5 秒一次 Checkpoint 的频率，太频繁会增加 HDFS 的写入压力，太稀疏则可能丢失较多数据。
Checkpoint 存储路径配置到了 HDFS，这意味着在本地 IDE 里直接跑这段代码会报错，必须先启动 HDFS 集群。文档给出的hdfs://hadoop102:8020是一个标准配置，你在自己环境里需要替换为实际 NameNode 的地址和端口。RETAIN_ON_CANCELLATION 也很关键，它保证你在 WebUI 上手动取消作业后，最后一次 Checkpoint 仍然保留在 HDFS 上，后续可以从这个保留的 Checkpoint 恢复位点。
再来看 MySqlSource 的构建：
MySqlSource<String> mySqlSource = MySqlSource.<String>builder() .hostname("hadoop103") .port(3306) .databaseList("test") // 设置捕获的数据库 .tableList("test.t1") // 设置捕获的数据表 .username("root") .password("000000") .deserializer(new JsonDebeziumDeserializationSchema()) // 将SourceRecord转成JSON字符串 .startupOptions(StartupOptions.initial()) .build();builder 模式的可读性很好。hostname 和 port 指定了 MySQL 地址，databaseList 用逗号分隔多个库名，tableList 的格式是"库名.表名"，同样支持逗号分隔多个表。deserializer 选择 JsonDebeziumDeserializationSchema，则每条变更记录会以 JSON 格式输出，包含 binlog 文件名、位点偏移、操作类型、数据内容等字段。startupOptions 决定了作业启动时从什么位置开始读取：
- initial：如果表里已有数据，先做一次全量快照，再继续读最新的 Binlog 变更。这是最常用的选项，兼顾历史数据和实时增量
- earliest：不做全量快照，直接从 Binlog 最早的位置开始读。使用前提是 Binlog 保留了完整的变更历史
- latest：不做全量快照，只从当前时刻开始读增量，作业启动前的数据变更不会被捕获
- specificOffset：从一个指定的 Binlog 位点开始读，适用于从历史备份恢复场景
- timestamp：从某个时间点之后的 Binlog 开始读
最后一个关键部分是接入和执行：
// TODO 4.使用CDC Source从MySQL读取数据 DataStreamSource<String> mysqlDS = env.fromSource( mySqlSource, WatermarkStrategy.noWatermarks(), "MysqlSource"); // TODO 5.打印输出 mysqlDS.print(); // TODO 6.执行任务 env.execute();将构建好的 mySqlSource 通过 env.fromSource 接入流处理环境，WatermarkStrategy.noWatermarks() 表示这个数据源不需要分配水位线，因为 CDC 数据本身没有事件时间的处理需求，用处理时间即可。print() 会把每条变更事件以 JSON 字符串形式输出到 TaskManager 日志，用于验证捕获效果。这样一个最简单的 CDC 作业就完成了，打包提交到 Flink 集群，就能看到 MySQL 里 test.t1 表的每次增删改都会以 JSON 格式输出一条记录。
3.4 Savepoint 恢复与断点续传验证流程
文档在测试部分给出了一个完整的「创建 Savepoint → 重启作业 → 验证断点续传」流程。这里简要梳理一下其中的操作逻辑：
# 给作业创建Savepoint，JobId在WebUI上可以找到 bin/flink savepoint JobId hdfs://hadoop102:8020/flinkCDC/save # 在WebUI上cancel掉当前作业 # 在MySQL的test.t1表中执行insert、update、delete操作 # 从Savepoint重启作业 bin/flink run -s hdfs://hadoop102:8020/flinkCDC/save/savepoint-5dadae-02c69ee54885 \ -c com.atguigu.cdc.FlinkCDCDataStreamTest ./gmall-flink-cdc.jar这段操作验证的是整个 CDC 管道的可靠性：作业从 Savepoint 启动后，会跳过已经消费过的 Binlog 事件，直接从保存的位点继续读取。你在作业停止期间对 MySQL 做的 DML 操作，重启后都会被正确捕获，不会重放历史数据，也不会丢数据。
4. FlinkSQL 模式应用：CDC 源表化与实时链路简化
4.1 用 CREATE TABLE DDL 声明 CDC 数据源
DataStream 模式虽然灵活，但需要写不少代码，而且对上游表结构的变化不敏感。FlinkSQL 模式则把 CDC source 声明成一张维表，用 DDL 的方式描述表结构、连接参数和读取模式，开发效率会高不少。资料中给出的示例代码如下：
package com.atguigu; import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; import org.apache.flink.table.api.Table; 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); // 通过DDL声明一张MySQL CDC映射表 tableEnv.executeSql("" + "create table t1(" + " id string primary key NOT ENFORCED," + " name string" + " ) WITH (" + " 'connector' = 'mysql-cdc'," + " 'hostname' = 'hadoop103'," + " 'port' = '3306'," + " 'username' = 'root'," + " 'password' = '000000'," + " 'database-name' = 'test'," + " 'table-name' = 't1'" + " ) "); // 查询表中数据 Table table = tableEnv.sqlQuery("select * from t1"); table.execute().print(); } }这段代码背后的原理值得说明：当你在 FlinkSQL 里注册这张表时，系统会自动根据 connector = 'mysql-cdc' 定位到对应的 source connector，并根据 database-name 和 table-name 去 MySQL 中读取对应的表结构和数据。在声明字段时，id 被标记为 primary key NOT ENFORCED，意思是声明了主键但不由 Flink 本身强制校验，主键的作用主要是供精确一次处理时识别重复记录。
FlinkSQL 模式相比 DataStream 模式的好处是省掉了反序列化工作，JsonDebeziumDeserializationSchema 的行为被内置封装了。你查询出来的数据直接就是对应的 Java/SQL 类型，不再需要手动解析 JSON。对于复杂的数据加工场景，可以直接在 SQL 里做 join、聚合、窗口计算，语法生态远比 DataStream API 便宜。
4.2 从 SQL 到实时数仓 ODS 层的封装思路
实际生产环境中 FlinkSQL 模式最常见的用途是把 CDC 数据接入实时数仓 ODS 层。一个典型的做法是：在 FLinkSQL 里声明 MySQL CDC 源表，然后创建一张 Kafka sink 表，通过一条 INSERT INTO 语句把变更数据持续写入 Kafka topic，下游再通过 Kafka 数据源接入其他组件做加工处理。
-- 声明上游MySQL CDC表 CREATE TABLE ods_mysql_t1 ( id STRING PRIMARY KEY NOT ENFORCED, name STRING ) WITH ( 'connector' = 'mysql-cdc', 'hostname' = 'hadoop103', 'port' = '3306', 'username' = 'root', 'password' = '000000', 'database-name' = 'test', 'table-name' = 't1', 'scan.startup.mode' = 'initial' ); -- 声明下游Kafka sink表 CREATE TABLE kafka_t1 ( id STRING, name STRING, op_type STRING ) WITH ( 'connector' = 'kafka', 'topic' = 'ods_t1', 'properties.bootstrap.servers' = 'hadoop102:9092', 'format' = 'json' ); -- 将变更数据写入Kafka INSERT INTO kafka_t1 SELECT id, name, 'UPSERT' FROM ods_mysql_t1;这种设计思路是把 CDC 的实时性优势真正利用起来。与离线数仓每天一次全量同步不同，CDC 让 ODS 层能从分钟级延迟降到秒级延迟，而且天然保留了数据的变更历史。你可以通过查询 Binlog 中的 op 字段来判断一条记录是 INSERT、UPDATE 还是 DELETE，这为后续的增量数仓建模提供了重要依据。
5. MySQL 到 Doris 的 Streaming ETL 核心应用：YAML 配置化同步实战
5.1 环境准备与应用部署
前面两个章节讲的是编程 API 的用法，而 Flink CDC 3.0 最吸引人的新特性其实在 YAML 配置化同步这里。它把常见的数据同步场景从「写代码」直接变成「写配置文件」，这在小规模团队或者快速交付场景下非常实用。
准备阶段分三步走：
# 1. 解压Flink CDC 3.0安装包到目标目录 tar -zxvf flink-cdc-3.0.0-bin.tar.gz -C /opt/module/ # 2. 将MySQL和Doris的pipeline连接器放入lib目录 cp flink-cdc-pipeline-connector-doris-3.0.0.jar /opt/module/flink-cdc-3.0.0/lib/ cp flink-cdc-pipeline-connector-mysql-3.0.0.jar /opt/module/flink-cdc-3.0.0/lib/ # 3. 启动Doris FE和BE bin/start_fe.sh bin/start_be.sh # 4. 在Doris中创建目标数据库 mysql -uroot -p000000 -P9030 -hhadoop102 CREATE DATABASE test; CREATE DATABASE doris_test_route;有几个细节需要强调。flink-cdc-pipeline-connector-doris 和 flink-cdc-pipeline-connector-mysql 这两个 jar 必须手动放进 flink-cdc-3.0.0 安装包的 lib 目录，因为官方发布包默认不包含特定数据库的 pipeline 连接器。Doris 的前端端口是 9030，FE 端口是 8030，HTTP 端口是 8040，很多人会把 9030 和 8030 混在一起，导致客户端连接失败。Doris 本身兼容 MySQL 协议，所以可以使用 MySQL 命令行客户端直接访问。
5.2 YAML 配置文件逐行解析：source、sink、pipeline
Flink CDC 3.0 的 YAML 同步配置从结构上分成 source、sink、pipeline 三大块。先看同步变更场景的配置：
source: type: mysql hostname: node01 port: 3306 username: root password: "123" tables: test.\.* server-id: 5400-5404 server-time-zone: UTC+8 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.\.*是一个正则表达式，表示匹配 test 库下的所有表。如果你只想同步特定表，可以写成test.t1或test.t[0-9]，这种正则匹配的灵活性在多表同步场景里非常实用。server-id 需要配置一个范围，4000 到 5404 意味着可以同时开 5 个 Binlog 消费线程，对于多表同步场景建议至少配置 2 个以上的 server-id 段，避免表数量多时单一线程成为瓶颈。
sink 部分的 fenodes 是 Doris 的 FE 节点地址。table.create.properties.light_schema_change开启后，Doris 建表时启用轻量级 Schema Change 特性，当源端新增列时，目标端无需重写数据文件，只需做元数据变更，这个参数在业务表结构经常调整的场景下建议打开。replication_num: 1则在 BE 上只保留一个副本，如果你的 Doris 集群有多个 BE 节点且启用了高可用，这里需要相应调大。pipeline 部分只需配置 name 和 parallelism，前者用于展示同步任务名称，后者控制并发度。
这套配置对开发者的意义在于，你在 MySQL 里对 test 库的任何 DML 或 DDL 变更，都会经过 Flink CDC 管道的解析和转换，自动落到 Doris 对应的表中。
5.3 路由变更配置与多表映射实战
路由变更场景是 Flink CDC 3.0 区别于 2.x 的重要能力。来看资料中的路由配置：
source: type: mysql hostname: hadoop103 port: 3306 username: root password: "000000" tables: test_route.\.* server-id: 5400-5404 server-time-zone: "UTC+8" route: - source-table: test_route.\.* sink-table: doris_test_route.\.*路由规则的本质是实现源表到目标表的映射转换。test_route.\.*和doris_test_route.\.*的映射关系是把 test_route 库下的所有表同步到 doris_test_route 库下。有了这个机制，你可以在不修改业务库表结构的前提下，对目标库的表名、库名做统一规划——比如把生产环境的 test_route 同步到数仓的 ods_test_route。
在实际项目里，基于这套理念你可以做不少有意思的事情。同一张 MySQL 表可以同时配置两条管道，分别写入 Doris 的不同表中，实现一份数据多种用途；可以通过路由把多个 MySQL 实例的表汇聚到同一个 Doris 表，前提是表结构一致；也可以在目标端通过额外的 schema 变更逻辑实现增量列、删除列的自动映射。
5.4 启动任务与 DDL 自动同步验证
配置完成后，启动命令很简洁：
# 启动Flink CDC同步变更任务 bin/flink-cdc.sh job/mysql-to-doris.yaml任务启动后，Flink CDC 会先对指定的表做一次全量快照，把已有数据全部写入 Doris，然后监听 Binlog 持续同步增量变更。验证过程中，可以在 MySQL 中执行新增列、修改列类型等 DDL 操作，然后刷新 Doris 中的表结构，观察 Schema 变化是否自动同步。这个能力对正在迭代中的业务尤为重要，因为只要源端表加了新列，同步管道能自动感知，不需要手动在 Doris 里执行对应的 ALTER TABLE 语句。
不过 DDL 同步也不是万能的。删除列的 DDL 在 Doris 端的支持可能不够完善，如果线上执行 DROP COLUMN，建议先在测试环境验证一遍再同步到生产管道。
6. 常见问题排查与实验环境避坑记录
6.1 MySQL 连不上：Binlog 没开或 server-id 冲突
这个是我在跑通案例时遇到的第一个问题。现象是我照着文档配好了 /etc/my.cnf 后重启 MySQL，然后用 DataStream 方式启动 Flink 作业，发现报错信息显示无法连接 MySQL，或者连接后立刻被拒绝。
原因有两类。一是很多 MySQL 安装包默认关闭了 Binlog，如果你在 my.cnf 里只加了 log-bin 和 binlog_format 但没有加 server-id，MySQL 可能仍然不写入 Binlog。二是 server-id 配置不能和另一个正在运行的 MySQL 实例相同，如果集群里有多个 MySQL 实例，server-id 重复会导致从库连接被踢掉。
解决办法是在 my.cnf 的 [mysqld] 区段同时配置 server-id、log-bin、binlog_format 三个参数，然后执行SHOW MASTER STATUS确认 Binlog 文件是否生成。确认没有问题后再启动 Flink CDC 作业，这个问题就解决了。
6.2 已提交 Flink 作业但看不到数据输出
第二个问题也比较典型：作业成功提交到 Flink 集群，WebUI 上状态是 RUNNING,但 TaskManager 日志里完全看不到 print() 输出的数据。
这种情况通常是表匹配正则没写好。CDC source 的 tableList 参数如果填了test.t1，但实际 MySQL 中表名是T1或者表在另一个库中，就一个事件都匹配不到。另外，如果你使用了 databaseList 和 tableList 的组合，Flink CDC 会按"数据库名 + 表名"做联合匹配，其中一个不匹配就拉不到数据。
排查办法是先用初始状态做验证：在 MySQL 里手动执行一条 INSERT 语句，然后立刻观察 TaskManager 日志里是否有对应的 JSON 输出。如果没有，去 WebUI 检查作业是否有反序列化异常，同时确认日志中是否出现了连接 MySQL 的相关配置信息。我一般会先把 databaseList 和 tableList 的配置简化成只匹配一张表，跑通后再慢慢扩展范围。
6.3 Savepoint 恢复后数据重复或丢失
这个坑在断点续传验证流程中非常常见。现象是创建 Savepoint 后取消了作业，然后在 MySQL 里做了增删改操作，再从 Savepoint 恢复作业，看到日志里既有历史数据的重放，又有恢复之后的新变更。
原因在于 Flink CDC 的位点状态保存在状态中，而状态中不仅包含 Binlog 位点，还包含已经处理过的数据变化。如果 Savepoint 创建的时刻和取消作业的时刻之间存在时间差，作业可能已经消费了更多 Binlog 事件但尚未触发新的 Checkpoint，这些事件会丢失。反过来，如果 Checkpoint 开启的间隔过短，Savepoint 可能包含了部分未完全处理的数据，恢复后引发重复消费。
解决方案是在创建 Savepoint 之前先停止作业，给 Flink 一个足够的缓冲时间来完成最后一次 Checkpoint，然后再创建 Savepoint。此外，需要确认状态后端是 HashMapStateBackend 还是 RocksDBStateBackend，两者在恢复语义上有细微差别，推荐生产环境使用 RocksDB 来降低大状态下的内存压力。
6.4 Doris 写入时 Schema 类型不匹配导致任务失败
最后一个高频问题是源端 MySQL 和目标端 Doris 的字段类型不一致。Doris 的 DECIMAL 类型在底层还是整型存储，精度和标注不同，例如 MySQL 的 DECIMAL(10,2) 同步到 Doris 的 DECIMAL(10,2) 一般没有问题，但如果某个字段在 MySQL 是 VARCHAR(255)，而 Doris 里对应的维度表已经建成了 VARCHAR(100)，那么超长值写入时就会报错。
另外轻量级 Schema Change 默认打开，如果表结构频繁变更，Doris 可能在处理多个并发 ALTER 时发生冲突。我的处理方式是：第一次涉及时先手动在 Doris 建表验证字段类型映射关系，跑通后再启用 pipeline 自动建表；生产环境建议关闭 light_schema_change，因为默认的同步一致性更可控。
从这四条踩坑记录里能总结出一个共性：Flink CDC 链路的大部分问题出在「源端配置」和「目标端兼容性」上，而不是 Flink 计算层本身。花时间把 MySQL 侧 Binlog 配置和 Doris 建表规范提前梳理清楚，后续的同步作业会省很多心力。
7. 二次开发实战：如何把这套同步体系改造为更适合你业务的方案
7.1 增加字段过滤与敏感信息脱敏
原始案例把整行数据原样同步到 Doris，但真实业务往往不是这样。比如订单表里有用户手机号、身份证号，你不想把它们原样落到数仓；或者目标表只需要业务核心字段，不需要 Binlog 的元数据字段。改造思路是在 Flink 层面做一层数据投影与脱敏。
如果是 FlinkSQL 模式，最简单的做法是把 SELECT 的列收窄：
CREATE TABLE mysql_cdc_orders ( id STRING PRIMARY KEY NOT ENFORCED, user_id STRING, phone STRING, order_amount DECIMAL(10,2), create_time TIMESTAMP(3) ) WITH ( 'connector' = 'mysql-cdc', 'hostname' = 'hadoop103', 'port' = '3306', 'username' = 'root', 'password' = '000000', 'database-name' = 'test', 'table-name' = 't1' ); CREATE TABLE doris_orders_safe ( id STRING, user_id STRING, order_amount DECIMAL(10,2), create_time TIMESTAMP(3) ) WITH ( 'connector' = 'doris', 'fenodes' = 'node01:7030', 'table.identifier' = 'test.orders_safe', 'username' = 'root', 'password' = '000000', 'sink.properties.format' = 'json', 'sink.properties.read_json_by_line' = 'true' ); -- 只同步业务字段 INSERT INTO doris_orders_safe SELECT id, user_id, order_amount, create_time FROM mysql_cdc_orders;如果你有更复杂的脱敏逻辑，比如手机号中间四位用星号替代，可以进一步用 SQL 的 REGEXP_REPLACE 实现，或者在 DataStream 模式中自定义一个 map 算子，在写 Doris 之前把敏感字段替换掉。这种做法的好处是改动范围仅在同步链路中间层，源端和目标端完全无感知。
7.2 整合进自研数据同步平台时的接口设计思路
当你接入的 MySQL 实例越来越多，光靠 YAML 文件管理会变得难以维护。常见的做法是用一个配置中心或元数据库来统一管理同步任务，每个同步管道在业务侧通过一个抽象接口描述：源端连接信息、目标端连接信息、表映射规则、同步模式(全量/增量/全量+增量)、Schema 变更策略。Flink CDC 3.0 底层足够稳定之后，上层平台做的其实就是把 YAML 变成模板、把模板变量存库，让运维通过 Web 页面操作自动化生成同步任务。
我一般会这样做：把上面提到的 YAML 拆成「源配置」「目标配置」「路由配置」「Pipeline 配置」四个独立模板，用 FreeMarker 或 Velocity 动态渲染，再通过 REST API 提交到 Flink CDC。这样，开发者在页面上配置好同步需求后，系统自动生成模板，随后触发同步任务。再配合一个监控模块定时扫描任务运行指标，比手动改 YAML 文件的方式可维护性高不少。
7.3 状态一致性与指标监控的强化
Flink CDC 3.0 的同步任务可靠性强依赖 Checkpoint，但默认的 HashMapStateBackend 在状态量大的情况下容易因为堆内存溢出而触发 Full GC。我从实践经验里的建议是：生产环境务必换成 RocksDBStateBackend，并且把 CheckpointStorage 选择在 HDFS 或独立的分布式文件系统上，避免本机磁盘问题导致状态丢失。
如果需要监控同步任务的延迟和吞吐指标，可以在 Sink 端增加自定义回调函数，统计每条记录从 Binlog 捕获到写入 Doris 的端到端耗时，定期写入 Prometheus。一旦发现延迟超过业务阈值，通过告警规则及时通知值班人员。这套体系在项目上线后给我省了不少半夜被叫起来处理问题的时间。从那以后我每次搭建新的同步管道都会强制走一遍：源端配置验证、Checkpoint 参数调优、Schema 变更策略确认、监控告警接入。这四个环节缺一不可，希望对你的项目也能有所帮助。
本文还有配套的精品资源，点击获取
