企业官网建设流程全解析

简介：面向具备一定 Flink 基础、希望掌握实时数据同步链路的研发人员，这份由尚硅谷研究院编写的 Flink CDC 3.0 文档以 MySQL 变更捕获与写入 Doris 为主线，系统梳理 CDC 概念、基于查询与基于 Binlog 两类方案的差异，以及 flink-cdc-connectors 组件的使用方式。资源包共 1 个 docx 文件，约 145KB，内容集中为可跟随操作的技术文档，涵盖数据准备、DataStream 与 Flink SQL 两种实现路径，并给出 MySQL 到 Doris 的 Streaming ETL 环境搭建、配置文件编写、任务启动与测试步骤。已有 1789 人学习，适合用作项目落地前的方案比对和实操参照。读者可从中获得开启 Binlog、设置检查点等关键配置要点，理解全量加增量读取的流程，并借助完整示例快速搭建从源库到目标库的同步任务，减少联调与排错成本。

1. 版本先对齐：Flink CDC 3.0 这条链路装错一步就白跑

把一个 Flink CDC 3.0 的作业从 MySQL 实时同步到 Doris，听上去是写个 yaml 敲一行bin/flink-cdc.sh的事，真动手才发现是四五个组件互相限制版本：Flink 1.18.0 的 classpath 里必须能找到flink-connector-base，flink-connector-mysql-cdc3.0.0 的坐标一旦和 Flink 版本错开，作业会在提交阶段就抛 NoSuchMethod；MySQL 8.0.31 的binlog_format不是 ROW，source 初始化直接失败；Doris 1.2.4 的 FE、BE 没起全，sink 建表也会卡住。我见过最多的失败情形，是 MySQL 的log-bin根本没开，程序跑了三十秒退出，日志里只有一行含糊的 connect 异常，让人对着配置文件反复怀疑自己。这套环境适合已经能跑 Flink 作业、准备把业务库变更接到 Doris 做实时报表的开发者，下面从装什么版本一路写到装坏了怎么救。

2. 环境清单与 MySQL 侧准备：Flink 1.18.0、CDC 3.0.0、MySQL 8.0.31 的版本对齐

这一章先把“装什么、装在哪、版本怎么搭”确定下来。装环境最怕的不是步骤多，而是装完才发现某个组件的版本天生不适配，回头重装一遍，之前配好的 checkpoint 目录、Doris 端口全要重来。

2.1 四个组件的版本对应关系

这套链路里真正会互相制约的是 Flink 和 CDC 连接器，MySQL 与 Doris 更多是配合的角色。CDC 3.0.0 这一版把 DataStream、FlinkSQL、Pipeline 三种用法统一收进一个发行包，Pipeline 模式依赖 Flink 1.18 的运行时，所以 Flink 侧我一般固定在 1.18.0，Scala 二进制选 2.12，别用 2.13 的包去配_2.12后缀的 planner，类加载时找不到类又不报清楚的原因。

表里最后两行是很多人忽略的：MySQL 和 Doris 看起来“本来就在跑”，但 Flink CDC 对它们是有要求的。MySQL 必须是 ROW 格式的 binlog，Doris 必须至少有一个 BE 存活并且副本数不超过 BE 数量。

2.2 Flink 1.18.0 与 Doris 1.2.4 的最小安装改动

Flink 解压完不要急着提交作业，Pipeline 模式靠 checkpoint 记录 binlog 位点，没有 checkpoint 配置，任务重启就是从头发一遍，或者干脆不落位点。把下面几行追加到conf/flink-conf.yaml，分发到集群其它机器后再启动。

tar -zxf flink-1.18.0-bin-scala_2.12.tgz -C /opt/module/ cd /opt/module/flink-1.18.0 cat >> conf/flink-conf.yaml <<'EOF' # Pipeline 模式必须打开，间隔按业务量调，5 秒是起步值 execution.checkpointing.interval: 5000 # 状态后端与 checkpoint 落盘位置，放 HDFS 才能扛住 TaskManager 挂掉 state.backend: hashmap state.checkpoints.dir: hdfs://node101:8020/flink-cdc/checkpoints EOF bin/start-cluster.sh

execution.checkpointing.interval的单位是毫秒，配 5000 表示 5 秒一次；state.checkpoints.dir写在配置里比写在代码里更省事，因为 Pipeline 模式的作业根本走不到你的代码。这一段最容易忽略的是 HDFS 路径的可写权限，后面避坑章会专门讲。

