企业官网建设流程全解析

简介：围绕 Flink CDC 3.0 实现 MySQL 到 Doris 实时同步的完整技术文档，由尚硅谷研究院编写，面向具备一定 Flink 基础、对实时数据处理和流式 ETL 感兴趣的大数据研发工程师。资源包含一个 docx 文档，压缩包约 145KB，文档系统梳理了 CDC 概念，对比基于查询与基于 Binlog 两种方式的差异，阐述 flink-cdc-connectors 组件作为 Source 直接读取 MySQL、PostgreSQL 等数据源的工作原理。目前已有 1788 人学习下载。案例部分包含数据准备、DataStream 与 Flink SQL 两种接入方式，并给出 MySQL 到 Doris 的 Streaming ETL 全流程，涵盖环境搭建、配置文件编写、任务启动与测试验证；同时强调开启 Binlog、设置检查点等关键细节，提供完整代码示例。阅读后可掌握 Flink CDC 核心技术概念与原理，独立完成实时数据同步项目部署，确保数据一致性与任务可靠性，对实时数仓建设与数据同步等场景具有直接帮助作用。

1. Flink CDC 3.0 到底是什么：先搞清楚它和 2.x 的差别再动手

做实时数仓的开发者，最近应该都被 Flink CDC 3.0 刷屏了。如果你还在用 2.x 那套手动写 DataStream Source 去接 MySQL Binlog，再自己整 Debezium 反序列化，那你应该明白 3.0 最大的变化是什么——它把 MySQL、Doris 这类连接器做成了 Pipeline 模式，配一个 YAML 文件就能把整库同步跑起来，不再需要写一行 Java 代码。这篇文章我会带你从 MySQL Binlog 开启、Flink 1.18 环境准备，一路走到 DataStream 方式、Flink SQL 方式，最后用 3.0 的 Pipeline 配置把 MySQL 数据实时同步到 Doris，每个环节都给你能直接抄的配置和踩坑记录。

先说一个最容易翻车的点：很多人以为 Flink CDC 3.0 只依赖 Flink 本身，结果下载了flink-cdc-3.0.0-bin.tar.gz之后发现里面压根没有内置的 Doris 和 MySQL connector，必须手动往 lib 目录里丢flink-cdc-pipeline-connector-doris-3.0.0.jar和flink-cdc-pipeline-connector-mysql-3.0.0.jar。第一次装没经验的人，很容易卡在这一步大半天。这套环境适合谁？有一定 Flink 基础、想快速把 MySQL 实时同步到 Doris 做 OLAP 分析的工程师，或者正在做数仓异构数据同步选型的技术负责人。如果你是零基础，建议先把 Flink 的基本 API 和集群启动跑一遍再来，否则后面排查问题会有点吃力。下面我们直接从环境准备开始，按我实际装过的路径一步步走。

2. 环境准备与 MySQL Binlog 配置：装错一个参数整条链路都不亮

2.1 Flink 与 Flink CDC 的版本选型

先把版本对应关系列清楚，这一步选错后面全是坑。Flink CDC 3.0.0 官方推荐搭配 Flink 1.18.0 使用，我这次也是按照这个组合装的。JDK 用 1.8，Maven 编译目标也设成 8，不要一上来就上 JDK 17，Flink 1.18 虽然支持，但 CDC 的序列化组件在 JDK 17 下偶尔会有模块访问报错，少给自己找麻烦。

下载清单我给列一下，你在装之前先核对一遍：

Flink CDC 3.0 的flink-cdc-3.0.0-bin.tar.gz解压之后，目录结构和 Flink 本身是两套东西。这个包是给 Pipeline 模式用的独立工具，不是 Flink 的插件包。很多人一开始把它解压到 Flink 的 lib 目录下，那是错的。我一般习惯把它单独解压到/opt/module/flink-cdc-3.0.0，跟 Flink 目录平级，两边互不干扰。

2.2 MySQL 端 Binlog 开启：my.cnf 里的四个关键配置

