企业官网建设流程全解析

简介：本资源是面向大数据开发工程师的Flink CDC 3.0实战指南，聚焦MySQL到Doris的实时数据同步场景，适用于具备Flink基础、需落地Streaming ETL方案的中高级研发人员。文档由尚硅谷研究院出品，系统梳理CDC核心原理（含Binlog与查询式对比）、flink-cdc-connectors组件机制，并提供从环境搭建、MySQL Binlog开启、检查点配置，到DataStream/Flink SQL双路径实操及MySQL→Doris全链路同步的完整步骤，覆盖数据准备、建库建表、任务提交与结果验证等关键环节。资源为1个145KB的docx文件，内容结构清晰，含代码片段、命令行操作示例及配置要点说明，便于快速复现与工程化迁移。目前已有1789人学习下载，读者可直接获取可运行的配置模板、典型问题规避提示及生产级稳定性保障建议，显著降低实时同步项目落地门槛。

1. 为什么用 Flink CDC 3.0 做 MySQL → Doris 同步，而不是写个定时脚本或改用 Kafka 中转？

你有没有遇到过这种场景：业务库 MySQL 里一张用户行为表每秒新增 200 条记录，下游 BI 系统要基于这张表做实时漏斗分析，但每次跑 SQL 都卡在“查全量 + 拉增量”上——要么加索引拖慢写入，要么用WHERE update_time > ?漏掉事务未提交的数据，更别说跨库路由、DDL 变更、断点续传这些玄学问题。这时候翻文档发现 Flink CDC 3.0 已原生支持 MySQL Binlog 全量+增量一体化捕获、自动 schema 演化、Doris 批流一体 sink，并且配置文件驱动、无需写 Java/SQL 代码就能跑通端到端 pipeline。这不是“又一个实时同步工具”，而是把 CDC 这个黑匣子真正封装成可交付模块的分水岭版本。本文档面向两类人：一是正在做毕设/课设需要交一份完整可运行、含环境搭建→代码实现→故障复现→验收验证全流程的学生；二是已在产线用 Flink 但还没落地过 CDC 场景的工程师——你要的不是概念图，是能直接cp过去改三行参数就上线的方案。我们不讲“CDC 是什么”，只解决“怎么让 MySQL 的每一行变更，毫秒级、不丢不重、带结构信息地落到 Doris 里，并且答辩时能现场演示增删改查全过程”。

2. 总体架构设计：三层解耦结构与四个核心模块的职责划分

Flink CDC 3.0 的 MySQL → Doris 同步不是单点工具链，而是一个职责清晰的三层架构：数据源层（MySQL）→ 计算层（Flink CDC Pipeline）→ 目标层（Doris）。其中计算层被进一步拆解为四个不可替代的模块：Binlog 解析器、全量快照引擎、变更事件路由器、Doris 写入适配器。这个结构决定了它比传统 DataX/Sqoop 更适合实时场景，也比自研 Debezium + Flink SQL 方案更易维护。下面逐层说明设计逻辑和模块边界。

2.1 数据源层：MySQL 必须开启的 Binlog 配置及其底层原理

Flink CDC 3.0 依赖 MySQL 的 Row-based Binlog 实现精确变更捕获。这要求 MySQL 不仅开启 Binlog，还要满足三个硬性条件：

• server-id 唯一性：集群中每个 MySQL 实例必须有全局唯一 server-id（如server-id = 1），否则 Flink CDC 无法建立独立复制通道；
• binlog_format=row：必须设为ROW模式，因为只有该模式下 Binlog 才记录每行数据的完整前后镜像（before/after image），STATEMENT或MIXED模式会导致 UPDATE/DELETE 事件丢失字段级变更细节；
• binlog-do-db 精确声明：不能写binlog-do-db=test%这类模糊匹配，Flink CDC 3.0 要求显式列出所有需监听的库名（如binlog-do-db=test和binlog-do-db=test_route），否则启动时会报No tables captured错误——这是学生最容易翻车的第一步。

提示：修改/etc/my.cnf后必须执行sudo systemctl restart mysqld，且重启后需验证SHOW VARIABLES LIKE 'log_bin';返回ON，SHOW VARIABLES LIKE 'binlog_format';返回ROW，缺一不可。

2.2 计算层：Flink CDC Pipeline 的四大模块如何协同工作