Doris 侧按 FE、BE 顺序起，起来之后用 MySQL 客户端连查询端口确认后端节点注册状态。

cd /opt/module/doris/fe && bin/start_fe.sh --daemon cd /opt/module/doris/be && bin/start_be.sh --daemon # 9030 是 Doris 的查询端口，7030 是 FE 的 HTTP 端口，别混 mysql -uroot -p000000 -P9030 -hnode101 -e "show backends\G"

show backends里的Alive必须是 true，AvailCapacity不能是 0。这一步没过就往下走，后面 sink 建表一定会报副本数不满足，报错信息还很长，容易误判成权限问题。

2.3 MySQL 建库建表与 Binlog 参数

测试数据按教程里的结构建就够验证全链路：主库test放t1、t2、t3三张表，另一套test_route放同名表，用来验证路由。表结构简单到只有主键加两个字段，正是为了排除字段类型带来的干扰。

-- 主库，模拟业务库 CREATE DATABASE IF NOT EXISTS test; 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)); -- 第二套库，后面验证路由和多库同步 CREATE DATABASE IF NOT EXISTS test_route; 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)); -- 塞几条初始数据，全量快照阶段要读到 INSERT INTO test.t1 VALUES ('1001','zhangsan'),('1002','lisi'),('1003','wangwu'); INSERT INTO test.t3 VALUES ('1001','F'),('1002','F'),('1003','M');

建表时主键不能省，CDC 的 update、delete 事件要靠主键定位目标行，没有主键的表在 Doris 侧也建不出合适的模型。

真正让新手卡住的是 binlog。默认安装的 MySQL 8 是不开 binlog 的，CDC 读的是 binlog 而不是查询表，所以这一段必须配。改my.cnf加四行，然后重启 MySQL。

[mysqld] # 集群内唯一，多个 CDC 作业或主从复制时重复会直接握手失败 server-id = 1 # 开启 binlog，值作为文件前缀 log-bin = mysql-bin # Flink CDC 只认 ROW，STATEMENT/MIXED 都会报错 binlog_format = row # 只捕获这两个库，生产上按实际库名收窄 binlog-do-db = test binlog-do-db = test_route

binlog-do-db是按“当前默认库”过滤的，用错库名执行更新，事件不会写进 binlog，表现就是“我明明改了数据，Doris 没变”。这类问题没有日志告警，只能靠自己记住这条规则。

2.4 三行 SQL 确认 binlog 真的开了

重启 MySQL 之后别相信配置文件写对了，用 SQL 回读变量。

SHOW VARIABLES LIKE 'log_bin'; -- 期望 ON SHOW VARIABLES LIKE 'binlog_format'; -- 期望 ROW SHOW MASTER STATUS; -- File、Position 都要有值

SHOW MASTER STATUS返回空结果，说明 binlog 没真正生效，八成是my.cnf放错位置（有的发行版读mysqld.cnf）或者没重启服务。这三条命令建议写进自己的检查清单，后面所有疑难杂症都从这三条开始排查。

3. DataStream 方式接入：pom 依赖、MySqlSource 参数与 Checkpoint 断点续传

DataStream 写法是最能把原理看清的一种：全量快照怎么切、位点怎么存、重启怎么续，全部在代码里摊开。生产上如果只是同步一两张表到 Doris，用后面的 Pipeline 模式更省事；但只要涉及自定义加工，最后还是回到 DataStream。

3.1 pom 里必须出现的依赖与作用域

依赖写错的表现通常是ClassNotFoundException或NoSuchMethodError，两者都很费时间。原则是：Flink 自带的东西打provided，CDC 自己的东西和 JDBC 驱动打进去。

<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> <!-- Flink 运行时集群上已有，provided 避免和集群 jar 打架 --> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-streaming-java</artifactId> <version>${flink-version}</version> <scope>provided</scope> </dependency> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-clients</artifactId> <version>${flink-version}</version> <scope>provided</scope> </dependency> <!-- 后缀 _2.12 必须和 Flink 发行包的 scala 版本一致 --> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-table-planner_2.12</artifactId> <version>${flink-version}</version> <scope>provided</scope> </dependency> <!-- CDC source 依赖的桥接包，漏了会在 Source 构建时报找不到工厂 --> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-connector-base</artifactId> <version>${flink-version}</version> </dependency> <!-- CDC 本体，3.0.0 才带 Pipeline 相关类 --> <dependency> <groupId>com.ververica</groupId> <artifactId>flink-connector-mysql-cdc</artifactId> <version>3.0.0</version> </dependency> <!-- MySQL 8 必须用 8.x 驱动，5.1.x 会报认证插件不支持 --> <dependency> <groupId>mysql</groupId> <artifactId>mysql-connector-java</artifactId> <version>8.0.31</version> </dependency> </dependencies>