MySQL 要作为 CDC 数据源，必须开启 Binlog，而且格式必须设成 row。这一步是整条链路的起点，没开 Binlog 的话 Flink CDC 连接器虽然能启动，但日志里会一直报找不到 binlog 文件的错。

登录 MySQL 先验证一下当前 Binlog 状态：

mysql -uroot -p000000 mysql> SHOW VARIABLES LIKE 'log_bin';

如果看到OFF，就去改配置文件。MySQL 8.0 的配置文件路径一般是/etc/my.cnf，用 sudo 编辑：

sudo vim /etc/my.cnf

在[mysqld]段下追加以下配置：

server-id = 1 log-bin = mysql-bin binlog_format = row binlog-do-db = test binlog-do-db = test_route

说明一下这几个参数的含义：server-id是 MySQL 在复制链路里的唯一标识，Flink CDC 连接器也会以某个 server-id 去模拟一个从库拉取 Binlog，所以 MySQL 自身的 server-id 不能和 CDC 里配置的冲突，否则会出现拉取不到数据的情况。log-bin指定 Binlog 文件的前缀，生成的文件会叫mysql-bin.000001这样递增的编号。binlog_format必须设成row，因为只有 row 格式才记录每行数据变更前后的完整镜像，基于 statement 格式的话 Flink CDC 拿不到具体的数据变化内容。binlog-do-db是白名单，只对指定的数据库开启 Binlog。原文里只加了test和test_route两个库，如果你的业务库更多，就继续追加多行。

改完配置必须重启 MySQL 才能生效：

sudo systemctl restart mysqld

重启后再查一次log_bin，确认是ON。这里有个坑——如果你原来 MySQL 里已经跑着其他复制任务，改server-id可能影响已有的主从关系。我实际遇到过把 server-id 改成 1 之后，原本的一主一从复制直接断掉的情况。建议先确认机器上没有其他复制链路，再动这个参数。

2.3 MySQL 建库建表与插入测试数据

Binlog 开好之后，把测试库和表建出来。原文里给了两个库，test和test_route，每个库下面建三张表t1、t2、t3，字段是id和name，t3稍有不同，第二个字段是sex。

CREATE 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) );

注意test_route库的t2表，原文插入的数据是1004、1005、1006，跟test库的t1、t2数据并不一样。这个设计是故意的，后面做路由变更测试时，正好可以验证目标库和源库表数据不一致的情况下，同步会不会串数据。

插入测试数据：

USE test; INSERT INTO t1 VALUES('1001','zhangsan'); INSERT INTO t1 VALUES('1002','lisi'); INSERT INTO t1 VALUES('1003','wangwu'); INSERT INTO t2 VALUES('1001','zhangsan'); INSERT INTO t2 VALUES('1002','lisi'); INSERT INTO t2 VALUES('1003','wangwu'); INSERT INTO t3 VALUES('1001','F'); INSERT INTO t3 VALUES('1002','F'); INSERT INTO t3 VALUES('1003','M'); USE test_route; INSERT INTO t1 VALUES('1001','zhangsan'); INSERT INTO t1 VALUES('1002','lisi'); INSERT INTO t1 VALUES('1003','wangwu'); INSERT INTO t2 VALUES('1004','zhangsan'); INSERT INTO t2 VALUES('1005','lisi'); INSERT INTO t2 VALUES('1006','wangwu'); INSERT INTO t3 VALUES('1001','F'); INSERT INTO t3 VALUES('1002','F'); INSERT INTO t3 VALUES('1003','M');

数据插完之后，先在 MySQL 命令行里查一下SHOW MASTER STATUS;，记下当前的 Binlog 文件名和 Position。这个信息后面排查问题时很有用，可以确认 CDC 连接的 Binlog 位点是否在推进。

2.4 Flink 1.18.0 集群启动与检查点配置

