简介面向具备一定 Flink 基础的大数据研发人员这份尚硅谷研究院编写的 Flink CDC 3.0 文档围绕 MySQL 变更数据实时捕获与同步至 Doris 展开先厘清 CDC 概念及基于查询、基于 Binlog 两类方案的差异再引入 flink-cdc-connectors 组件的使用方式帮助读者建立从原理到落地的完整认知。包内为 1 个 docx 文件压缩包约 145KB篇幅集中、结构完整按 CDC 简介、案例实操等章节组织便于顺序阅读或按需查阅。文档给出数据准备、DataStream 与 Flink SQL 两种实现路径并完整覆盖 MySQL 到 Doris 的 Streaming ETL环境搭建、配置文件编写、任务启动与测试均有对应说明同时强调开启 Binlog、设置检查点等关键配置便于排查同步链路中的一致性与稳定性问题。目前已有 1789 人学习适合希望独立完成实时同步项目部署、积累参数调整经验的工程师参考。1. 从一次凌晨两点的对账说起Flink CDC 3.0 到底补上了哪块短板业务库里的订单表一天几百万行变动报表库却只能等凌晨跑完批处理才更新第二天上班第一件事就是对账差一行就得翻一上午日志。这套以 Flink CDC 3.0 为主线的实操资料讲的正是把 MySQL 的变更通过 Binlog 实时捕获、再同步进 Doris 的完整链路从 CDC 概念、Binlog 开启、DataStream 与 Flink SQL 两种写法一直到 3.0 才有的 Pipeline YAML 整库同步与路由。它适合已经能跑起 Flink 任务、又被 T1 报表卡住的工程师新手可以照着命令一行行敲熟手能直接看参数边界和失败现场。资料本身是一份按章节推进的实操文档配置和代码占比很高不是概念科普。2. 选型先落地查询型 CDC 和 Binlog 型 CDC 差的不只是延迟动手之前把路线定错后面所有配置都是白费。CDC 这个词被用得很泛Sqoop 定时拉一遍也叫捕获变更Canal 读 Binlog 也叫捕获变更但两者的代价模型完全不同。这一章先把两条路线的账算清楚再把 MySQL 侧三个绕不开的参数定死最后造一批可验证的数据让后面每个实验都有据可查。2.1 一张表把两条路线的代价分开维度基于查询的 CDC基于 Binlog 的 CDC代表产品Sqoop、DataXCanal、Maxwell、Debezium、Flink CDC执行模式BatchStreaming能否捕获删除否只能靠全量比对推断是delete 事件带完整主键变化延迟高取决于调度周期低通常在秒级对源库压力每次扫描都压库大表容易打满 IO只读 Binlog 流压力集中在网络与解析是否要求开启 Binlog否是且必须是 ROW 格式看完这张表基本就能下判断只要业务要求「删了也要同步掉」「改了要在一分钟内出现在报表里」查询型路线就直接出局。真正需要权衡的是运维成本——Binlog 路线要求 DBA 配合改配置、开权限、规划 server-id这些动作在规范化程度不高的团队里往往比写代码更耗时间。Flink CDC 的位置比较特殊它把 Debezium 的解析能力包进了 Flink Source全量和增量在同一个作业里完成启动时先做快照读读完自动切到 Binlog 流不需要额外维护一张位点表也不需要写两套代码。这一点在 3.0 里被进一步放大整库同步只需要一份 YAML。2.2 binlog_format、server-id、账号权限三个必须在动手前定死的参数MySQL 侧的配置是所有失败案例的第一现场。下面这段是常见的最小配置改完必须重启 MySQL 才生效热加载不行。# /etc/my.cnf [mysqld] 段内 server-id 1 # 打开 binlog值会作为文件前缀 log-bin mysql-bin # 必须是 ROWSTATEMENT/MIXED 拿不到变更前后的完整行镜像 binlog_format ROW # 只记录白名单内的库漏掉的库后面怎么配都同步不过来 binlog-do-db test binlog-do-db test_route # 行镜像保留完整字段遇到 UPDATE 未改动的列也能拿到旧值 binlog_row_image FULLbinlog_format必须设成ROW这是硬性前提。设成STATEMENT时 Binlog 里存的是 SQL 语句Flink CDC 拿不到变更前后的行数据反序列化阶段就会报字段缺失设成MIXED更麻烦平时跑得好好的某条语句被 MySQL 判定为不确定函数后切成 ROW任务时好时坏属于典型的玄学故障。binlog-do-db是白名单机制不是黑名单。它的判定依据是「当前默认库」所以跨库更新语句可能出现记录行为与预期不一致的情况。我的习惯是同步哪些库就把哪些库全部列进去后续新增库一定要记得回来补配置并重启——这个坑后面还会提。server-id在单机单任务时随便写但只要同一个 MySQL 实例上挂两个以上的同步任务就必须规划成不重叠的区间。Flink CDC 3.0 允许写成5400-5404这样的范围意思是给最多 5 个并行 reader 各分配一个 id两个任务都用5400-5404MySQL 会直接拒绝后连上来的那个。账号权限同样要给够只给 SELECT 是不够的-- 同步账号全量阶段要读表增量阶段要读 binlog CREATE USER cdc_user% IDENTIFIED BY Cdc123456; GRANT SELECT, RELOAD, SHOW DATABASES, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO cdc_user%; FLUSH PRIVILEGES;REPLICATION SLAVE和REPLICATION CLIENT是读 Binlog 的门票缺了会在增量启动时抛权限异常RELOAD用于执行FLUSH TABLES之类的操作全量快照阶段会用到SHOW DATABASES让连接器能枚举库表结构。授权语句里的库名用*.*更省事生产上可以按需收窄到具体库。2.3 建库建表造数让后面每个实验都有数据可查配置改完不代表链路通得先有能验证的数据。下面这套结构来自资料里的案例两个库、三张表、主键都是字符串故意留了一张字段不一致的表t3 只有 id 和 sex用来观察 schema 差异时的表现。-- 源库与目标库分开建方便后面演示路由同步 CREATE DATABASE IF NOT EXISTS test; CREATE DATABASE IF NOT EXISTS 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)); 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); -- 第二个库的表结构与第一个库一致但数据主键错开便于区分来源 USE test_route; CREATE TABLE t1 (id VARCHAR(255) PRIMARY KEY, name VARCHAR(255)); INSERT INTO t1 VALUES (2001,zhaoliu),(2002,qianqi);建完之后别急着写 Flink 代码先确认 Binlog 真的在写# 1. 确认格式与是否开启 mysql -ucdc_user -p -e SHOW VARIABLES LIKE binlog_format; SHOW VARIABLES LIKE log_bin; # 2. 看当前写到哪个文件、哪个位点 mysql -ucdc_user -p -e SHOW MASTER STATUS\G # 3. 制造一条变更后再看位点是否前移位点不动说明白名单没覆盖这个库 mysql -ucdc_user -p -e INSERT INTO test.t1 VALUES (1004,sunba); SHOW MASTER STATUS\G三步都通过才说明 MySQL 侧是干净的。很多「任务启动了但没数据」的问题其实在这一步就已经能暴露出来只是现场直接跳到了 Flink 日志里翻白白多花两小时。3. DataStream 与 Flink SQL同一份 MySQL 数据的两条入口Flink CDC 3.0 提供了不止一种接入方式DataStream API 控制力最强能精确设置 Checkpoint、重启策略和状态后端Flink SQL 写法最短适合快速验证和把它当成一张维表/源表嵌进更复杂的查询里。两条路读的是同一个 Binlog区别在于你能干预多少细节。3.1 依赖与打包版本错一个就是 ClassNotFound依赖这块翻车率极高核心原因是 Flink 主版本、CDC 连接器版本、MySQL 驱动版本三者必须互相对齐。下面这份 POM 是资料里验证过能跑的组合Flink 1.18 配 CDC 3.0.0。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 !-- Flink 运行时与 Table 相关依赖provided 还是 compile 取决于提交方式 -- dependency groupIdorg.apache.flink/groupId artifactIdflink-streaming-java/artifactId version${flink.version}/version /dependency dependency groupIdorg.apache.flink/groupId artifactIdflink-clients/artifactId version${flink.version}/version /dependency dependency groupIdorg.apache.flink/groupId artifactIdflink-table-planner_2.12/artifactId version${flink.version}/version /dependency dependency groupIdorg.apache.flink/groupId artifactIdflink-table-runtime/artifactId version${flink.version}/version /dependency !-- CDC Source 连接器坐标仍在 com.ververica 下 -- dependency groupIdcom.ververica/groupId artifactIdflink-connector-mysql-cdc/artifactId version3.0.0/version /dependency !-- 驱动版本要和 MySQL 服务端匹配8.0 服务端别用 5.1 驱动 -- dependency groupIdmysql/groupId artifactIdmysql-connector-java/artifactId version8.0.31/version /dependency /dependencies打包用maven-assembly-plugin的jar-with-dependencies描述符把所有依赖塞进一个胖包提交时只传一个文件。这里有个容易混淆的点3.0 里的 Pipeline 连接器后面第 4 章要用坐标在org.apache.flink下和这里的com.ververica:flink-connector-mysql-cdc不是同一个包两者混着引会出现类冲突或者直接找不到类。我的做法是 DataStream 项目就只引 Source 连接器Pipeline 走官方发行包自带的 lib 目录两边物理隔离。3.2 DataStreamMySqlSource 的必填项与反序列化public class MysqlCdcToPrintJob { public static void main(String[] args) throws Exception { StreamExecutionEnvironment env StreamExecutionEnvironment.getExecutionEnvironment(); env.setParallelism(1); // 验证阶段先单并行度方便对日志 // 位点信息以状态形式存在 Checkpoint 里不开 CK 就没有断点续传能力 env.enableCheckpointing(5000L, CheckpointingMode.EXACTLY_ONCE); env.getCheckpointConfig().setCheckpointTimeout(60 * 1000L); env.getCheckpointConfig().setMinPauseBetweenCheckpoints(3000L); // 任务被取消时保留最后一次 CK方便排查和手动恢复 env.getCheckpointConfig().enableExternalizedCheckpoints( CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION); env.getCheckpointConfig().setCheckpointStorage(hdfs://node01:8020/flink-cdc/ck); System.setProperty(HADOOP_USER_NAME, flink); MySqlSourceString source MySqlSource.Stringbuilder() .hostname(node02) .port(3306) .databaseList(test) // 捕获的库 .tableList(test.t1, test.t2) // 捕获的表格式 库.表支持多个 .username(cdc_user) .password(Cdc123456) .serverId(5400-5404) // 范围写法按并行度给每个 reader 分一个 .serverTimeZone(Asia/Shanghai) // 不写可能差 8 小时 .startupOptions(StartupOptions.initial()) // 先全量快照再切增量 .deserializer(new JsonDebeziumDeserializationSchema()) // 输出 JSON 字符串 .build(); env.fromSource(source, WatermarkStrategy.noWatermarks(), mysql-cdc-source) .print(); env.execute(mysql-cdc-demo); } }逐个说关键点。databaseList和tableList是包含关系只给库不给表会同步整库给了表就只同步这几张tableList的写法必须是「库.表」写错成t1会静默过滤掉所有数据日志里还看不到明显报错。startupOptions决定了第一次启动的行为资料里列了五种实际常用的是三种initial()先做全量快照再读增量适合新接入的业务latest()直接从当前 Binlog 末尾开始历史数据不管适合只关心增量或者已经用别的方式灌过历史earliest()从 Binlog 起点读风险最大Binlog 被清理过就直接失败。生产上我一般先用initial()灌一次历史之后从 Savepoint 恢复不会再走快照。serverId建议显式写。不写时连接器会随机生成一个单任务看不出问题多任务时撞上的概率不低报错信息又晦涩。范围写法里数字的个数对应最大 reader 数并行度超过这个数会报错5400-5404最多支持 5 个。deserializer用的是JsonDebeziumDeserializationSchema输出的是 Debezium 风格的 JSON包含before、after、op、ts_ms几个字段op的取值c/u/d/r分别对应插入、更新、删除、快照读。下游要写 Doris 就得解析这个结构或者换成自定义反序列化直接产出目标格式。3.3 Checkpoint 与 Savepoint断点续传的后悔药上面代码里那几行 Checkpoint 配置不是可选项。Flink CDC 的读取位点保存在算子状态里没有 Checkpoint 就没有状态任务一停位点就丢重启要么重头全量、要么从最新开始两个结果都不能接受。# 提交任务到已有集群注意 -c 指定主类jar 在后 bin/flink run -m node01:8081 \ -c com.example.cdc.MysqlCdcToPrintJob \ ./flink-cdc-demo.jar # 升级代码或改配置前先打一个 Savepoint bin/flink savepoint jobId hdfs://node01:8020/flink-cdc/save # 从 Savepoint 恢复其他参数与提交时保持一致 bin/flink run -s hdfs://node01:8020/flink-cdc/save/savepoint-xxxxxx \ -c com.example.cdc.MysqlCdcToPrintJob \ ./flink-cdc-demo.jar恢复时最容易出问题的是并行度变化。状态里按算子实例分配了 key group并行度从 1 改成 2 之后如果代码里没给算子设置固定的uidFlink 无法把旧状态映射到新算子会退化成无状态启动表现出来就是「恢复成功了但数据重复了一大批」。习惯做法是给每个算子和 Source 都加上.uid(xxx)改代码时不动这个 id。另一个细节是EXACTLY_ONCE在源端的含义。对 MySQL CDC 来说它保证的是位点提交与 Checkpoint 对齐配合支持两阶段提交的 SinkDoris 连接器就支持才能端到端做到不重不丢如果 Sink 只支持 at-least-once那实际语义就是至少一次重复由下游去重解决。3.4 Flink SQL五分钟起一条同步链路的写法不想写 Java 的时候SQL 方式更直接尤其适合验证 Binlog 通不通。-- Flink SQL 里把 MySQL 表映射成一张有主键的源表 CREATE TABLE mysql_t1 ( id STRING, name STRING, PRIMARY KEY (id) NOT ENFORCED -- 主键声明必须加否则 upsert 语义退化 ) WITH ( connector mysql-cdc, hostname node02, port 3306, username cdc_user, password Cdc123456, database-name test, table-name t1, server-time-zone Asia/Shanghai, scan.startup.mode initial ); -- 打印到标准输出本地 IDE 直接跑就能看到 changelog SELECT * FROM mysql_t1;PRIMARY KEY ... NOT ENFORCED是这段 DDL 里最容易漏的一句。不加主键声明Flink 会把这张表当成 append 流后续接 Doris 或做聚合时拿不到更新语义看到的永远是插入记录。NOT ENFORCED表示 Flink 不做主键约束校验只是用它来推导 changelog 类型别理解成「主键可以重复」。scan.startup.mode与 DataStream 的startupOptions一一对应取值initial、latest、earliest、specific-offset、timestamp。SQL 方式下改这个参数比改代码快得多排查问题时我经常用它在本地快速验证某个时间点的数据能不能读出来。4. MySQL 到 Doris3.0 的 Pipeline 与路由怎么写前面两种写法都得自己写代码、自己接 Sink。3.0 真正省事的地方是 Pipeline整库同步不用写一行 Java一份 YAML 描述 source、sink 和运行参数用flink-cdc.sh提交全量和增量、建表、加列都由框架处理。代价是可干预的细节变少出问题时只能靠日志和配置反推。4.1 目录、jar 与启动脚本# 解压发行包到统一目录 tar -zxvf flink-cdc-3.0.0-bin.tar.gz -C /opt/module/ # Pipeline 连接器不是内置的需要手动放进 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/ # 存放作业配置 mkdir -p /opt/module/flink-cdc-3.0.0/joblib 目录里少了任何一个连接器启动时都会报「找不到 type 为 mysql/doris 的 connector 工厂」这个报错信息不会告诉你缺哪个包只能凭经验判断。我的习惯是解压完先ls lib/看一眼确认两个连接器都在再去写 YAML。4.2 YAML 三段式source、sink、pipelinesource: type: mysql hostname: node02 port: 3306 username: cdc_user password: Cdc123456 tables: test.\.* # 正则匹配 test 库下所有表 server-id: 5400-5404 # 与并行度匹配别和别的任务重叠 server-time-zone: Asia/Shanghai sink: type: doris fenodes: node03:8030 # FE 的 HTTP 端口按实际部署改 username: root password: ****** table.create.properties.light_schema_change: true # 支持加列等轻量 schema 变更 table.create.properties.replication_num: 1 # 副本数不能超过 BE 数量 pipeline: name: Sync MySQL test to Doris parallelism: 1tables用的是正则test.\.*表示 test 库下全部表。写正则时注意转义.在正则里匹配任意字符要匹配表名分隔符必须写成\.否则可能误匹配到别的库名。server-id的范围要和parallelism对得上。并行度 1 时给一个 id 就够但建议留出余量因为快照阶段每个分片可能各起一个 reader范围太小会报「no server id available」。table.create.properties.*是透传给 Doris 建表语句的属性。light_schema_change打开后源端加列能自动同步过去replication_num必须小于等于 BE 节点数单节点测试环境写 1生产上一般写 3写成 3 但只有 1 个 BE 会直接建表失败。Doris 侧不用提前建表连接器会根据源表结构自动建但库要预先存在-- 连到 Doris FE 的 9030 端口不是 8030 mysql -uroot -p -P9030 -hnode03 CREATE DATABASE IF NOT EXISTS test;4.3 路由同步多库并一库的表名重写实际业务里经常要把多个源库的表收进同一个 Doris 库或者给表名加前缀避免冲突。3.0 的 Pipeline 支持在pipeline段里配路由规则把源表名重写到目标表名。source: type: mysql hostname: node02 port: 3306 username: cdc_user password: Cdc123456 tables: test_route.\.* server-id: 5410-5414 server-time-zone: Asia/Shanghai sink: type: doris fenodes: node03:8030 username: root password: ****** table.create.properties.light_schema_change: true table.create.properties.replication_num: 1 pipeline: name: Sync test_route to Doris with route parallelism: 1 route: - source-table: test_route\.t1 sink-table: test_route.t1_route # 落到 Doris 时改名为 t1_route - source-table: test_route\.t2 sink-table: test_route.t2_route路由规则的source-table同样是正则sink-table是目标库表全名。配了 route 之后没有匹配到的表默认会被忽略所以要么把规则写全要么保留一条通配兜底。这一点资料里没有展开但它是我见过最容易导致「同步了一半表」的原因——加表的时候忘了改 route任务显示运行正常数据就是不来。4.4 启动顺序与观测点# 1. Flink 侧开启周期性 CheckpointPipeline 依赖它提交位点和两阶段提交 vim /opt/module/flink-1.18.0/conf/flink-conf.yaml # 追加一行 # execution.checkpointing.interval: 5000 # 2. 分发配置并启动集群 /opt/module/flink-1.18.0/bin/start-cluster.sh # 3. 启动 Doris FE / BE顺序不能反 /opt/module/doris/fe/bin/start_fe.sh --daemon /opt/module/doris/be/bin/start_be.sh --daemon # 4. 提交 Pipeline 作业 /opt/module/flink-cdc-3.0.0/bin/flink-cdc.sh \ /opt/module/flink-cdc-3.0.0/job/mysql-to-doris.yamlexecution.checkpointing.interval这一行是必须的。Pipeline 的位点提交、Doris 的两阶段提交都挂在 Checkpoint 上不配这个参数任务能跑起来、数据也能进去但重启后行为完全不可控属于典型的「当时没感觉出事查不出」的配置。提交成功后去 Flink Web UI 看三件事作业是否在 RUNNING 状态、Checkpoint 是否有成功记录、Source 算子的numRecordsOut是否在增长。再去 Doris 侧SELECT COUNT(*)看数据落没落。资料里的验证方式是往 MySQL 里做新增、修改、加列然后刷新 Doris 观察变化——加列这一步最能体现light_schema_change是否生效。5. 避坑与排查五个让同步任务翻车的现场下面这些不是理论推演是这条链路上出现频率最高、且报错信息最不友好的几类问题。每条按现象、原因、解决三段写遇到时可以直接对照。5.1 全量阶段一切正常切到增量后一条数据都不来现象任务启动后能读出历史数据日志里也能看到快照完成的记录之后 MySQL 里怎么改都不见新数据作业状态还是 RUNNING。原因通常有三个binlog_format不是 ROW增量事件解析不出来但不报错binlog-do-db白名单里没有当前库Binlog 里压根没记录这次变更或者startupOptions用了latest()任务启动位点之后才有变更之前的改动被跳过。第四个可能是账号缺REPLICATION SLAVE权限但这种情况一般会直接抛异常反而不难查。解决先SHOW VARIABLES LIKE binlog_format和SHOW MASTER STATUS确认格式与位点再核对binlog-do-db是否覆盖目标库最后检查代码里的启动模式。三步按顺序排查基本十分钟内能定位。5.2 任务重启后数据重复一大批或者直接从头再灌一遍现象改了个参数重启作业Doris 里主键相同的行数翻倍或者本应续传的任务重新做了一次全量快照。原因没开 Checkpoint位点根本没保存或者从 Savepoint 恢复时并行度变了、算子uid变了状态映射不上Flink 按无状态启动处理。还有一种情况是RETAIN_ON_CANCELLATION没开取消任务时最后一次 Checkpoint 被清掉只能重来。解决固定uid恢复时保持并行度不变Checkpoint 目录用 HDFS 这类持久化存储而不是本地磁盘。Doris 侧如果是主键模型重复写入会被覆盖问题不明显如果是明细模型重复数据会一直留着只能靠手工清理。5.3 同一个 MySQL 实例挂第二个任务就报 server-id 冲突现象第一个任务跑得好好的第二个任务启动后报连接异常日志里出现 slave 注册失败之类的提示或者两个任务互相把对方顶掉。原因两个任务用了同一段server-id。MySQL 把每个 Binlog 消费者当成一个从库从库 id 必须全局唯一重复时后连上来的会被拒绝或者把先连上的踢掉表现出来就是任务反复重启。解决给每个任务分配不重叠的区间比如第一个5400-5404、第二个5410-5414、Pipeline 用5420-5424。区间大小按最大并行度留足余量改配置比改代码便宜。5.4 Doris 侧建表失败或者写入报版本过多现象任务起来后 Sink 阶段报错提示副本数不满足要求或者运行一段时间后写入变慢、报 tablet 版本数量超限。原因replication_num设成了 3 但集群只有 1 个 BE或者 Checkpoint 间隔设得太短导入频率过高小文件和小版本堆积Doris 的 compaction 跟不上。解决replication_num与 BE 数量对齐测试环境写 1生产写 3Checkpoint 间隔不要低于 5 秒数据量大时适当拉长到 10 到 30 秒用批次换吞吐。另外注意 MySQL 的TINYINT、DECIMAL映射到 Doris 时精度可能变化涉及金额字段建议在源端就统一成DECIMAL(18,2)。5.5 时间字段差 8 小时或者中文写进去变问号现象MySQL 里2024-05-01 10:00:00Doris 里查出来是02:00:00或者姓名列出现乱码。原因时区没显式指定连接器默认用 UTC 解析而 MySQL 服务端是东八区乱码多半是连接串或建表时字符集不一致。解决DataStream 里加.serverTimeZone(Asia/Shanghai)SQL 里加server-time-zone Asia/ShanghaiPipeline 的 YAML 里写server-time-zone: Asia/Shanghai与UTC8等价但更直观。建库建表统一用utf8mb4别用utf8后者在 Doris 和 MySQL 里的实际含义不完全一致。6. 上线前的验收三条口径和一个改不掉的习惯同步跑起来不等于能交付。我交付前的固定动作是三条口径对账。第一条比总数MySQL 和 Doris 各跑一次SELECT COUNT(*)数字必须一致不一致先看是不是全量阶段有表被 route 规则漏掉了。第二条比主键集合用SELECT id FROM t1 ORDER BY id LIMIT 100抽前后各 100 个主键确认没有断裂。第三条比最新变更在 MySQL 里插一条带时间戳的记录等一个 Checkpoint 周期后在 Doris 里按时间倒序查看这条是不是最新。-- 主键抽样两侧结果直接 diff比全表比对快得多 SELECT id, name FROM test.t1 ORDER BY id DESC LIMIT 100; -- 变更延迟观测插入后记录当前时间到 Doris 侧算差值 INSERT INTO test.t1 VALUES (9999, CONCAT(probe_, UNIX_TIMESTAMP()));除了对账我还会额外做一次「加列演练」在 MySQL 的 t1 上加一个age INT看 Doris 侧能不能自动多出这一列并写入新数据。这一步能同时验证light_schema_change配置、Pipeline 的 schema 演进能力和账号的 DDL 权限比事后出问题再补要划算得多。从那以后我每次上线这类同步任务都强制走一遍「总数、主键、最新一条、加列」四步哪怕业务方催得再急也不跳过——有一次省了加列演练三天后源端上线新字段Doris 侧写入直接失败回滚和补数花了整整一个通宵。希望帮到你。本文还有配套的精品资源点击获取 
长河君 · 2026/11/4 2:45:27 · 阅读 1.2k Flink CDC 3.0：MySQL Binlog 实时同步 Doris 实战
简介面向具备一定 Flink 基础的大数据研发人员这份尚硅谷研究院编写的 Flink CDC 3.0 文档围绕 MySQL 变更数据实时捕获与同步至 Doris 展开先厘清 CDC 概念及基于查询、基于 Binlog 两类方案的差异再引入 flink-cdc-connectors 组件的使用方式帮助读者建立从原理到落地的完整认知。包内为 1 个 docx 文件压缩包约 145KB篇幅集中、结构完整按 CDC 简介、案例实操等章节组织便于顺序阅读或按需查阅。文档给出数据准备、DataStream 与 Flink SQL 两种实现路径并完整覆盖 MySQL 到 Doris 的 Streaming ETL环境搭建、配置文件编写、任务启动与测试均有对应说明同时强调开启 Binlog、设置检查点等关键配置便于排查同步链路中的一致性与稳定性问题。目前已有 1789 人学习适合希望独立完成实时同步项目部署、积累参数调整经验的工程师参考。1. 从一次凌晨两点的对账说起Flink CDC 3.0 到底补上了哪块短板业务库里的订单表一天几百万行变动报表库却只能等凌晨跑完批处理才更新第二天上班第一件事就是对账差一行就得翻一上午日志。这套以 Flink CDC 3.0 为主线的实操资料讲的正是把 MySQL 的变更通过 Binlog 实时捕获、再同步进 Doris 的完整链路从 CDC 概念、Binlog 开启、DataStream 与 Flink SQL 两种写法一直到 3.0 才有的 Pipeline YAML 整库同步与路由。它适合已经能跑起 Flink 任务、又被 T1 报表卡住的工程师新手可以照着命令一行行敲熟手能直接看参数边界和失败现场。资料本身是一份按章节推进的实操文档配置和代码占比很高不是概念科普。2. 选型先落地查询型 CDC 和 Binlog 型 CDC 差的不只是延迟动手之前把路线定错后面所有配置都是白费。CDC 这个词被用得很泛Sqoop 定时拉一遍也叫捕获变更Canal 读 Binlog 也叫捕获变更但两者的代价模型完全不同。这一章先把两条路线的账算清楚再把 MySQL 侧三个绕不开的参数定死最后造一批可验证的数据让后面每个实验都有据可查。2.1 一张表把两条路线的代价分开维度基于查询的 CDC基于 Binlog 的 CDC代表产品Sqoop、DataXCanal、Maxwell、Debezium、Flink CDC执行模式BatchStreaming能否捕获删除否只能靠全量比对推断是delete 事件带完整主键变化延迟高取决于调度周期低通常在秒级对源库压力每次扫描都压库大表容易打满 IO只读 Binlog 流压力集中在网络与解析是否要求开启 Binlog否是且必须是 ROW 格式看完这张表基本就能下判断只要业务要求「删了也要同步掉」「改了要在一分钟内出现在报表里」查询型路线就直接出局。真正需要权衡的是运维成本——Binlog 路线要求 DBA 配合改配置、开权限、规划 server-id这些动作在规范化程度不高的团队里往往比写代码更耗时间。Flink CDC 的位置比较特殊它把 Debezium 的解析能力包进了 Flink Source全量和增量在同一个作业里完成启动时先做快照读读完自动切到 Binlog 流不需要额外维护一张位点表也不需要写两套代码。这一点在 3.0 里被进一步放大整库同步只需要一份 YAML。2.2 binlog_format、server-id、账号权限三个必须在动手前定死的参数MySQL 侧的配置是所有失败案例的第一现场。下面这段是常见的最小配置改完必须重启 MySQL 才生效热加载不行。# /etc/my.cnf [mysqld] 段内 server-id 1 # 打开 binlog值会作为文件前缀 log-bin mysql-bin # 必须是 ROWSTATEMENT/MIXED 拿不到变更前后的完整行镜像 binlog_format ROW # 只记录白名单内的库漏掉的库后面怎么配都同步不过来 binlog-do-db test binlog-do-db test_route # 行镜像保留完整字段遇到 UPDATE 未改动的列也能拿到旧值 binlog_row_image FULLbinlog_format必须设成ROW这是硬性前提。设成STATEMENT时 Binlog 里存的是 SQL 语句Flink CDC 拿不到变更前后的行数据反序列化阶段就会报字段缺失设成MIXED更麻烦平时跑得好好的某条语句被 MySQL 判定为不确定函数后切成 ROW任务时好时坏属于典型的玄学故障。binlog-do-db是白名单机制不是黑名单。它的判定依据是「当前默认库」所以跨库更新语句可能出现记录行为与预期不一致的情况。我的习惯是同步哪些库就把哪些库全部列进去后续新增库一定要记得回来补配置并重启——这个坑后面还会提。server-id在单机单任务时随便写但只要同一个 MySQL 实例上挂两个以上的同步任务就必须规划成不重叠的区间。Flink CDC 3.0 允许写成5400-5404这样的范围意思是给最多 5 个并行 reader 各分配一个 id两个任务都用5400-5404MySQL 会直接拒绝后连上来的那个。账号权限同样要给够只给 SELECT 是不够的-- 同步账号全量阶段要读表增量阶段要读 binlog CREATE USER cdc_user% IDENTIFIED BY Cdc123456; GRANT SELECT, RELOAD, SHOW DATABASES, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO cdc_user%; FLUSH PRIVILEGES;REPLICATION SLAVE和REPLICATION CLIENT是读 Binlog 的门票缺了会在增量启动时抛权限异常RELOAD用于执行FLUSH TABLES之类的操作全量快照阶段会用到SHOW DATABASES让连接器能枚举库表结构。授权语句里的库名用*.*更省事生产上可以按需收窄到具体库。2.3 建库建表造数让后面每个实验都有数据可查配置改完不代表链路通得先有能验证的数据。下面这套结构来自资料里的案例两个库、三张表、主键都是字符串故意留了一张字段不一致的表t3 只有 id 和 sex用来观察 schema 差异时的表现。-- 源库与目标库分开建方便后面演示路由同步 CREATE DATABASE IF NOT EXISTS test; CREATE DATABASE IF NOT EXISTS 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)); 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); -- 第二个库的表结构与第一个库一致但数据主键错开便于区分来源 USE test_route; CREATE TABLE t1 (id VARCHAR(255) PRIMARY KEY, name VARCHAR(255)); INSERT INTO t1 VALUES (2001,zhaoliu),(2002,qianqi);建完之后别急着写 Flink 代码先确认 Binlog 真的在写# 1. 确认格式与是否开启 mysql -ucdc_user -p -e SHOW VARIABLES LIKE binlog_format; SHOW VARIABLES LIKE log_bin; # 2. 看当前写到哪个文件、哪个位点 mysql -ucdc_user -p -e SHOW MASTER STATUS\G # 3. 制造一条变更后再看位点是否前移位点不动说明白名单没覆盖这个库 mysql -ucdc_user -p -e INSERT INTO test.t1 VALUES (1004,sunba); SHOW MASTER STATUS\G三步都通过才说明 MySQL 侧是干净的。很多「任务启动了但没数据」的问题其实在这一步就已经能暴露出来只是现场直接跳到了 Flink 日志里翻白白多花两小时。3. DataStream 与 Flink SQL同一份 MySQL 数据的两条入口Flink CDC 3.0 提供了不止一种接入方式DataStream API 控制力最强能精确设置 Checkpoint、重启策略和状态后端Flink SQL 写法最短适合快速验证和把它当成一张维表/源表嵌进更复杂的查询里。两条路读的是同一个 Binlog区别在于你能干预多少细节。3.1 依赖与打包版本错一个就是 ClassNotFound依赖这块翻车率极高核心原因是 Flink 主版本、CDC 连接器版本、MySQL 驱动版本三者必须互相对齐。下面这份 POM 是资料里验证过能跑的组合Flink 1.18 配 CDC 3.0.0。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 !-- Flink 运行时与 Table 相关依赖provided 还是 compile 取决于提交方式 -- dependency groupIdorg.apache.flink/groupId artifactIdflink-streaming-java/artifactId version${flink.version}/version /dependency dependency groupIdorg.apache.flink/groupId artifactIdflink-clients/artifactId version${flink.version}/version /dependency dependency groupIdorg.apache.flink/groupId artifactIdflink-table-planner_2.12/artifactId version${flink.version}/version /dependency dependency groupIdorg.apache.flink/groupId artifactIdflink-table-runtime/artifactId version${flink.version}/version /dependency !-- CDC Source 连接器坐标仍在 com.ververica 下 -- dependency groupIdcom.ververica/groupId artifactIdflink-connector-mysql-cdc/artifactId version3.0.0/version /dependency !-- 驱动版本要和 MySQL 服务端匹配8.0 服务端别用 5.1 驱动 -- dependency groupIdmysql/groupId artifactIdmysql-connector-java/artifactId version8.0.31/version /dependency /dependencies打包用maven-assembly-plugin的jar-with-dependencies描述符把所有依赖塞进一个胖包提交时只传一个文件。这里有个容易混淆的点3.0 里的 Pipeline 连接器后面第 4 章要用坐标在org.apache.flink下和这里的com.ververica:flink-connector-mysql-cdc不是同一个包两者混着引会出现类冲突或者直接找不到类。我的做法是 DataStream 项目就只引 Source 连接器Pipeline 走官方发行包自带的 lib 目录两边物理隔离。3.2 DataStreamMySqlSource 的必填项与反序列化public class MysqlCdcToPrintJob { public static void main(String[] args) throws Exception { StreamExecutionEnvironment env StreamExecutionEnvironment.getExecutionEnvironment(); env.setParallelism(1); // 验证阶段先单并行度方便对日志 // 位点信息以状态形式存在 Checkpoint 里不开 CK 就没有断点续传能力 env.enableCheckpointing(5000L, CheckpointingMode.EXACTLY_ONCE); env.getCheckpointConfig().setCheckpointTimeout(60 * 1000L); env.getCheckpointConfig().setMinPauseBetweenCheckpoints(3000L); // 任务被取消时保留最后一次 CK方便排查和手动恢复 env.getCheckpointConfig().enableExternalizedCheckpoints( CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION); env.getCheckpointConfig().setCheckpointStorage(hdfs://node01:8020/flink-cdc/ck); System.setProperty(HADOOP_USER_NAME, flink); MySqlSourceString source MySqlSource.Stringbuilder() .hostname(node02) .port(3306) .databaseList(test) // 捕获的库 .tableList(test.t1, test.t2) // 捕获的表格式 库.表支持多个 .username(cdc_user) .password(Cdc123456) .serverId(5400-5404) // 范围写法按并行度给每个 reader 分一个 .serverTimeZone(Asia/Shanghai) // 不写可能差 8 小时 .startupOptions(StartupOptions.initial()) // 先全量快照再切增量 .deserializer(new JsonDebeziumDeserializationSchema()) // 输出 JSON 字符串 .build(); env.fromSource(source, WatermarkStrategy.noWatermarks(), mysql-cdc-source) .print(); env.execute(mysql-cdc-demo); } }逐个说关键点。databaseList和tableList是包含关系只给库不给表会同步整库给了表就只同步这几张tableList的写法必须是「库.表」写错成t1会静默过滤掉所有数据日志里还看不到明显报错。startupOptions决定了第一次启动的行为资料里列了五种实际常用的是三种initial()先做全量快照再读增量适合新接入的业务latest()直接从当前 Binlog 末尾开始历史数据不管适合只关心增量或者已经用别的方式灌过历史earliest()从 Binlog 起点读风险最大Binlog 被清理过就直接失败。生产上我一般先用initial()灌一次历史之后从 Savepoint 恢复不会再走快照。serverId建议显式写。不写时连接器会随机生成一个单任务看不出问题多任务时撞上的概率不低报错信息又晦涩。范围写法里数字的个数对应最大 reader 数并行度超过这个数会报错5400-5404最多支持 5 个。deserializer用的是JsonDebeziumDeserializationSchema输出的是 Debezium 风格的 JSON包含before、after、op、ts_ms几个字段op的取值c/u/d/r分别对应插入、更新、删除、快照读。下游要写 Doris 就得解析这个结构或者换成自定义反序列化直接产出目标格式。3.3 Checkpoint 与 Savepoint断点续传的后悔药上面代码里那几行 Checkpoint 配置不是可选项。Flink CDC 的读取位点保存在算子状态里没有 Checkpoint 就没有状态任务一停位点就丢重启要么重头全量、要么从最新开始两个结果都不能接受。# 提交任务到已有集群注意 -c 指定主类jar 在后 bin/flink run -m node01:8081 \ -c com.example.cdc.MysqlCdcToPrintJob \ ./flink-cdc-demo.jar # 升级代码或改配置前先打一个 Savepoint bin/flink savepoint jobId hdfs://node01:8020/flink-cdc/save # 从 Savepoint 恢复其他参数与提交时保持一致 bin/flink run -s hdfs://node01:8020/flink-cdc/save/savepoint-xxxxxx \ -c com.example.cdc.MysqlCdcToPrintJob \ ./flink-cdc-demo.jar恢复时最容易出问题的是并行度变化。状态里按算子实例分配了 key group并行度从 1 改成 2 之后如果代码里没给算子设置固定的uidFlink 无法把旧状态映射到新算子会退化成无状态启动表现出来就是「恢复成功了但数据重复了一大批」。习惯做法是给每个算子和 Source 都加上.uid(xxx)改代码时不动这个 id。另一个细节是EXACTLY_ONCE在源端的含义。对 MySQL CDC 来说它保证的是位点提交与 Checkpoint 对齐配合支持两阶段提交的 SinkDoris 连接器就支持才能端到端做到不重不丢如果 Sink 只支持 at-least-once那实际语义就是至少一次重复由下游去重解决。3.4 Flink SQL五分钟起一条同步链路的写法不想写 Java 的时候SQL 方式更直接尤其适合验证 Binlog 通不通。-- Flink SQL 里把 MySQL 表映射成一张有主键的源表 CREATE TABLE mysql_t1 ( id STRING, name STRING, PRIMARY KEY (id) NOT ENFORCED -- 主键声明必须加否则 upsert 语义退化 ) WITH ( connector mysql-cdc, hostname node02, port 3306, username cdc_user, password Cdc123456, database-name test, table-name t1, server-time-zone Asia/Shanghai, scan.startup.mode initial ); -- 打印到标准输出本地 IDE 直接跑就能看到 changelog SELECT * FROM mysql_t1;PRIMARY KEY ... NOT ENFORCED是这段 DDL 里最容易漏的一句。不加主键声明Flink 会把这张表当成 append 流后续接 Doris 或做聚合时拿不到更新语义看到的永远是插入记录。NOT ENFORCED表示 Flink 不做主键约束校验只是用它来推导 changelog 类型别理解成「主键可以重复」。scan.startup.mode与 DataStream 的startupOptions一一对应取值initial、latest、earliest、specific-offset、timestamp。SQL 方式下改这个参数比改代码快得多排查问题时我经常用它在本地快速验证某个时间点的数据能不能读出来。4. MySQL 到 Doris3.0 的 Pipeline 与路由怎么写前面两种写法都得自己写代码、自己接 Sink。3.0 真正省事的地方是 Pipeline整库同步不用写一行 Java一份 YAML 描述 source、sink 和运行参数用flink-cdc.sh提交全量和增量、建表、加列都由框架处理。代价是可干预的细节变少出问题时只能靠日志和配置反推。4.1 目录、jar 与启动脚本# 解压发行包到统一目录 tar -zxvf flink-cdc-3.0.0-bin.tar.gz -C /opt/module/ # Pipeline 连接器不是内置的需要手动放进 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/ # 存放作业配置 mkdir -p /opt/module/flink-cdc-3.0.0/joblib 目录里少了任何一个连接器启动时都会报「找不到 type 为 mysql/doris 的 connector 工厂」这个报错信息不会告诉你缺哪个包只能凭经验判断。我的习惯是解压完先ls lib/看一眼确认两个连接器都在再去写 YAML。4.2 YAML 三段式source、sink、pipelinesource: type: mysql hostname: node02 port: 3306 username: cdc_user password: Cdc123456 tables: test.\.* # 正则匹配 test 库下所有表 server-id: 5400-5404 # 与并行度匹配别和别的任务重叠 server-time-zone: Asia/Shanghai sink: type: doris fenodes: node03:8030 # FE 的 HTTP 端口按实际部署改 username: root password: ****** table.create.properties.light_schema_change: true # 支持加列等轻量 schema 变更 table.create.properties.replication_num: 1 # 副本数不能超过 BE 数量 pipeline: name: Sync MySQL test to Doris parallelism: 1tables用的是正则test.\.*表示 test 库下全部表。写正则时注意转义.在正则里匹配任意字符要匹配表名分隔符必须写成\.否则可能误匹配到别的库名。server-id的范围要和parallelism对得上。并行度 1 时给一个 id 就够但建议留出余量因为快照阶段每个分片可能各起一个 reader范围太小会报「no server id available」。table.create.properties.*是透传给 Doris 建表语句的属性。light_schema_change打开后源端加列能自动同步过去replication_num必须小于等于 BE 节点数单节点测试环境写 1生产上一般写 3写成 3 但只有 1 个 BE 会直接建表失败。Doris 侧不用提前建表连接器会根据源表结构自动建但库要预先存在-- 连到 Doris FE 的 9030 端口不是 8030 mysql -uroot -p -P9030 -hnode03 CREATE DATABASE IF NOT EXISTS test;4.3 路由同步多库并一库的表名重写实际业务里经常要把多个源库的表收进同一个 Doris 库或者给表名加前缀避免冲突。3.0 的 Pipeline 支持在pipeline段里配路由规则把源表名重写到目标表名。source: type: mysql hostname: node02 port: 3306 username: cdc_user password: Cdc123456 tables: test_route.\.* server-id: 5410-5414 server-time-zone: Asia/Shanghai sink: type: doris fenodes: node03:8030 username: root password: ****** table.create.properties.light_schema_change: true table.create.properties.replication_num: 1 pipeline: name: Sync test_route to Doris with route parallelism: 1 route: - source-table: test_route\.t1 sink-table: test_route.t1_route # 落到 Doris 时改名为 t1_route - source-table: test_route\.t2 sink-table: test_route.t2_route路由规则的source-table同样是正则sink-table是目标库表全名。配了 route 之后没有匹配到的表默认会被忽略所以要么把规则写全要么保留一条通配兜底。这一点资料里没有展开但它是我见过最容易导致「同步了一半表」的原因——加表的时候忘了改 route任务显示运行正常数据就是不来。4.4 启动顺序与观测点# 1. Flink 侧开启周期性 CheckpointPipeline 依赖它提交位点和两阶段提交 vim /opt/module/flink-1.18.0/conf/flink-conf.yaml # 追加一行 # execution.checkpointing.interval: 5000 # 2. 分发配置并启动集群 /opt/module/flink-1.18.0/bin/start-cluster.sh # 3. 启动 Doris FE / BE顺序不能反 /opt/module/doris/fe/bin/start_fe.sh --daemon /opt/module/doris/be/bin/start_be.sh --daemon # 4. 提交 Pipeline 作业 /opt/module/flink-cdc-3.0.0/bin/flink-cdc.sh \ /opt/module/flink-cdc-3.0.0/job/mysql-to-doris.yamlexecution.checkpointing.interval这一行是必须的。Pipeline 的位点提交、Doris 的两阶段提交都挂在 Checkpoint 上不配这个参数任务能跑起来、数据也能进去但重启后行为完全不可控属于典型的「当时没感觉出事查不出」的配置。提交成功后去 Flink Web UI 看三件事作业是否在 RUNNING 状态、Checkpoint 是否有成功记录、Source 算子的numRecordsOut是否在增长。再去 Doris 侧SELECT COUNT(*)看数据落没落。资料里的验证方式是往 MySQL 里做新增、修改、加列然后刷新 Doris 观察变化——加列这一步最能体现light_schema_change是否生效。5. 避坑与排查五个让同步任务翻车的现场下面这些不是理论推演是这条链路上出现频率最高、且报错信息最不友好的几类问题。每条按现象、原因、解决三段写遇到时可以直接对照。5.1 全量阶段一切正常切到增量后一条数据都不来现象任务启动后能读出历史数据日志里也能看到快照完成的记录之后 MySQL 里怎么改都不见新数据作业状态还是 RUNNING。原因通常有三个binlog_format不是 ROW增量事件解析不出来但不报错binlog-do-db白名单里没有当前库Binlog 里压根没记录这次变更或者startupOptions用了latest()任务启动位点之后才有变更之前的改动被跳过。第四个可能是账号缺REPLICATION SLAVE权限但这种情况一般会直接抛异常反而不难查。解决先SHOW VARIABLES LIKE binlog_format和SHOW MASTER STATUS确认格式与位点再核对binlog-do-db是否覆盖目标库最后检查代码里的启动模式。三步按顺序排查基本十分钟内能定位。5.2 任务重启后数据重复一大批或者直接从头再灌一遍现象改了个参数重启作业Doris 里主键相同的行数翻倍或者本应续传的任务重新做了一次全量快照。原因没开 Checkpoint位点根本没保存或者从 Savepoint 恢复时并行度变了、算子uid变了状态映射不上Flink 按无状态启动处理。还有一种情况是RETAIN_ON_CANCELLATION没开取消任务时最后一次 Checkpoint 被清掉只能重来。解决固定uid恢复时保持并行度不变Checkpoint 目录用 HDFS 这类持久化存储而不是本地磁盘。Doris 侧如果是主键模型重复写入会被覆盖问题不明显如果是明细模型重复数据会一直留着只能靠手工清理。5.3 同一个 MySQL 实例挂第二个任务就报 server-id 冲突现象第一个任务跑得好好的第二个任务启动后报连接异常日志里出现 slave 注册失败之类的提示或者两个任务互相把对方顶掉。原因两个任务用了同一段server-id。MySQL 把每个 Binlog 消费者当成一个从库从库 id 必须全局唯一重复时后连上来的会被拒绝或者把先连上的踢掉表现出来就是任务反复重启。解决给每个任务分配不重叠的区间比如第一个5400-5404、第二个5410-5414、Pipeline 用5420-5424。区间大小按最大并行度留足余量改配置比改代码便宜。5.4 Doris 侧建表失败或者写入报版本过多现象任务起来后 Sink 阶段报错提示副本数不满足要求或者运行一段时间后写入变慢、报 tablet 版本数量超限。原因replication_num设成了 3 但集群只有 1 个 BE或者 Checkpoint 间隔设得太短导入频率过高小文件和小版本堆积Doris 的 compaction 跟不上。解决replication_num与 BE 数量对齐测试环境写 1生产写 3Checkpoint 间隔不要低于 5 秒数据量大时适当拉长到 10 到 30 秒用批次换吞吐。另外注意 MySQL 的TINYINT、DECIMAL映射到 Doris 时精度可能变化涉及金额字段建议在源端就统一成DECIMAL(18,2)。5.5 时间字段差 8 小时或者中文写进去变问号现象MySQL 里2024-05-01 10:00:00Doris 里查出来是02:00:00或者姓名列出现乱码。原因时区没显式指定连接器默认用 UTC 解析而 MySQL 服务端是东八区乱码多半是连接串或建表时字符集不一致。解决DataStream 里加.serverTimeZone(Asia/Shanghai)SQL 里加server-time-zone Asia/ShanghaiPipeline 的 YAML 里写server-time-zone: Asia/Shanghai与UTC8等价但更直观。建库建表统一用utf8mb4别用utf8后者在 Doris 和 MySQL 里的实际含义不完全一致。6. 上线前的验收三条口径和一个改不掉的习惯同步跑起来不等于能交付。我交付前的固定动作是三条口径对账。第一条比总数MySQL 和 Doris 各跑一次SELECT COUNT(*)数字必须一致不一致先看是不是全量阶段有表被 route 规则漏掉了。第二条比主键集合用SELECT id FROM t1 ORDER BY id LIMIT 100抽前后各 100 个主键确认没有断裂。第三条比最新变更在 MySQL 里插一条带时间戳的记录等一个 Checkpoint 周期后在 Doris 里按时间倒序查看这条是不是最新。-- 主键抽样两侧结果直接 diff比全表比对快得多 SELECT id, name FROM test.t1 ORDER BY id DESC LIMIT 100; -- 变更延迟观测插入后记录当前时间到 Doris 侧算差值 INSERT INTO test.t1 VALUES (9999, CONCAT(probe_, UNIX_TIMESTAMP()));除了对账我还会额外做一次「加列演练」在 MySQL 的 t1 上加一个age INT看 Doris 侧能不能自动多出这一列并写入新数据。这一步能同时验证light_schema_change配置、Pipeline 的 schema 演进能力和账号的 DDL 权限比事后出问题再补要划算得多。从那以后我每次上线这类同步任务都强制走一遍「总数、主键、最新一条、加列」四步哪怕业务方催得再急也不跳过——有一次省了加列演练三天后源端上线新字段Doris 侧写入直接失败回滚和补数花了整整一个通宵。希望帮到你。本文还有配套的精品资源点击获取