打provided这件事我踩过坑：早期图省事把flink-streaming-java一起打进 fat jar，结果集群上出现两份 Flink 类，作业能提交，运行到算子链初始化时抛IllegalStateException，日志里根本看不出是 jar 冲突。打包插件用maven-assembly-plugin的jar-with-dependencies，phase绑到package，出一条带依赖的 jar 就够。

3.2 MySqlSource 构建参数逐项拆解

Builder 里每一个参数都会影响行为，尤其是startupOptions，选错就是“为什么我的作业只读增量、历史数据一条没有”。

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; MySqlSource<String> mySqlSource = MySqlSource.<String>builder() .hostname("node103") // MySQL 地址，别写 localhost .port(3306) .databaseList("test") // 捕获的库，和 binlog-do-db 保持一致 .tableList("test.t1") // 库名.表名，支持 test\\.t.* 正则 .username("root") .password("000000") // 反序列化成 JSON，方便 print 和环境验证时肉眼看 .deserializer(new JsonDebeziumDeserializationSchema()) // initial：先全量快照，再接着读最新 binlog .startupOptions(StartupOptions.initial()) .build();

earliest是新手最容易误用的：如果 MySQL 的 binlog 已经被expire_logs_days清掉，作业从第一天就绪不了，报错信息非常隐晦。我一般先用initial把全量收上来，后续扩容重跑再换latest。

3.3 Checkpoint 与状态后端：断点续传的唯一可靠方式

CDC 读取的 binlog 位点是存在状态里的，没有 checkpoint 就没有断点续传，作业重启等于从头再来一遍，Doris 侧可能重复写入。所以开 checkpoint 不是可选项。

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); env.setParallelism(1); // 单表同步先单并行度，排查简单 // 3 秒一次 CK，EXACTLY_ONCE 保证位点不丢 env.enableCheckpointing(3000L, CheckpointingMode.EXACTLY_ONCE); env.getCheckpointConfig().setCheckpointTimeout(60 * 1000L); env.getCheckpointConfig().setMinPauseBetweenCheckpoints(3000L); // 任务被 cancel 时保留最后一次 CK，这是升级、改配置后的后悔药 env.getCheckpointConfig().enableExternalizedCheckpoints( CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION); // 失败率重启：一天内最多 3 次，每次间隔 1 分钟 env.setRestartStrategy(RestartStrategies.failureRateRestart( 3, Time.days(1L), Time.minutes(1L))); env.setStateBackend(new HashMapStateBackend()); env.getCheckpointConfig().setCheckpointStorage("hdfs://node101:8020/flink-cdc"); // 访问 HDFS 的身份，不设就用提交作业的 Linux 用户，多半没写权限 System.setProperty("HADOOP_USER_NAME", "deploy");

EXACTLY_ONCE在 CDC 场景下的含义是位点和数据一致，但 Doris 侧是否幂等还要看表模型；CheckpointingMode.AT_LEAST_ONCE更快，代价是重启后可能重复几条。RETAIN_ON_CANCELLATION这三个词值得牢记，没有它，你改完一个参数重启，就只能重新全量读一遍。

3.4 打包、提交与 Savepoint 恢复

本地 run 一遍确认能读到数据，再打成 jar 提交到集群。提交时指定 JobManager 地址和入口类，jar 后面跟的是程序参数，不是 Flink 参数。

# 打包，产出一个带全部依赖的 jar mvn clean package -DskipTests # 提交到 standalone 集群，8081 是 WebUI 端口 bin/flink run -m node101:8081 \ -c com.demo.cdc.FlinkCDCDataStreamTest \ ./flink-cdc-test.jar # 记下 JobId 之后，先做一次 savepoint 再动数据 bin/flink savepoint <JobId> hdfs://node101:8020/flink-cdc/save # 带 savepoint 重启，-s 后面是这个目录里带 savepoint-xxx 的那一层 bin/flink run -s hdfs://node101:8020/flink-cdc/save/savepoint-xxxxxx \ -c com.demo.cdc.FlinkCDCDataStreamTest ./flink-cdc-test.jar

-s的路径必须指到具体的savepoint-xxxxxx目录而不是它的父目录，指错了 Flink 会当成没有状态，直接从startupOptions重新开始。验证方式很直接：在 MySQL 里 update 一行，看 TaskManager 日志里打印的 JSON 是op:u而不是又一轮全量op:r。