Flink 集群的搭建这里不展开细讲，默认你已经把flink-1.18.0解压到/opt/module下。有一个关键配置必须提前改，否则 CDC 任务跑着跑着会因为检查点超时被重启——Flink 默认的检查点间隔是无穷大，也就是关闭状态。在conf/flink-conf.yaml里追加：

execution.checkpointing.interval: 5000 state.backend: hashmap state.checkpoints.dir: hdfs://hadoop102:8020/flinkCDC

原文里启动了 HDFS 集群之后才启动 Flink，这步顺序不要乱。state.backend设成hashmap是 CDC 任务的常规选择，因为 CDC 的 Source 状态不大，主要是 Binlog 位点信息，不需要用 RocksDB。state.checkpoints.dir指向 HDFS 路径，如果没配 HDFS，Flink 默认用 JobManager 本地目录，任务一重启状态就丢了，做不到断点续传。

依次启动：

start-dfs.sh bin/start-cluster.sh

启动完访问hadoop102:8081确认 Web UI 正常。这里有个小坑：如果 HDFS 没起来就先把 Flink 集群启动了，后面提交 CDC 任务时状态后端初始化会报FileSystem相关的异常。我一般是先确保 HDFS 的 8020 端口通，再启动 Flink。

3. DataStream 方式读取 MySQL：从 Maven 依赖到 Savepoint 断点续传

3.1 Maven 依赖清单与打包配置

如果你需要在代码里灵活控制 CDC 的启动模式、反序列化方式，那 DataStream 方式是首选。创建一个 Maven 工程，pom.xml里的关键依赖如下：

<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>

打包插件建议用maven-assembly-plugin，把依赖打成一个 fat jar，否则提交到集群后到处缺类。注意flink-connector-mysql-cdc的 groupId 是com.ververica，不是org.apache.flink。我第一次用的时候按 Flink 官方依赖的习惯写成了org.apache.flink，发现 3.x 版本根本拉不到包，后来才意识到 Flink CDC 虽然是 Flink 社区维护，但发布时沿用了 Ververica 的坐标。这个细节最容易让熟手翻车。

3.2 实现 MySqlSource 并配置 Checkpoint