Flink CDC 3.0 将整个同步流程抽象为声明式 pipeline，其核心是四个模块的流水线协作：

• Binlog 解析器：基于 Debezium 封装，负责连接 MySQL 并解析 Binlog event 流，输出标准化的SourceRecord对象（含op类型、ts_ms时间戳、before/after字段快照）；
• 全量快照引擎：在首次启动时自动触发对指定表的全量读取（SELECT * FROM t1），并将结果按主键哈希分片并行拉取，完成后无缝切换至 Binlog 增量消费——关键在于它与 Binlog 位点严格对齐，避免“先拉全量再启 Binlog”导致中间数据丢失；
• 变更事件路由器：根据配置中的tables: test_route\..*正则表达式，将不同库表的变更事件路由到对应 Doris 表，支持库表名映射（如test.t1→doris_test.t1）；
• Doris 写入适配器：将SourceRecord转换为 Doris 支持的 Stream Load 格式（JSON 或 CSV），通过 HTTP 接口批量写入，自动处理INSERT/UPDATE/DELETE语义（DELETE 通过_delete_sign=1标记实现）。

这四个模块全部由flink-cdc-pipeline-connector-mysql-3.0.0.jar和flink-cdc-pipeline-connector-doris-3.0.0.jar封装，用户只需配置 YAML 文件，无需接触底层 API。

2.3 目标层：Doris 侧必须预建的表结构与关键参数

Doris 并非无脑接收数据，它要求目标表具备明确的 Schema 和特定属性才能兼容 CDC 的 Upsert 语义：

• 主键模型（Primary Key Model）：必须使用PRIMARY KEY模型建表（而非 Duplicate/Aggregate），因为只有该模型支持按主键更新（REPLACE语义），DELETE操作才有效；
• Light Schema Change 开关：配置中table.create.properties.light_schema_change: true表示允许在 Doris 表新增列时自动扩展 Schema（如 MySQL 新增age INT字段，Doris 会自动添加该列），否则会因 Schema 不匹配导致写入失败；
• 副本数设置：table.create.properties.replication_num: 1在单节点测试环境足够，但生产环境建议设为3以保障高可用——注意此参数仅在 Doris 自动建表时生效，若手动建表则需提前指定。

建表示例（Doris CLI）：

CREATE DATABASE IF NOT EXISTS test; USE test; CREATE TABLE IF NOT EXISTS t1 ( id VARCHAR(255) PRIMARY KEY COMMENT "主键", name VARCHAR(255) COMMENT "姓名" ) ENGINE=OLAP PRIMARY KEY(id) COMMENT "MySQL t1 表同步目标" DISTRIBUTED BY HASH(id) BUCKETS 10 PROPERTIES( "replication_num" = "1" );

2.4 配置驱动的核心优势：YAML 文件即部署单元

Flink CDC 3.0 最大变革是废弃 Java/SQL 编码方式，改用纯 YAML 驱动 pipeline。这意味着：