4. FlinkSQL 与 Pipeline 两种写法：把 MySQL 变更落到 Doris 1.2.4

DataStream 灵活但代码量大，更常见的情况是“整库同步到 Doris”，这时候 FlinkSQL 适合做验证，CDC 3.0 的 Pipeline 模式适合做上线。

4.1 FlinkSQL 建一张 MySQL CDC 表

FlinkSQL 的好处是不用管序列化，建表带connector = 'mysql-cdc'就能查。注意主键要写NOT ENFORCED，Flink 不校验但下游需要它来推 update、delete。

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' = 'node103',\n" + " 'port' = '3306',\n" + " 'username' = 'root',\n" + " 'password' = '000000',\n" + " 'database-name' = 'test',\n" + " 'table-name' = 't1'\n" + ")"); // 直接查，结果会随 MySQL 变更持续更新 tableEnv.sqlQuery("SELECT * FROM t1").execute().print();

connector名字是mysql-cdc而不是mysql，写错会提示找不到 factory，报错信息会列出所有可用 connector，扫一眼就能确认 CDC 的 jar 有没有加载上。本地 IDE 跑直接运行 main 就行，打包到集群提交方式和 DataStream 完全一样。

4.2 Pipeline 模式：yaml 配置与依赖 jar 放置

CDC 3.0 新增的 Pipeline 模式不需要写代码，解压发行包，把两个 pipeline connector 的 jar 放进lib目录，剩下就是写 yaml。

tar -zxf flink-cdc-3.0.0-bin.tar.gz -C /opt/module/ cd /opt/module/flink-cdc-3.0.0 # 这两个 jar 不放进来，启动会报 connector 无法识别 cp /path/to/flink-cdc-pipeline-connector-mysql-3.0.0.jar lib/ cp /path/to/flink-cdc-pipeline-connector-doris-3.0.0.jar lib/ mkdir -p job && cd job

yaml 分三段：source描述 MySQL，sink描述 Doris，pipeline描述作业本身。

source: type: mysql hostname: node103 port: 3306 username: root password: "000000" tables: 'test\.t.*' # 正则，匹配 t1、t2、t3 server-id: 5400-5404 # 区间形式，多并行度时逐个分配 server-time-zone: UTC+8 # 和 MySQL 时区一致，否则时间字段差 8 小时 sink: type: doris fenodes: node101:7030 # FE 的 HTTP 端口，不是 9030 username: root password: "000000" # 源表加列时自动在 Doris 侧做轻量 schema change table.create.properties.light_schema_change: true # 副本数必须小于等于 BE 数量，单机测试写 1 table.create.properties.replication_num: 1 pipeline: name: Sync MySQL Database to Doris parallelism: 1

fenodes写 8030 还是 7030 取决于 FE 配置，很多发行版是 8030，教程里是 7030，连不上先用show frontends确认 HTTP 端口。server-id用区间是 3.0 的写法，MySQL 侧server-id = 1是服务端自己的编号，两者不冲突但也不能在同一实例上撞车。

4.3 启动顺序与路由变更

启动顺序有讲究：Doris 先就绪，Flink 集群次之，最后起 CDC 任务。目标库要先建出来，Pipeline 不会替你建库，只会建表。

# 1. Doris 侧建目标库 mysql -uroot -p000000 -P9030 -hnode101 -e "create database if not exists test" # 2. Flink 集群已启动且开了 checkpoint # 3. 提交同步任务 cd /opt/module/flink-cdc-3.0.0 bin/flink-cdc.sh job/mysql-to-doris.yaml

任务起来后去 Doris 里show tables，会看到t1、t2、t3自动建出来，字段和主键都跟着源表走。这时候在 MySQL 里 insert、update、delete 各来一条，刷新 Doris 就能看到变化。

路由变更解决的是“源库名和目标库名不一样”的场景，比如test_route要落到doris_test_route。在 yaml 里追加一段route，用正则把源表映射到目标表。

# 接在 source / sink 之后 route: - source-table: 'test_route\.t.*' sink-table: 'doris_test_route.t.*'

路由规则是正则替换，source-table里的点必须转义成\.，否则会被当成任意字符——这个坑非常隐蔽，表现是表全被同步到一个乱七八糟的名字下，或者干脆没同步。

5. 避坑排查：五个让任务三十秒就退出的现场

下面这些是我在装这套环境时真撞过的，按“现象 → 原因 → 解决”记，遇到同类问题可以直接对着查。