核心代码逻辑不长，但每一段都有讲究。直接给一个能跑的版本：

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.Properties; public class FlinkCDCDataStreamTest { public static void main(String[] args) throws Exception { // 1. 准备流处理环境 StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); env.setParallelism(1); // 2. 开启检查点 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.days(1L), Time.minutes(1L))); env.setStateBackend(new HashMapStateBackend()); env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop102:8020/flinkCDC"); System.setProperty("HADOOP_USER_NAME", "atguigu"); // 3. 创建 Flink-MySQL-CDC 的 Source MySqlSource<String> mySqlSource = MySqlSource.<String>builder() .hostname("hadoop103") .port(3306) .databaseList("test") .tableList("test.t1") .username("root") .password("000000") .deserializer(new JsonDebeziumDeserializationSchema()) .startupOptions(StartupOptions.initial()) .build(); // 4. 使用 CDC Source 从 MySQL 读取数据 DataStreamSource<String> mysqlDS = env.fromSource( mySqlSource, WatermarkStrategy.noWatermarks(), "MySqlSource"); // 5. 打印输出 mysqlDS.print(); // 6. 执行任务 env.execute(); } }

逐段说明一下。检查点部分，enableCheckpointing(3000L, CheckpointingMode.EXACTLY_ONCE)意思是每 3 秒做一次精确一次的检查点。很多人以为 CDC 任务把并行度设成 1 就不用检查点了，这是大错特错——Flink CDC 是把 Binlog 位点存在状态里的，不开启检查点，任务一重启就从最早位点重新读全量数据，目标库会出现大量重复数据。RETAIN_ON_CANCELLATION表示手动取消任务时保留最后一次检查点数据，这样后续可以从 Savepoint 恢复。setMinPauseBetweenCheckpoints(3000L)限制两次检查点之间至少间隔 3 秒，防止频繁检查点压垮磁盘。setCheckpointStorage把检查点落到 HDFS，配合System.setProperty("HADOOP_USER_NAME", "atguigu")解决 HDFS 权限问题。

再看 Source 构建。databaseList("test")只监听test库，tableList("test.t1")更进一步限定到test.t1表，两者是包含关系，不写tableList就监听整个库。deserializer(new JsonDebeziumDeserializationSchema())把 Debezium 的 SourceRecord 转成 JSON 字符串输出，每条变更记录是一个包含before、after、op字段的 JSON。startupOptions(StartupOptions.initial())表示任务第一次启动时先做一次全量快照，读取当前的表数据，然后从快照结束时的 Binlog 位点继续监听增量变化。

startupOptions有几个可选值，我平时最常用的是initial和latest。earliest是无论 Binlog 被清理过没有，都从最早位点读，如果 Binlog 文件已经被 MySQL 清理，会直接报错，生产环境不建议用。latest是从任务启动时刻开始只读增量，适用于已经做过全量数据初始化的场景。timestamp和specificOffset用于指定时间点和位点启动，一般排查问题时偶尔用。

3.3 打包提交与 Savepoint 重启验证

打包上传到服务器后，按下面的顺序启动。先确认 HDFS 和 Flink 集群都活着：

# 在 hadoop102 上 start-dfs.sh # 启动 Flink 集群 bin/start-cluster.sh # 提交 CDC 任务 bin/flink run -m hadoop102:8081 -c com.atguigu.cdc.FlinkCDCDataStreamTest ./flink-cdc-test.jar

任务跑起来后，打开 TaskManager 日志，能看到从 MySQL 读取的存量数据一行一行打出来，从1001开始。这时不要急着看效果，先做一个 Savepoint，验证断点续传：

# 为当前任务创建 Savepoint bin/flink savepoint <JobId> hdfs://hadoop102:8020/flinkCDC/save

拿到 Savepoint 路径后，在 Web UI 上取消当前 Job。接着往 MySQL 的test.t1里插几条新数据：

USE test; INSERT INTO t1 VALUES('1004','zhaoliu'); UPDATE t1 SET name = 'zhangsan2' WHERE id = '1001'; DELETE FROM t1 WHERE id = '1002';

然后从 Savepoint 重启任务：

bin/flink run -s hdfs://hadoop102:8020/flinkCDC/save/savepoint-xxxx-xxxx -c com.atguigu.cdc.FlinkCDCDataStreamTest ./flink-cdc-test.jar

重启后再看日志，你会发现没有重新打印全量数据，而是直接输出刚才插入、修改、删除的几条变更记录。这就是 CDC 断点续传的核心效果——Binlog 位点存在状态里，从 Savepoint 恢复后接着上次的位置继续读。如果这一步输出的是全量数据，说明状态没有正确恢复，去查 Savepoint 路径是否写对、任务名是否一致。

4. Flink SQL 方式读取 MySQL：DDL 配 5 个参数就能跑

4.1 Flink SQL CDC 的原理和适用场景

DataStream 方式灵活，但代码量不小。如果你只是想把 MySQL 的表映射成 Flink 表，然后做简单的查询或者写入下游，Flink SQL 方式明显更轻量——在StreamTableEnvironment里执行一段CREATE TABLEDDL，把connector指定为mysql-cdc，之后这张表就能当普通 Flink 表用了。

Flink SQL 方式的底层也是调用 MySQL CDC 连接器，但把参数全部收口到 WITH 选项里，不需要引入 Java 类。这种方式适合数据同步链路已经跑通、后续只想通过 SQL 做过滤或 join 的场景。注意一点：SQL 方式下的 CDC 表，DDL 里必须声明主键，而且要用PRIMARY KEY NOT ENFORCED的写法，因为它不检查客户端约束，只做语义声明。

4.2 编写 FlinkSQLTest 完整代码

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); tableEnv.executeSql( "CREATE TABLE t1 (\n" + " id STRING PRIMARY KEY NOT ENFORCED,\n" + " name STRING\n" + ") WITH (\n" + " 'connector' = 'mysql-cdc',\n" + " 'hostname' = 'hadoop103',\n" + " 'port' = '3306',\n" + " 'username' = 'root',\n" + " 'password' = '000000',\n" + " 'database-name' = 'test',\n" + " 'table-name' = 't1'\n" + ")"); Table table = tableEnv.sqlQuery("SELECT * FROM t1"); table.execute().print(); } }