• 环境隔离：开发、测试、生产只需替换mysql-to-doris.yaml中的hostname/port/username，无需重新编译 JAR；
• 多任务管理：一个 Flink CDC 实例可同时运行多个 YAML 文件（如mysql-to-doris.yaml和pg-to-doris.yaml），通过bin/flink-cdc.sh job/*.yaml批量启动；
• 版本可追溯：YAML 文件纳入 Git，每次变更都有 commit 记录，回滚只需git checkout上一版配置。

典型mysql-to-doris.yaml结构如下（已按生产环境加固）：

source: type: mysql hostname: hadoop103 port: 3306 username: root password: "000000" tables: test\..* # 注意：正则需转义点号，test.* 会匹配失败 server-id: 5400-5404 # 必须是范围，单值如 5400 会报错 server-time-zone: "Asia/Shanghai" # 与 MySQL 时区一致，避免时间字段错乱 scan.startup.mode: initial # 初始模式：全量+增量，生产环境推荐 sink: type: doris fenodes: hadoop102:7030 username: root password: "000000" table.create.properties.light_schema_change: true table.create.properties.replication_num: 1 pipeline: name: Sync MySQL test DB to Doris parallelism: 1 # 单库同步设为 1，避免 Binlog 位点竞争 checkpoint.interval: 10000 # 每 10 秒保存一次检查点，与 Flink 集群配置对齐

2.5 避坑：Flink CDC 3.0 MySQL → Doris 同步的五大高频翻车点

实际部署中，80% 的失败源于配置细节疏忽。以下是我在某高校大数据实验室带学生做课设时，整理出的最常踩坑的五个点，按“现象 → 原因 → 解决”给出可立即执行的排查路径：

2.5.1 现象：Flink CDC 启动后日志显示No tables captured，但 MySQL 确实存在配置的库表

原因：binlog-do-db未精确匹配，或 MySQL 用户权限不足（缺少SELECT,RELOAD,SHOW DATABASES,REPLICATION SLAVE,REPLICATION CLIENT权限）
解决：

1. 登录 MySQL 执行SHOW VARIABLES LIKE 'binlog_do_db';确认输出包含test和test_route；
2. 执行SHOW GRANTS FOR 'root'@'%';检查权限，缺失则执行：

GRANT SELECT, RELOAD, SHOW DATABASES, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'root'@'%'; FLUSH PRIVILEGES;

2.5.2 现象：全量同步完成后，MySQL 新增数据未出现在 Doris，且 Flink CDC 日志无 Binlog 消费记录

原因：server-id在 MySQL 配置中重复（如多实例共用server-id = 1），导致 Binlog 复制通道被拒绝
解决：

1. 查看 MySQL 错误日志：sudo tail -f /var/log/mysqld.log，搜索Duplicate server-id；
2. 修改/etc/my.cnf，为每个 MySQL 实例分配唯一server-id（如hadoop103设为5401，hadoop102设为5402）；
3. 重启 MySQL：sudo systemctl restart mysqld。

2.5.3 现象：Doris 中数据出现重复，同一主键 ID 对应多条记录（非 Upsert 效果）

原因：Doris 目标表未使用PRIMARY KEY模型，或建表时未声明PRIMARY KEY(id)
解决：

1. 执行DESCRIBE test.t1;检查Key列是否为PRI；
2. 若为DUPLICATE模型，删除重建：

DROP TABLE test.t1; CREATE TABLE test.t1 (id VARCHAR(255) PRIMARY KEY, name VARCHAR(255)) ENGINE=OLAP PRIMARY KEY(id) DISTRIBUTED BY HASH(id) BUCKETS 10;

2.5.4 现象：MySQL 执行ALTER TABLE t1 ADD COLUMN age INT;后，Flink CDC 报错Cannot convert field 'age'

原因：light_schema_change: true未生效，或 Doris 版本低于 1.2.0（该功能在 1.2.0+ 引入）
解决：

1. 确认 Doris 版本：mysql -h hadoop102 -P9030 -uroot -p000000 -e "SELECT VERSION();"；
2. 若版本正确，检查 YAML 中light_schema_change是否拼写为light_schema_change（注意下划线，非lightSchemaChange）；
3. 删除 Doris 中旧表，让 CDC 自动重建（DROP TABLE test.t1;）。

2.5.5 现象：Flink CDC 任务运行一段时间后 OOM（OutOfMemoryError），TaskManager 频繁重启

原因：parallelism: 1与checkpoint.interval: 10000不匹配，导致状态后端积压大量 Binlog 位点
解决：

1. 在flink-conf.yaml中增加状态后端配置：

state.backend: hashmap state.checkpoints.dir: hdfs://hadoop102:8020/flink/checkpoints state.savepoints.dir: hdfs://hadoop102:8020/flink/savepoints

2. 重启 Flink 集群后，再启动 CDC 任务；
3. 监控 HDFS 中flink/checkpoints目录大小，若持续增长超 1GB，需调大checkpoint.timeout至60000。

3. DataStream 与 Flink SQL 双模式实现：从编码到声明的演进路径

Flink CDC 3.0 提供两种接入方式：DataStream API（适合深度定制）和Flink SQL（适合快速验证）。二者本质都是构建 Source，但适用阶段不同——课设初期用 SQL 快速跑通链路，毕设深化时用 DataStream 实现字段过滤、数据脱敏等业务逻辑。下面分别给出可直接运行的代码及设计说明。

3.1 DataStream 模式：全量+增量一体化的 Java 实现与状态管理要点

DataStream 模式给予开发者最大控制权，尤其在需要处理复杂业务逻辑（如字段映射、空值过滤、多表关联）时不可替代。以下代码是尚硅谷案例的生产级增强版，重点强化了断点续传可靠性和异常容错能力：

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 org.apache.flink.streaming.api.functions.ProcessFunction; import org.apache.flink.util.Collector; import java.util.Properties; public class FlinkCDCDataStreamProduction { public static void main(String[] args) throws Exception { // TODO 1. 创建流环境并设置并行度 StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); env.setParallelism(1); // CDC source 必须单并行度，避免 Binlog 位点冲突 // TODO 2. 配置高可靠 Checkpoint（生产环境必选） env.enableCheckpointing(10000L, CheckpointingMode.EXACTLY_ONCE); env.getCheckpointConfig().setCheckpointTimeout(60000L); env.getCheckpointConfig().setMinPauseBetweenCheckpoints(5000L); env.getCheckpointConfig().enableExternalizedCheckpoints( CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION); env.setRestartStrategy(RestartStrategies.failureRateRestart( 3, Time.days(1L), Time.minutes(1L))); // 失败3次内1分钟内重启 env.setStateBackend(new HashMapStateBackend()); env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop102:8020/flink/checkpoints"); System.setProperty("HADOOP_USER_NAME", "atguigu"); // TODO 3. 构建 MySQL CDC Source（关键：startupOptions 精准控制） MySqlSource<String> mySqlSource = MySqlSource.<String>builder() .hostname("hadoop103") .port(3306) .databaseList("test") // 仅监听 test 库 .tableList("test.t1", "test.t2") // 显式指定表，避免 .* 匹配风险 .username("root") .password("000000") .serverId("5400-5404") // 必须是范围，单值无效 .deserializer(new JsonDebeziumDeserializationSchema()) // 输出 JSON 字符串 .startupOptions(StartupOptions.initial()) // 生产环境首选 initial .build(); // TODO 4. 从 Source 读取数据流 DataStreamSource<String> mysqlDS = env.fromSource( mySqlSource, WatermarkStrategy.noWatermarks(), "MySQL-CDC-Source" ); // TODO 5. 【可选】添加业务处理逻辑：过滤空 name 字段 mysqlDS.process(new ProcessFunction<String, String>() { @Override public void processElement(String value, Context ctx, Collector<String> out) throws Exception { // 解析 JSON，跳过 name 为空的记录（模拟业务规则） if (value.contains("\"name\":\"\"") || value.contains("\"name\":null")) { return; } out.collect(value); } }).print("Processed-Data"); // 输出到控制台，便于调试 env.execute("Flink-CDC-DataStream-Production"); } }

设计说明与参数解读：

• serverId("5400-5404")：必须传入范围字符串，Flink CDC 会从中随机选取一个 ID 作为复制客户端 ID，单值（如"5400"）会导致IllegalArgumentException；
• startupOptions(StartupOptions.initial())：initial模式最稳妥，首次启动先全量再增量；earliest要求 Binlog 保留全量历史，生产环境几乎不可用；
• JsonDebeziumDeserializationSchema()：输出为标准 Debezium JSON 格式，包含op（c=create,u=update,d=delete）、ts_ms（操作时间）、before/after（变更前后快照），后续可直接解析入库；
• ProcessFunction：此处演示如何插入轻量级业务逻辑，实际课设中可替换为MapFunction做字段转换，或KeyedProcessFunction做窗口聚合。

3.2 Flink SQL 模式：零代码快速验证的声明式实现与动态表管理

Flink SQL 模式是验证 CDC 链路是否通的最快方式，无需编译打包，直接在本地 IDE 运行main方法即可看到 MySQL 变更实时打印。其核心是将 MySQL 表注册为 Flink 动态表，然后用标准 SQL 查询：

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 FlinkSQLQuickStart { public static void main(String[] args) { StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); env.setParallelism(1); StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env); // TODO 1. 创建 MySQL CDC 动态表（自动建表，无需 Doris 侧预建） 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'," + " 'scan.startup.mode' = 'initial'" + ")" ); // TODO 2. 查询并打印（实时流式输出） Table result = tableEnv.sqlQuery("SELECT * FROM t1"); result.execute().print(); // 注意：此处不调用 env.execute()，Flink SQL 会自动触发执行 } }

关键设计点：

• 'connector' = 'mysql-cdc'：Flink 1.18+ 内置 connector，无需额外引入flink-connector-mysql-cdc依赖（但需确保lib/下有对应 JAR）；
• PRIMARY KEY NOT ENFORCED：声明主键但不强制校验，因为 CDC 数据来自外部，Flink 不做约束；
• scan.startup.mode = 'initial'：与 DataStream 的StartupOptions.initial()对应，保证首次启动读全量；
• result.execute().print()：调用execute()启动流任务，print()将变更事件输出到控制台，格式为+I[1001, zhangsan]（+I=insert,-U=update before,+U=update after,-D=delete）。

提示：此模式适合课设第一阶段——5 分钟内验证 MySQL → Flink 链路。若需导出到 Doris，只需将print()替换为executeSql("INSERT INTO doris_table SELECT * FROM t1")，前提是已注册 Doris connector。

3.3 双模式对比：何时用 DataStream，何时用 Flink SQL？

我的经验：带学生做课设时，强制要求第一周只用 Flink SQL 模式跑通全链路，第二周再切到 DataStream 加业务逻辑。这样避免陷入“环境没搭好，代码先写废”的死循环。

3.4 从 SQL 到 Pipeline：YAML 配置的平滑迁移路径

Flink CDC 3.0 的 YAML 模式并非取代前两者，而是更高阶的封装。它的迁移路径非常自然：

• Step 1：用 Flink SQL 验证test.t1表同步正常；
• Step 2：将 SQL 中的连接参数（hostname/port/username）提取为 YAML 的source段；
• Step 3：将INSERT INTO doris_table逻辑替换为 YAML 的sink段；
• Step 4：用bin/flink-cdc.sh job/mysql-to-doris.yaml启动，此时不再依赖 Flink 集群的sql-client，而是独立进程管理。

这种渐进式迁移让学生清晰看到：SQL 是逻辑描述，YAML 是部署契约，DataStream 是能力底座——三者不是互斥选项，而是同一问题的不同抽象层次。

4. MySQL 到 Doris 的 Streaming ETL 全流程实战：从环境搭建到增删改查验证

本章将手把手带你走完一个真实可交付的 Streaming ETL 项目：在三台虚拟机（hadoop102/hadoop103/hadoop104）上，完成 MySQL → Doris 的端到端同步，并现场演示 INSERT/UPDATE/DELETE 全操作闭环。所有命令均来自尚硅谷实操文档，我做了生产环境加固和错误预防处理，确保你复制粘贴就能成功。

4.1 环境准备：四组件版本对齐与依赖注入

Flink CDC 3.0 对组件版本极其敏感，版本错配是第二大翻车原因。以下是经验证的黄金组合（全部来自尚硅谷配套环境）：

依赖注入步骤（必须执行）：

1. 将flink-cdc-pipeline-connector-mysql-3.0.0.jar和flink-cdc-pipeline-connector-doris-3.0.0.jar复制到 Flink CDC 的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/

2. 将mysql-connector-java-8.0.31.jar复制到 Flink 的lib/目录（Flink CDC 本身不带 MySQL 驱动）：

cp mysql-connector-java-8.0.31.jar /opt/module/flink-1.18.0/lib/

3. 重启所有服务：先停 MySQL，再启 MySQL（加载新配置），再启 Flink，最后启 Doris。

提示：flink-cdc-3.0.0-bin.tar.gz解压后默认不含lib/目录，需手动创建并放入 JAR，否则启动时报ClassNotFoundException: com.mysql.cj.jdbc.Driver。

4.2 数据准备：双库双表结构与初始化数据脚本

为验证路由能力，我们准备两个 MySQL 数据库：test（直连同步）和test_route（路由到 Doris 的doris_test_route库）。建表语句已在第 2 章给出，此处提供一键初始化脚本init_mysql_data.sql：

-- 创建数据库 CREATE DATABASE IF NOT EXISTS test; CREATE DATABASE IF NOT EXISTS test_route; -- test 库建表 USE test; CREATE TABLE IF NOT EXISTS t1 (id VARCHAR(255) PRIMARY KEY, name VARCHAR(255)); CREATE TABLE IF NOT EXISTS t2 (id VARCHAR(255) PRIMARY KEY, name VARCHAR(255)); CREATE TABLE IF NOT EXISTS t3 (id VARCHAR(255) PRIMARY KEY, sex VARCHAR(255)); -- test_route 库建表 USE test_route; CREATE TABLE IF NOT EXISTS t1 (id VARCHAR(255) PRIMARY KEY, name VARCHAR(255)); CREATE TABLE IF NOT EXISTS t2 (id VARCHAR(255) PRIMARY KEY, name VARCHAR(255)); CREATE TABLE IF NOT EXISTS t3 (id VARCHAR(255) PRIMARY KEY, sex VARCHAR(255)); -- 插入初始数据（test 库） USE test; INSERT INTO t1 VALUES('1001','zhangsan'),('1002','lisi'),('1003','wangwu'); INSERT INTO t2 VALUES('1001','zhangsan'),('1002','lisi'),('1003','wangwu'); INSERT INTO t3 VALUES('1001','F'),('1002','F'),('1003','M'); -- 插入初始数据（test_route 库） USE test_route; INSERT INTO t1 VALUES('1001','zhangsan'),('1002','lisi'),('1003','wangwu'); INSERT INTO t2 VALUES('1004','zhangsan'),('1005','lisi'),('1006','wangwu'); INSERT INTO t3 VALUES('1001','F'),('1002','F'),('1003','M');

执行方式：

mysql -uroot -p000000 < init_mysql_data.sql

4.3 同步任务启动：两个 YAML 文件分别处理直连与路由场景

Flink CDC 3.0 支持同时运行多个 pipeline，我们创建两个配置文件：

4.3.1mysql-to-doris-direct.yaml：直连同步test库到 Doristest库

source: type: mysql hostname: hadoop103 port: 3306 username: root password: "000000" tables: test\..* server-id: 5400-5404 server-time-zone: "Asia/Shanghai" sink: type: doris fenodes: hadoop102:7030 username: root password: "000000" table.create.properties.light_schema_change: true table.create.properties.replication_num: 1 pipeline: name: Direct Sync test DB parallelism: 1 checkpoint.interval: 10000

4.3.2mysql-to-doris-route.yaml：路由同步test_route库到 Dorisdoris_test_route库

source: type: mysql hostname: hadoop103 port: 3306 username: root password: "000000" tables: test_route\..* server-id: 5400-5404 server-time-zone: "Asia/Shanghai" sink: type: doris fenodes: hadoop102:7030 username: root password: "000000" table.create.properties.light_schema_change: true table.create.properties.replication_num: 1 # 关键：路由规则，将 test_route.t1 映射到 doris_test_route.t1 table.prefix: doris_test_route. pipeline: name: Route Sync test_route DB parallelism: 1 checkpoint.interval: 10000

注意：table.prefix: doris_test_route.是路由核心，Flink CDC 会自动在 Doris 表名前加此前缀，因此无需在 Doris 中手动创建doris_test_route库，CDC 会自动建库建表。

4.4 全场景验证：INSERT/UPDATE/DELETE 实时同步效果观测

启动两个 pipeline 后，我们通过四步操作验证同步效果：

4.4.1 步骤一：观察 Doris 自动建表与全量数据同步

启动命令：

cd /opt/module/flink-cdc-3.0.0 bin/flink-cdc.sh job/mysql-to-doris-direct.yaml bin/flink-cdc.sh job/mysql-to-doris-route.yaml

等待 30 秒后，登录 Doris：

mysql -h hadoop102 -P9030 -uroot -p000000

执行：

SHOW DATABASES; -- 应看到 test 和 doris_test_route USE test; SELECT COUNT(*) FROM t1; -- 应返回 3（全量同步完成） SELECT * FROM t1; -- 应显示 1001/zhangsan 等三条记录

4.4.2 步骤二：MySQL INSERT 新数据，验证 Doris 实时追加

在 MySQL 中执行：

USE test; INSERT INTO t1 VALUES('1004','zhaoliu');

立即在 Doris 中查询：

SELECT * FROM test.t1 WHERE id='1004'; -- 应立刻返回一行

4.4.3 步骤三：MySQL UPDATE 数据，验证 Doris 主键更新

在 MySQL 中执行：

USE test; UPDATE t1 SET name='liqi' WHERE id='1001';

在 Doris 中查询：

SELECT * FROM test.t1 WHERE id='1001'; -- name 应变为 'liqi'

4.4.4 步骤四：MySQL DELETE 数据，验证 Doris 逻辑删除

在 MySQL 中执行：

USE test; DELETE FROM t1 WHERE id='1002';

在 Doris 中查询：

SELECT * FROM test.t1 WHERE id='1002'; -- 应无返回结果（PRIMARY KEY 模型自动删除）

提示：若DELETE不生效，请

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