一、启动就报 binlog 相关错误现象：作业提交后立即失败，日志里有The MySQL server is not configured to use a row level binary log或类似提示。 原因：binlog_format不是 ROW，或者log-bin压根没开。有的环境是配置文件改了但服务没重启。 解决：回到 2.4 节那三条 SQL 回读变量，log_bin必须是 ON、binlog_format必须是 ROW。确认后在 MySQL 上执行FLUSH LOGS;再重试，别直接改代码怀疑连接器。

二、Checkpoint 一直失败，作业反复重启现象：WebUI 里 checkpoint 全是 FAILED，日志出现Permission denied或Filesystem closed。 原因：setCheckpointStorage指到 HDFS，但提交作业的 Linux 用户在 HDFS 上没有写权限，HADOOP_USER_NAME没设或者设成了不存在的用户。 解决：在 main 方法最前面加System.setProperty("HADOOP_USER_NAME", "deploy")，并确认hdfs dfs -ls /flink-cdc能看到这个用户建的目录。临时验证也可以先换成file:///tmp/flink-ck排除 HDFS 因素。

三、savepoint 恢复后数据重复或从头读现象：带-s重启，Doris 里出现重复行，或者日志显示又在做全量快照。 原因：-s路径指到了父目录，Flink 认为没有可用状态；或者startupOptions用的是initial，无状态启动必然重跑全量。 解决：路径要精确到savepoint-xxxxxx这一层。全量阶段本身不幂等，Doris 侧如果是 Duplicate 模型，重复行会实打实存在，这种表结构要改 Unique 模型。

四、Doris 建表失败，报副本数相关错误现象：Pipeline 启动后 sink 阶段报错，Doris 里表没建出来。 原因：replication_num写成了 3，但只有 1 个 BE 存活，副本数超过可用 BE 数量时建表直接拒绝。 解决：show backends看存活数量，测试环境统一写replication_num: 1。生产上先扩 BE 再改副本数，不要反过来。

五、时间字段差 8 小时现象：MySQL 里2024-01-01 10:00:00，Doris 里变成02:00:00或者反过来。 原因：server-time-zone没配，或者配成UTC而 MySQL 是东八区，Debezium 按 UTC 解析。 解决：yaml 里写server-time-zone: UTC+8，DataStream 方式则在 builder 上加.serverTimeZone("Asia/Shanghai")。这条不改，后面所有按时间做的对账都白做。

6. 装完怎么验证：全量、增量、断点续传三道关

装完能启动不代表同步对了。我通常分三道关验：全量快照的行数对不对、增量变更能不能实时追上、断点续传会不会丢或重。三道都过，这套环境才算真的装好。

第一关验全量。MySQL 侧和 Doris 侧各查一次行数，数字必须一致。注意 Doris 用 9030 端口连，别用 7030。

# 源端 mysql -uroot -p000000 -h127.0.0.1 -e "select count(*) from test.t1" # 目标端，端口和地址都不同 mysql -uroot -p000000 -P9030 -hnode101 -e "select count(*) from test.t1"

第二关验增量，覆盖 insert、update、delete 三种事件，加列这种 DDL 也顺手测一下，因为light_schema_change是否生效只有加列能看出来。

-- 在 MySQL 上依次执行，每执行一条就去 Doris 刷新一次 INSERT INTO test.t1 VALUES ('1004','zhaoliu'); UPDATE test.t1 SET name = 'lisi_new' WHERE id = '1002'; DELETE FROM test.t1 WHERE id = '1003'; ALTER TABLE test.t1 ADD COLUMN age INT;

预期是 Doris 侧先多一行、再改一行、再少一行，最后多出一个age列。如果 insert 生效但 delete 不生效，八成是 Doris 表模型建成了 Duplicate，主键没起作用。

第三关验断点续传，这一步最能暴露配置问题。做法是造一次 savepoint，停掉作业，期间在 MySQL 里改几条数据，再带 savepoint 重启，看新增的几条有没有被补上、老数据有没有被重复写。

还有一个容易被忽略的验证点：作业跑起来之后把 TaskManager 进程 kill 掉一个，看它能不能从 checkpoint 自动恢复。这一步不做，等真出故障的时候才发现重启策略配错了，损失的是线上数据。

从那以后我每次装这套链路，不管多急，都强制按“只同步一张表、插三条数据、删一条、重启一次”的顺序走一遍，跑通了再放大到整库。玄学问题几乎都出在跳过这一步。希望帮到你。

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