这里需要注意两个地方。第一，Flink SQL 的表名可以跟 MySQL 的表名不一致，比如你在 DDL 里把表名写成mysql_t1_bak，只要table-name参数指向test.t1，读的还是 MySQL 的t1表。第二，database-name和table-name不支持正则匹配多个表，如果监听多张表，得建多个 CDC 表或用UNION ALL关联起来。SQL 方式的启动模式默认是initial，跟 DataStream 方式一致，首次启动做全量快照然后接增量。

直接运行这个类，IDEA 里会看到控制台输出全量数据。把打包后的 JAR 提交到 Flink 集群，行为完全一样。SQL 方式的检查点配置没法通过 DDL 控制，依赖 Flink 集群全局的flink-conf.yaml，所以我前面才强调要先改检查点间隔再启动集群。

4.3 Flink SQL 方式的局限性对比

用 Flink SQL 方式做 CDC，有一个绕不开的痛：如果 MySQL 表结构发生了变化，比如新增一列，Flink 端不会自动感知，除非重启任务并重新执行 DDL。这一点在 Flink CDC 3.0 的 Pipeline 模式下得到了解决——Pipeline 模式会自动捕获 DDL 变更并同步到 Doris。所以你如果是要做 MySQL 到 Doris 的整库同步，直接用第 5 章的 Pipeline 方式，别再用 SQL 方式一张表一张表去搞，省下来的时间足够你把 Doris 的集群参数调一遍。

5. 从 MySQL 到 Doris 的 Pipeline 同步：3.0 的核心玩法，一个 YAML 搞定

5.1 安装 Flink CDC 3.0 并补齐 connector

Flink CDC 3.0 的 Pipeline 模式是一等公民，不需要写代码。它把同步任务定义成一个 YAML 文件，source 和 sink 的 connector 实现都封装在 lib 目录下的 JAR 包里。安装步骤很简单：

tar -zxvf flink-cdc-3.0.0-bin.tar.gz -C /opt/module/ cd /opt/module/flink-cdc-3.0.0/lib/

然后把两个连接器 JAR 拖进lib目录：

cp flink-cdc-pipeline-connector-mysql-3.0.0.jar /opt/module/flink-cdc-3.0.0/lib/ cp flink-cdc-pipeline-connector-doris-3.0.0.jar /opt/module/flink-cdc-3.0.0/lib/

注意这里说的是 Pipeline 模式的 connector，包名和 DataStream 方式用的flink-connector-mysql-cdc完全不同。把flink-connector-mysql-cdc丢到 Pipeline 的 lib 下没有任何作用，flink-cdc.sh脚本只认flink-cdc-pipeline-connector-*系列。这个区分是 3.0 最容易装错的地方，没有之一。

5.2 同步变更配置：mysql-to-doris.yaml 逐行解读

在 Flink CDC 目录下创建job文件夹，写一个同步配置文件：

mkdir -p /opt/module/flink-cdc-3.0.0/job cd /opt/module/flink-cdc-3.0.0/job vim mysql-to-doris.yaml

内容如下：

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: 1

逐项解读。source 里tables: test.\.*是个正则表达式，匹配test库下所有表。注意 YAML 转义，.*要写成\.*，否则 YAML 解析时点号会被吞掉。server-id: 5400-5404是给 MySQL CDC 连接器申请的从库 ID 范围，每个并行子任务占用一个 ID。如果并行度是 3，至少分配 3 个 ID，这个范围要留够。同一条机器上如果有其他 CDC 任务也在读同一个 MySQL，server-id 不能重叠，否则会造成 Binlog 拉取互相干扰，甚至被 MySQL 断开连接。server-time-zone: UTC+8确保时间类型字段解析时不会因为时区偏移差 8 小时。

