简介尚硅谷研究院编写的 Flink CDC 3.0 教程文档面向具备一定 Flink 基础、需要搭建实时数据同步链路的大数据研发人员。内容先从 CDC变更数据捕获概念切入对比基于查询与基于 Binlog 两类方案在执行模式、变化捕获能力、延迟与数据库压力上的差异再引出 flink-cdc-connectors 组件的用法。案例部分完整覆盖 MySQL 库表与测试数据准备、DataStream 与 Flink SQL 两种接入方式以及 MySQL 到 Doris 的 Streaming ETL 流程含环境搭建、配置文件编写、任务启动与结果验证并点明开启 Binlog、设置检查点等关键细节。资源包内为 1 个 docx 文件约 145KB即上述图文教程全文可按章节顺序阅读也可在实操时逐节对照。已有 1789 人学习下载。读者可借此独立完成从数据源接入、数据流处理到目标库写入的部署理解各项配置参数的实际作用并把同步方案迁移到自身业务场景中。1. Flink CDC 3.0 落地MySQL 一行变更到 Doris 报表之间缺的那一段业务库里改了一行订单状态下游报表要等第二天凌晨才变——这个落差几乎每个做数仓的人都撞过。手上这份 Flink CDC 3.0 的资料给的就是把这段路补上的完整链路先用两章把变更数据捕获的两条技术路线讲清楚再用 DataStream 和 FlinkSQL 各跑一遍 MySQL 到控制台的最小闭环最后用 3.0 才有的 pipeline 配置文件把 MySQL 整库直接同步进 Doris。它适合已经能起 Flink 集群、手上有 MySQL 业务库、正在被 T1 报表拖着走的工程师也适合刚接触 CDC、想照着命令一行行抄一遍的新手。材料是命令行、配置项、代码混排的形态没有多余的架构图我照着跑了两轮把其中容易翻车的地方单独记了下来。2. 选型先立住基于查询与基于 Binlog 的 CDC 差在哪三条线上2.1 两种捕获模式的执行模型对比CDC 的核心动作只有一句话盯住数据库里数据行和表的增删改把每一次变更按发生顺序记下来再送给下游订阅消费。听着简单落到实现上分成两条完全不同的路一条是定时拿 SELECT 去问数据库「你变了没有」另一条是让采集程序伪装成一台从库直接读 MySQL 的 Binlog 事件流。这两条路的差别不是快慢而是能不能看见完整的变更历史。基于查询的方案靠时间戳字段或者自增 ID 捞增量两次查询之间发生的中间状态会被压平DELETE 语句更是彻底看不见——行没了你只知道少了不知道少的是哪一行、什么时候少的。维度基于查询Sqoop、DataX 一类基于 BinlogCanal、Maxwell、Debezium、Flink CDC执行模式Batch一批一批拉Streaming事件一条条来能否捕获全部变更否删除和中间态会丢是Insert/Update/Delete 全有变化延迟高取决于调度周期低通常在秒级对源库的压力有每次都要走查询和锁极小读的是日志不是表断点续传靠字段水位容易错位靠 binlog 位点精确到 event选型的判断标准其实很朴素目标端的表如果要求跟源库严格一致包括删除那就只有 Binlog 这条路如果只是每天跑一次全量覆盖、下游能容忍延迟查询式方案成本更低没必要上流式。资料里把这条对比放在第 1 章不是凑篇幅是因为后面所有的配置复杂度——开 Binlog、配 server-id、开检查点——根子都在这个选择上。2.2 为什么 MySQL 侧必须开 row 格式的 BinlogFlink CDC 的 MySQL 连接器底层用的是 Debezium 的引擎Debezium 解析的是行级变更事件所以binlog_format必须是row。statement格式只记 SQL 原文遇到NOW()、UUID()这类函数时从库重放出来的值跟主库不一致mixed格式在语句和行之间来回切解析端要处理的边界情况陡增。这不是性能取舍是能不能解析出来的问题。binlog-do-db用来限定只对指定库记日志。它的语义是按「当前默认库」匹配不是按「被操作的表所属库」匹配所以跨库语句容易踩空。我一般会把它当成白名单来用把要同步的库逐条列进去同时把 binlog 的保留时长调大——binlog_expire_logs_seconds默认值在不少环境里只有几天任务停一个周末回来位点就过期了这是最典型的「重启即全量」事故来源。server-id是另一处必须提前想清楚的地方。连接器读 binlog 时是以从库身份连上去的MySQL 主库要求每个从库的server-id唯一。单并行度时连接器会自动生成一个一旦你把 Flink 任务的并行度调到 2 以上多个 reader 各自要一个 id就得手动给一段范围。2.3 把 MySQL 环境按资料里的参数配好资料里准备了两套库test和test_route各建三张表这个设计的用意是后面验证路由——两张同名表数据刻意错开同步完一眼就能看出有没有串库。# 编辑 MySQL 配置文件开启 Binlog sudo vim /etc/my.cnf # 在 [mysqld] 段落下追加 # server-id集群内唯一供 CDC 连接器以从库身份注册 server-id 1 # log-bin开启 binlog 并指定文件前缀 log-bin mysql-bin # binlog_format必须为 rowDebezium 只能解析行级事件 binlog_format row # binlog-do-db白名单逐条列出需要捕获的库 binlog-do-db test binlog-do-db test_route # 建议追加binlog 保留 7 天避免任务停机后位点丢失 binlog_expire_logs_seconds 604800改完配置重启 MySQL然后用两条命令自检确认参数真的生效而不是被 my.cnf 里别处的重复项覆盖-- 逐项确认关键参数 SHOW VARIABLES LIKE log_bin; -- 期望 ON SHOW VARIABLES LIKE binlog_format; -- 期望 ROW SHOW VARIABLES LIKE server_id; -- 期望与配置一致 SHOW VARIABLES LIKE binlog_expire_logs_seconds;建库建表这一步资料里给的结构很简单两张同步表加一张只有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)); -- 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)); -- 灌入初始数据test 与 test_route 的 t2 主键区间不同 INSERT INTO test.t1 VALUES (1001,zhangsan),(1002,lisi),(1003,wangwu); INSERT INTO test.t2 VALUES (1001,zhangsan),(1002,lisi),(1003,wangwu); INSERT INTO test.t3 VALUES (1001,F),(1002,F),(1003,M); INSERT INTO test_route.t1 VALUES (1001,zhangsan),(1002,lisi),(1003,wangwu); INSERT INTO test_route.t2 VALUES (1004,zhangsan),(1005,lisi),(1006,wangwu); INSERT INTO test_route.t3 VALUES (1001,F),(1002,F),(1003,M);这里有个细节值得留意id用VARCHAR而不是自增整型。好处是造数据时能精确控制坏处是位移和分片时无法按主键区间切。实际项目里主键几乎都是自增 BIGINTFlink CDC 会按主键区间做快照切分这种情况下的快照并发能力比字符串主键好得多。3. DataStream 方式跑通依赖、Source 与检查点三件套3.1 依赖矩阵Flink 1.18.0 与 connector 3.0.0 怎么对齐依赖表看着长真正决定能不能跑起来的只有几项Flink 的主版本、Scala 后缀、连接器版本。资料里用的是 Flink 1.18.0 配flink-connector-mysql-cdc3.0.0表规划器用的是flink-table-planner_2.12这个_2.12就是 Scala 版本后缀跟你集群发行包必须一致写成_2.11会在类加载阶段直接报找不到表工厂。properties maven.compiler.source8/maven.compiler.source maven.compiler.target8/maven.compiler.target flink-version1.18.0/flink-version project.build.sourceEncodingUTF-8/project.build.sourceEncoding /properties dependencies !-- 编译期需要提交到集群时由集群提供故用 provided -- dependency groupIdorg.apache.flink/groupId artifactIdflink-streaming-java/artifactId version${flink-version}/version scopeprovided/scope /dependency dependency groupIdorg.apache.flink/groupId artifactIdflink-clients/artifactId version${flink-version}/version scopeprovided/scope /dependency !-- FlinkSQL 路线必需注意 Scala 后缀与发行包一致 -- dependency groupIdorg.apache.flink/groupId artifactIdflink-table-planner_2.12/artifactId version${flink-version}/version /dependency dependency groupIdorg.apache.flink/groupId artifactIdflink-table-api-java-bridge/artifactId version${flink-version}/version /dependency !-- CDC SourcegroupId 与 Flink 官方不同是独立发布的 -- 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 /dependencies build plugins !-- 用 assembly 打成带依赖的胖包避免集群缺类 -- plugin groupIdorg.apache.maven.plugins/groupId artifactIdmaven-assembly-plugin/artifactId version3.0.0/version configuration descriptorRefs descriptorRefjar-with-dependencies/descriptorRef /descriptorRefs /configuration executions execution idmake-assembly/id phasepackage/phase goalsgoalsingle/goal/goals /execution /executions /plugin /plugins /buildscope这一栏最容易出问题。如果把flink-streaming-java打成 provided本地 IDE 跑不起来是正常的因为 IDE 不会自动补集群的类路径反过来如果不用 provided 而把 Flink 全家桶打进胖包提交到集群后会出现同一份 Flink 类被加载两次典型症状是IllegalStateException: Trying to access closed classloader这类莫名其妙的报错。我的习惯是Flink 核心和 table 规划器一律 providedCDC 连接器和 JDBC 驱动打进去然后本地测试时用 IDEA 的 provided 依赖勾选项临时打开。3.2 MySqlSource 的 builder 参数与五种启动模式MySqlSource是整个 DataStream 路线的入口它的 builder 上有十来个参数其中真正决定行为的是启动模式。资料里把五种模式的英文注释原封不动贴了出来这段注释比正文更值得读——它决定了任务第一次启动时到底从哪儿开始读。启动模式首次启动行为适用场景initial先做一次全量快照再接着读最新 binlog默认值绝大多数首次接入场景earliest不做快照从 binlog 最早位置读历史 binlog 完整且不想重扫全表latest不做快照只读启动之后的变更只关心增量历史数据另有来源specificOffset不做快照从指定 binlog 文件与位点读故障后精确回补某一段timestamp不做快照从指定时间戳开始找位点补数时有明确时间边界新手最容易在这里翻车以为latest会先同步一遍存量数据结果任务跑起来下游一张空表还以为是连接没通。latest的语义就是「装作我刚来之前的事我不管」历史数据必须靠initial或者另外跑一次全量。MySqlSourceString mySqlSource MySqlSource.Stringbuilder() .hostname(node01) // MySQL 地址 .port(3306) .databaseList(test) // 捕获的库可传多个 .tableList(test.t1) // 捕获的表格式 db.table支持正则 .username(root) .password(******) // 反序列化器把 Debezium 的 SourceRecord 转成 JSON 字符串 .deserializer(new JsonDebeziumDeserializationSchema()) // 首次启动先全量快照再持续读 binlog .startupOptions(StartupOptions.initial()) .build();databaseList和tableList要配合着写。只写databaseList(test)会捕获整库所有表tableList里如果再写test.t1就只保留 t1——两者是交集关系。tableList用的是正则匹配写test\\.t.*就能把 t1、t2、t3 一起捕获。生产上我一般显式列表宁可多写几行也不要用过宽的正则把无关表卷进来特别是带大字段的日志表。还有一个不在 builder 表面但影响很大的参数是serverId。并行度大于 1 时每个快照分片都要独立注册一个从库身份必须给它一段范围。不显式指定时连接器会在默认区间里随机取值多任务并行跑在同一个 MySQL 实例上就有概率撞车报错信息是「A slave with the same server_uuid/server_id as this slave has connected to the master」看着像玄学其实就是 id 重了。3.3 Checkpoint 配置与断点续传的完整代码Flink CDC 把读到的 binlog 位点信息存在状态里而状态只有在检查点成功时才会持久化。不开检查点任务一重启就是从零开始——这不是配置项的问题是整条链路能不能续上的问题。StreamExecutionEnvironment env StreamExecutionEnvironment.getExecutionEnvironment(); env.setParallelism(1); // 每 3 秒做一次检查点语义为精确一次 env.enableCheckpointing(3000L, CheckpointingMode.EXACTLY_ONCE); CheckpointConfig ckConfig env.getCheckpointConfig(); ckConfig.setCheckpointTimeout(60 * 1000L); // 单次 CK 超时 1 分钟 ckConfig.setMinPauseBetweenCheckpoints(3000L); // 两次 CK 最小间隔防止拖垮 ckConfig.setMaxConcurrentCheckpoints(1); // 同一时刻只允许一个 CK // 任务被 cancel 时保留最后一次 CK这是后续 -s 恢复的前提 ckConfig.enableExternalizedCheckpoints( CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION); ckConfig.setCheckpointStorage(hdfs://node01:8020/flinkCDC); // 失败率重启1 天内最多 3 次每次间隔 1 分钟 env.setRestartStrategy(RestartStrategies.failureRateRestart( 3, Time.days(1L), Time.minutes(1L))); // 状态后端与 HDFS 访问身份 env.setStateBackend(new HashMapStateBackend()); System.setProperty(HADOOP_USER_NAME, operator);几个参数值得单独说。setMinPauseBetweenCheckpoints设成跟间隔一样大是为了防止大状态下检查点写不完就叠下一个堆积起来会把 TaskManager 的 IO 打满。RETAIN_ON_CANCELLATION是断点续传的后悔药不开启的话任务一 cancelHDFS 上的检查点目录会被清掉想恢复也没得恢复。HADOOP_USER_NAME这一行看着像补丁但在没配 Kerberos 的测试集群里不设它经常报权限不足写 HDFS 直接失败。检查点间隔也别无脑往小里调。3 秒一次对演示环境没问题但同步链路本身有延迟检查点太密会让位点推进速度跟不上数据量反而拖慢整体吞吐。我一般按数据量给 10 秒到 1 分钟宁可恢复时多补一点数据也不要让检查点成为瓶颈。3.4 打包提交与 Savepoint 恢复验证代码写完只是第一步验证「断点续传真的有效」才是这条链路能不能上生产的分水岭。完整的验证动作是这样一串# 1. 打包产物是两个 jar带依赖的那个才是提交用的 mvn clean package -DskipTests # 2. 启动 HDFS检查点目录在 HDFS 上 start-dfs.sh # 3. 启动 Flink 集群 bin/start-cluster.sh # 4. 提交任务 bin/flink run -m node01:8081 \ -c com.example.cdc.FlinkCDCDataStreamTest \ ./flink-cdc-test-jar-with-dependencies.jar # 5. 观察 TaskManager 日志此时应打印出 t1 的全量三行 tail -f log/flink-*-taskexecutor-*.out # 6. 给运行中的任务打一个 Savepoint bin/flink savepoint JobId hdfs://node01:8020/flinkCDC/save # 7. 在 WebUI 里 cancel 任务然后去 MySQL 里改几条数据 # UPDATE test.t1 SET namezhaoliu WHERE id1001; # DELETE FROM test.t1 WHERE id1003; # 8. 从 Savepoint 恢复 bin/flink run -s hdfs://node01:8020/flinkCDC/save/savepoint-xxxx \ -c com.example.cdc.FlinkCDCDataStreamTest \ ./flink-cdc-test-jar-with-dependencies.jar判断恢复是否真的成功的标志不是任务起来了而是第 8 步之后的日志里出现了第 7 步改动的两条记录一条 UPDATE、一条 DELETE而且没有重新打印全量三行。如果日志里又是从头开始的三行 INSERT说明位点没存进去八成是检查点没开或者路径写错了。这一步我每次接新库都会强制走一遍因为它是唯一能在出故障之前证明「这套配置能续上」的办法。4. FlinkSQL 与 pipeline YAML两条少写代码的落地路径4.1 一条 DDL 把 MySQL 表变成动态表同一个 MySQL 源换成 SQL 方式后代码量掉了一个数量级。核心是一条带connector mysql-cdc的建表语句声明完之后它就是一张会自己更新的动态表后面接什么算子都行。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 // 主键不强制校验但 CDC 需要它来做更新合并 name STRING\n ) WITH (\n connector mysql-cdc,\n hostname node01,\n port 3306,\n username root,\n password ******,\n database-name test,\n table-name t1,\n server-id 5401-5404,\n // 并行度 1 时必须显式给范围 server-time-zone Asia/Shanghai\n // 与 MySQL 时区对齐否则时间字段差 8 小时 ) ); // 简单过滤只有 name 非空的行往下走 tableEnv.executeSql(SELECT id, name FROM t1 WHERE name IS NOT NULL) .print();PRIMARY KEY NOT ENFORCED这半句必须写。NOT ENFORCED是 Flink 的语法要求表示主键只用于优化和去重语义不做运行期约束校验而PRIMARY KEY本身对 CDC 至关重要——连接器靠它识别一行数据的唯一标识从而把 INSERT 和 UPDATE 合并成 upsert 语义。少写主键下游收到的会是带I、-U、U标记的 changelog 流而不是最终状态。server-time-zone是我吃过亏的地方。MySQL 里存的是DATETIME不带时区信息连接器用 JVM 默认时区去解释它容器里的 JVM 时区常常是 UTC结果同步到 Doris 之后所有时间字段整体偏移 8 小时。这个错误不会报异常只会让你在报表上看到「凌晨三点的订单」排查起来很费时间。4.2 SQL 与 DataStream 两条路的取舍两种写法都能跑通选哪个看的是维护成本而不是能力上限。维度DataStream APIFlinkSQL代码量多样板代码占大头少一条 DDL 加一条查询细粒度控制强可自定义反序列化、水印、状态弱受连接器选项限制多表关联需要手写 join 和状态管理直接写 SQL join由规划器优化字段变更适应性改 schema 要改代码、重打包多数情况改 DDL 即可上线节奏慢走完整的构建发布快适合频繁调整口径经验上的分界线是下游只需要一张表或简单映射SQL 够用一旦涉及自定义解析、复杂事件时间处理、多流合并老老实实回到 DataStream。还有一个隐性成本——SQL 路线的报错信息经常被规划器包一层栈里看不到原始异常排查链路比 DataStream 长这一点在赶工期的时候很致命。4.3 pipeline 模式MySQL 到 Doris 的三段式 YAML3.0 版本最实用的变化是 pipeline 模式不写一行 Java 代码用一份 YAML 描述「从哪读、往哪写、怎么跑」命令行提交即可。整库同步这种重复劳动的活用它比手写 DataStream 省太多。先把两个连接器插件放进 Flink CDC 安装目录的lib下版本号必须与主程序一致tar -zxvf flink-cdc-3.0.0-bin.tar.gz -C /opt/module/ cd /opt/module/flink-cdc-3.0.0 # 把 MySQL 源连接器和 Doris 目标连接器都放进 lib cp flink-cdc-pipeline-connector-mysql-3.0.0.jar lib/ cp flink-cdc-pipeline-connector-doris-3.0.0.jar lib/ ls lib/配置文件分三段source、sink、pipeline各管一摊source: type: mysql hostname: node01 port: 3306 username: root password: ****** tables: test.\.* # 正则test 库下所有表 server-id: 5400-5404 # 范围必须够并行度用 server-time-zone: UTC8 sink: type: doris fenodes: node01:7030 # Doris FE 的 HTTP 端口不是 MySQL 协议端口 username: root password: ****** # 源端加列时Doris 侧做轻量 schema change避免整表重建 table.create.properties.light_schema_change: true # 单 BE 环境副本数只能是 1多 BE 建议 3 table.create.properties.replication_num: 1 pipeline: name: Sync MySQL Database to Doris parallelism: 1tables用的是正则test.\.*里那个反斜杠是转义点号表示「库名 test表名任意」。正则写宽了会把整库卷进来写窄了漏表建议先在 MySQL 里SHOW TABLES对一遍。server-id给的是一个区间而不是单值因为 pipeline 内部会按并行度切分快照每个分片都要独立的从库身份区间长度要大于等于parallelism。sink段里fenodes最容易填错——它要的是 Doris FE 的 HTTP 查询端口通常是 8030资料里写的是 7030实际部署时按自己环境fe.conf里的http_port填。填成 BE 地址或者 MySQL 协议端口症状是「连接超时」而不是明确的配置错误提示。4.4 路由变更把多库多表映射到目标端默认行为是同名入库同表源端test.t1就落到目标端test.t1。当需要把两个源库合并到目标端一个库或者给表名加前后缀时就要用 route 段route: # 源端 test_route 库的 t1落到目标端 doris_test_route 库的 t1 - source-table: test_route\.t1 sink-table: doris_test_route.t1 # 源端其余表按原样落到同名库 - source-table: test\.(.*) sink-table: test.$1两侧都接受正则sink-table里可以用$1引用捕获组。这里的关键是先跑通无路由的默认同步再加上 route 段重跑一次对比两次结果才能确认路由规则真的按预期生成了目标表而不是把所有表都写进了同一个地方。资料里那两张主键区间刻意错开的t2就是给这一步准备的校验样本如果test.t2和test_route.t2落进了同一张表主键区间一合并立刻就能看出来。4.5 启动顺序与首轮同步验证pipeline 模式依赖 Flink 的检查点机制所以启动前必须先让集群具备检查点条件# 1. Flink 侧开启检查点写进配置文件后分发到所有节点 vim conf/flink-conf.yaml # 追加execution.checkpointing.interval: 5000 bin/start-cluster.sh # 2. Doris 侧启动 FE 与 BE /opt/module/doris/fe/bin/start_fe.sh --daemon /opt/module/doris/be/bin/start_be.sh --daemon # 3. 目标库需要先建好pipeline 只自动建表不自动建库 mysql -uroot -p****** -P9030 -hnode01 \ -e CREATE DATABASE IF NOT EXISTS test; # 4. 提交同步任务 bin/flink-cdc.sh job/mysql-to-doris.yaml # 5. 到 Doris 侧验证 mysql -uroot -p****** -P9030 -hnode01 \ -e SELECT * FROM test.t1 ORDER BY id;第 3 步的顺序不能省。pipeline 会自动建表但库必须提前存在否则任务启动阶段就报「database not found」而且这个错在 Flink WebUI 上只显示成一句笼统的初始化失败。任务起来之后去 MySQL 里对几张表做增、改、加列操作再刷新 Doris 侧三个动作都要验新数据出现、修改生效、新列自动补上。5. 避坑清单五个把同步任务跑挂的现场记录5.1 任务重启后从零开始全量数据又灌了一遍现象任务因为集群重启中断恢复之后日志里重新打印了全量快照下游出现大量重复写入。原因检查点没开或者开了但没开启外部化保留cancel 时检查点目录被一起清掉。Flink CDC 的位点存在状态里状态没了就等价于第一次启动。解决enableCheckpointing必须开enableExternalizedCheckpoints(RETAIN_ON_CANCELLATION)必须开检查点存储放到 HDFS 这类共享存储上而不是本地磁盘。恢复时用flink run -s savepoint路径而不是直接重新提交。5.2 并行度调到 2 以上就报 server_id 冲突现象单并行度跑得好好的把parallelism改成 3任务启动阶段抛异常提示从库 id 已被占用。原因连接器每个快照分片都要以一个独立从库身份连 MySQL不显式指定server-id时会在默认区间随机取取值碰撞就报错同时跑多个 CDC 任务时碰撞概率更高。解决显式给server-id一个范围长度不小于并行度且不同任务之间相互避开。我在配置文件里按任务编号分段比如同步任务 A 用 5400-5410、任务 B 用 5420-5430从源头上不重叠。5.3 目标端时间字段整体差 8 小时现象数据行数对得上主键也对得上但所有时间列在 Doris 里比 MySQL 里早或晚了 8 小时。原因MySQL 的DATETIME不带时区连接器用 JVM 默认时区解释它。容器镜像里 JVM 默认时区常是 UTC而业务数据是按东八区写进去的。解决源端配置加server-time-zonepipeline 用server-time-zone: UTC8SQL 路线用server-time-zone Asia/Shanghai。改完之后必须重跑一次全量历史数据不会自动纠正。5.4 Doris 连接器类型识别不到报找不到 sink现象bin/flink-cdc.sh提交后立刻失败提示无法创建类型为 doris 的 sink或者ClassNotFoundException。原因flink-cdc-pipeline-connector-doris-3.0.0.jar没放进lib目录或者放进去的版本与主程序不一致。这个加载是插件式的缺包不会在启动时给友好提示。解决lib下同时确认 mysql 和 doris 两个 pipeline 连接器 jar 存在且版本号一致然后重启 Flink 集群让 classloader 重新扫描。加完包不重启是另一个常见疏漏。5.5 源端加了一列同步任务开始报 schema 不匹配现象MySQL 上执行ALTER TABLE ADD COLUMN之后同步任务报错或者新列数据丢失。原因目标端的表结构是建表时按当时的源端结构生成的源端加列后目标端没有对应字段写入时字段数对不上。解决pipeline 的 sink 段开启table.create.properties.light_schema_change: true让 Doris 走轻量 schema change同时确认replication_num与 BE 数量匹配副本数配大了建表会一直卡在等待副本。DDL 变更之后建议主动观察一轮日志确认 schema change 真的执行了而不是静默失败。6. 收尾这一公里Doris 端对账与三个值得调的参数数据同步跑起来只是及格线能证明数据对得上才算交付。我的习惯是在两端跑同一条聚合 SQL比对行数和内容摘要而不是肉眼看前几行。-- MySQL 侧执行 SELECT COUNT(*) AS cnt, SUM(CRC32(CONCAT_WS(|, id, name))) AS digest FROM test.t1; -- Doris 侧执行同一口径 SELECT COUNT(*) AS cnt, SUM(CRC32(CONCAT_WS(|, id, name))) AS digest FROM test.t1;两边的cnt和digest都要一致。CONCAT_WS里字段顺序必须两边完全一样NULL值的处理也要提前统一否则摘要一定对不上但数据其实没错。这个对账要跑三个时间点首轮全量完成后、做一轮增删改之后、任务重启恢复之后。第三个时间点最能暴露位点问题因为只有它会走到「从检查点续读」这条分支。参数调优上我只动了三个地方。第一是split-size快照阶段按主键区间切分的块大小默认值偏保守大表首轮同步会切成很多小片、来回切库调到几十 MB 一档能明显缩短首轮时间但不能超过单分片内存承受范围。第二是检查点间隔从 5 秒放宽到 30 秒后TaskManager 的 IO 等待明显下降恢复时需要补的数据量也在可接受范围。第三是 sink 端写入批次Doris 是按批提交的批次太小会放大导入次数太大则单批失败重试代价高通常按每秒数据量乘以 3 到 5 秒来估。有一次我图省事把检查点间隔调到 3 秒、split-size又调得很大首轮快照期间检查点一直超时任务反复重启最后是靠把间隔改回 30 秒才稳住。从那以后我每接一个新库都先把server-id范围、binlog 保留时长、检查点存储路径这三条写进上线检查单再动手改 YAML。希望帮到你。本文还有配套的精品资源点击获取