sink 部分，fenodes: node01:7030是 Doris FE 的地址，7030 是 FE 的 HTTP 端口，不是 MySQL 协议端口 9030。table.create.properties.light_schema_change: true是 Doris 1.2 之后支持的轻量 Schema 变更功能，开启后加列操作不涉及数据重写，同步 DDL 的延迟能控制在毫秒级。replication_num: 1是因为测试环境只有一台 BE，如果你有多台 BE，设成 2 或 3 保证数据副本。

pipeline 部分，parallelism: 1表示整个同步管线并行度为 1。对于轻量测试可以，但生产环境建议提高到 3 以上，同时把 server-id 范围相应扩大。所有配置项里最容易被忽略的是server-time-zone，MySQL 的datetime类型不带时区信息，如果 Flink CDC 和 Doris 之间的时区不一致，同步过去的时间字段会整体偏移几个小时。

5.3 启动 Doris 集群与 Flink CDC 任务

Doris 的部署这里不展开，默认 FE 和 BE 已经装好。启动顺序是 Doris FE、Doris BE，再启动 FlinkCDC 同步任务：

# 启动 Doris FE cd /opt/module/doris/fe/bin ./start_fe.sh # 启动 Doris BE cd /opt/module/doris/be/bin ./start_be.sh # 在 Doris 中创建两个目标数据库（通过 MySQL 协议连 FE） mysql -uroot -p000000 -P9030 -h hadoop102 mysql> CREATE DATABASE test; mysql> CREATE DATABASE doris_test_route; # 启动 Flink CDC 同步变更任务 cd /opt/module/flink-cdc-3.0.0 bin/flink-cdc.sh job/mysql-to-doris.yaml

任务启动后，Flink CDC 会在 Doris 端自动建表。你不需要在 Doris 里手动创建t1、t2、t3，sink 的table.create.properties会自动根据 MySQL 侧的表结构映射创建。第一次跑的时候去 Doris 的test库用SHOW TABLES;看一眼，如果没有自动建表，优先查两个地方：FE 日志里有没有 Schema 同步相关的报错，以及 Doris BE 节点是否被 FE 正确识别。BE 没注册上去的话，建表会一直处于PENDING状态。

5.4 路由变更配置：doris_test_route 的映射玩法

Flink CDC 3.0 的路由功能可以让你在同步时做库名和表名的重新映射。配置文件和同步变更基本一致，sink 部分会多几个路由参数：

source: type: mysql hostname: hadoop103 port: 3306 username: root password: "000000" tables: test_route.\.* server-id: 5400-5404 server-time-zone: UTC+8 sink: type: doris fenodes: hadoop103:7030 username: root password: "000000" table.create.properties.light_schema_change: true table.create.properties.replication_num: 1 route: - source-table: test_route.\.* sink-table: doris_test_route.\.* pipeline: name: Sync MySQL Database to Doris with route parallelism: 1

route的作用是把源库test_route的所有表同步到目标库doris_test_route下。这里我踩过一个坑：如果目标库在 Doris 里不存在，Flink CDC 会不会自动创建？我实际测下来，3.0 的 sink 会尝试自动建库，但它建库依赖 FE 的root权限，如果你的账号只有部分库的权限，就会报Access denied。建议在启动任务前手动把目标库建好，不要指望自动建库。

路由不只是库名映射，还可以做强匹配和正则匹配。比如你想把test_route.t1同步成doris_test_route.t1_bak，可以写：

route: - source-table: test_route.t1 sink-table: doris_test_route.t1_bak

5.5 全量加增量同步的完整验证

任务启动后，去 MySQL 的test库里做几种操作，然后在 Doris 里观察结果：

1. INSERT INTO t1 VALUES('1004','zhaoliu');
2. UPDATE t1 SET name = 'zhangsan_new' WHERE id = '1001';
3. ALTER TABLE t1 ADD COLUMN age VARCHAR(255);

每一步操作之后，稍等几秒，去 Doris 里查：

USE test; SELECT * FROM t1; SHOW CREATE TABLE t1;

你能看到新增的数据、修改的数据都实时出现了，而且age列也被自动加到了 Doris 的表结构里。这就是 Flink CDC 3.0 对比 2.x 最大的进步——DDL 同步不用你再手动停任务改表结构。

这里有一个细节值得注意：ALTER TABLE加列操作在源 MySQL 执行时，如果 Doris 侧开启了light_schema_change，加列是秒级的。但如果 Doris 版本低于 1.2，或者 BE 节点上的数据量很大，加列操作可能会触发数据重写，同步延迟会明显拉长。我在测试环境 1.2.4 上验证过，几十万行数据加列几乎无感知，但到了千万级，建议先在 Doris 侧手工建好新表再切流。

6. 装完怎么验证：从日志、数据一致性、DDL 同步三个维度检查

验证一套 Flink CDC 环境是不是真的装通了，不能只看任务不报错，那只是最基本的门槛。我每次装完都会强制自己走一遍下面三个验证步骤，任何一个不过关都不能算完事。

第一步，验证 Binlog 位点确实在推进。Flink CDC 任务启动后，回到 MySQL 执行：

SHOW MASTER STATUS;

记下当前的Position值，等一分钟再查一次，如果 Position 变了，说明 MySQL 的 Binlog 有新的写入；再去 TaskManager 日志里看 Flink CDC 有没有同步打印这些变更。如果 MySQL 的 Position 一直在变，而 Flink 日志里没有任何输出，大概率是databaseList或tableList写错了，或者 MySQL 用户权限不足。

第二步，验证全量加增量数据一致性。拿第 3 章的 DataStream 任务和第 5 章的 Pipeline 任务同时监听test.t1表，往 MySQL 插入一批数据，然后在两个消费端分别对id去重计数，数量一致就说明两条链路读到的 Binlog 事件没有丢。这一步对生产环境尤其重要——如果你只验证了 Pipeline 方式，DataStream 后续拿到代码里改一改逻辑，很可能因为反序列化配置不同产生脏数据。

第三步，验证 DDL 同步和 Schema 变更。在 MySQL 源库执行：

ALTER TABLE t1 ADD COLUMN email VARCHAR(255);

去 Doris 端执行SHOW CREATE TABLE t1;，如果能看到email字段，说明 DDL 同步链路是通的。这一步最容易出问题的场景是：源 MySQL 加了列，Doris 同步了，但 Flink SQL 方式建的 CDC 表没有映射新列——因为 Flink SQL 的表结构是启动时固定的。所以如果你在跑 Flink SQL 任务，切记 DDL 变更后必须重启任务重新执行 CREATE TABLE，这跟 Pipeline 方式自动同步 DDL 是两个路子。

最后补充一个我自己踩过很深的坑：临时文件占用导致 Doris BE 磁盘满，同步任务看起来还活着，但数据已经写不进去了。现象是 FE 日志里报Table is in corrupted state，BE 日志里有大量写超时的记录，但 Flink CDC 任务的 Web UI 显示 Running。排查下来是 Doris BE 的storage_root_path对应的磁盘写满了，清理掉旧的trash文件后，任务不需要重启就自己恢复了。从那以后我每次部署完都会顺手在机器上加一个 crontab，定时检查 Doris BE 的数据盘使用率，超过 85% 就告警。Flink CDC 这个工具本身装起来不复杂，但你要真的信它能扛住生产环境，就得把这种边缘情况也验证到位。希望这篇安装笔记能帮你在 Flink CDC 3.0 到 Doris 的同步链路上少走几趟弯路。

本文还有配套的精品资源，点击获取