简介面向具备一定 Flink 基础的大数据研发与实时同步工程师这份由尚硅谷研究院编写的教程文档围绕 Flink CDC 3.0 展开重点解决 MySQL 变更数据实时捕获并稳定同步至 Doris 的落地问题。压缩包仅含 1 个 docx 文件大小约 145KB内容从 CDC 概念与基于查询、基于 Binlog 两种方案对比讲起延伸到 flink-cdc-connectors 组件的使用并给出数据准备、DataStream 与 Flink SQL 两套实操路径。文档还完整覆盖 MySQL 到 Doris 的 Streaming ETL 流程包括环境搭建、配置文件编写、任务启动与测试验证对 Binlog 开启、检查点设置等关键配置也有说明。已有 1789 人学习适合希望快速上手项目部署、理解各配置项作用并按业务调整参数的中高级读者。1. 从一张对不上的看板说起Flink CDC 3.0 值不值得花两天搭一遍某公司的订单库在 MySQL看板和报表查的是 Doris中间靠一个每小时跑一次的批处理搬运。上线第三周出过一次事运营改了一笔订单金额看板上还是旧值客服拿着两个数字来回问。后来换成 Flink CDC 3.0 做 MySQL 到 Doris 的实时同步常驻作业跑着binlog 一有变更毫秒级就能在 Doris 里查到。这份资料讲的正是这条链路CDC 到底怎么选、Flink CDC 3.0 的 DataStream 和 FlinkSQL 两种写法、以及 3.0 新增的 YAML Pipeline 整库同步到 Doris 怎么配。适合已经会写 Flink 作业、但没在生产上跑过 CDC 的数据开发也适合被批处理延迟折磨过的同学。2. CDC 选型Binlog 路线为什么比批量抽数稳2.1 基于查询和基于 Binlog差的不是延迟而是语义很多团队第一次做同步第一反应是写个定时任务select * from t where update_time 上次时间跑完记个水位线。这条路简单、无依赖、出问题好查所以它的边界也特别清楚。把它和 Binlog 路线放在一张表里看差距不在快多少而在于能不能拿到完整的变更语义。维度基于查询Sqoop / DataX / 自写 JDBC基于 BinlogCanal / Maxwell / Debezium / Flink CDC执行模式Batch一次一批Streaming常驻能否捕获物理删除否是变化延迟分钟级到小时级亚秒到秒级源库压力全表扫或索引扫随数据量线性增长读日志文件近似可忽略断点续传依据业务字段时间戳、自增 IDbinlog 位点 Flink checkpoint对源表的要求有可靠时间戳或自增列有主键、binlog_formatROW物理删除这一条最致命。批处理想感知删除只剩一条路把源表主键集合拉回来和目标表做差集。百万级还能忍千万级就是每次同步都全表比对跑一次顶一次。而 row 格式的 binlog 里DELETE 事件带着完整的行镜像主键清清楚楚直接下发就行。update_time这类字段看着能用实际用起来玄学得很。应用服务器时钟漂移、事务提交晚于字段赋值、批量更新时同一秒内几百行、软删除忘记加过滤条件任何一条都能让数据漏掉或者重复。我见过最典型的翻车是业务方为了减少写压力把update_time设成ON UPDATE CURRENT_TIMESTAMP结果批量修数脚本用INSERT ... ON DUPLICATE KEY UPDATE覆盖时没有真正改字段值MySQL 不更新这一列下游就永远不知道这行变过。2.2 Flink CDC 3.0 在架构里的位置从 Source 组件到 PipelineFlink CDC 早期是个 source 组件flink-cdc-connectors提供 MySQL、PostgreSQL 等库的 CDC Source读出来是 DataStream后面接什么 sink、做什么加工全凭自己写。它的好处是灵活可以在中间做双流 join、打宽、过滤甚至用 FlinkSQL 把 CDC 流和维表关联。代价是每加一张表就要改代码、重新打包、重新提交作业表多的时候维护成本很吓人。3.0 引入 Pipeline 连接器之后形态变了。MySQL source、Doris sink、库表路由、表结构变更传播被打包进一个 YAML 文件用bin/flink-cdc.sh job/xxx.yaml提交框架在内部把这份声明翻译成一个 Flink 作业。加表只改一行正则加列不用改代码schema evolution 会顺着链路传到 Doris 侧。两条路怎么选我一般这么判断表在十张以内、中间需要加工逻辑比如多表合并、维度补齐、字段脱敏写 DataStream 或 FlinkSQL可控性最高整库搬迁、几十上百张表、逻辑就是原样同步用 Pipeline省下的是大量重复代码和上线流程。别小看后者——给五十张表各写一遍 sink 建表语句出一次字段类型对不上的事故就够喝一壶。2.3 什么情况下别硬上 CDCCDC 不是万能药有几个场景我会直接劝退。源表没有主键。Flink CDC 的增量快照要靠主键切分 chunkDoris 的 Unique Key 模型也要主键来做 upsert。没有主键的表接进来要么退化成单并行度全表扫要么下游建不了合适的模型更新变成重复插入。遇到这种表先推动业务加主键比在链路里打补丁划算得多。源库根本拿不到 binlog。托管实例有的只给 binlog 文件不给订阅权限有的只保留很短时间binlog_expire_logs_seconds默认 30 天有的被调成几小时。如果全量初始化要跑五个小时而 binlog 只留两小时中途一旦失败重来位点就断了。上线前一定确认保留时长大于最坏情况下的初始化耗时。只需要 T1 的报表。链路越实时运维成本越高checkpoint 要配、位点要监控、故障要能续跑。如果业务方看的就是昨天的汇总数跑批足够别为了技术先进性给自己找活干。需要跨多库多表做重口径加工。CDC 的定位是搬运不是计算。复杂的关联聚合老老实实落到数仓分层里做让 CDC 只负责把 ODS 层喂满。3. 把 MySQL 侧的地基打好Binlog 四个参数与初始化数据3.1 开 Binlog四个参数决定后面所有事CDC 读不到数据九成问题出在 MySQL 这一侧而且基本都是配置问题。需要改的是my.cnf或者my.ini改完必须重启SET GLOBAL只能临时生效重启就回去了。# /etc/my.cnf [mysqld] # 实例唯一标识CDC 连接时会伪装成一个从库挂上去不能和已有的从库冲突 server-id 1 # 开启 binlog值作为日志文件前缀实际文件形如 mysql-bin.000001 log-bin mysql-bin # 必须是 ROWSTATEMENT/MIXED 下拿不到行级前后镜像 binlog_format ROW # 白名单只有这里列出的库产生的变更才会写进 binlog binlog-do-db test binlog-do-db test_route # 保留时长按最坏情况下的全量初始化耗时来定默认 30 天别调到小时级 binlog_expire_logs_seconds 604800四个参数里server-id和binlog-do-db是最容易出事的。前者在于多个 CDC 任务如果共用同一个 server-idMySQL 侧会认为是从库重复接入第二个任务直接连不上或者被踢掉后者在于它是按当前默认库判定的白名单一个事务里USE a; UPDATE b.t ...这种跨库写法判定用的是ab的变更就漏了。改完重启用两条命令确认。-- 确认关键变量 SHOW VARIABLES LIKE log_bin; -- 期望 ON SHOW VARIABLES LIKE binlog_format; -- 期望 ROW SHOW VARIABLES LIKE binlog_do_db; -- 期望看到 test,test_route -- 确认 binlog 文件已经生成 SHOW MASTER STATUS\GSHOW MASTER STATUS在 8.0 里被SHOW BINARY LOG STATUS取代了老版本用前者。输出里的File和Position两列是 CDC 的起点initial模式下会先做全量快照再从这个位置往后读。账号权限也要提前给足。CDC 需要的权限比想象中多SELECT读表结构、RELOAD执行FLUSH TABLES之类、SHOW DATABASES、REPLICATION SLAVE挂从库、REPLICATION CLIENT读位点信息。生产上别直接用 root单独建一个账号密码里避免:和这类会和连接串打架的字符。提示云上托管实例如果只提供内网地址注意 Flink TaskManager 所在机器能不能通到这个地址别在本地 IDE 里跑通了就以为集群也行。3.2 建库建表与初始化数据为后面的路由实验铺路资料里准备了两个库test和test_route字段故意做得简单就为了后面验证整库同步和库表改名。表结构能跑通即可不用纠结业务含义。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) );两个库的表名完全一样这是刻意设计的。后面验证 route 段的时候如果两个库都同步到 Doris 的同一个库表名会撞车只有配上路由规则test_route的表才能落到另一个库里。数据也造两份注意test_route.t2用的是 1004 到 1006和test.t2的主键不重复方便观察是不是真的同步过来了。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); 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);建完先别急着跑 Flink用SELECT COUNT(*)对一下两边的行数记下来。后面每验证一步都拿这个数字做基线比看日志猜要快得多。3.3 依赖坐标与打包别让 classpath 毁掉一上午工程用 Maven 管依赖Flink 版本 1.18.0CDC 连接器 3.0.0MySQL 驱动 8.0.31。这三个版本必须互相兼容随便升一个是常见的翻车起点。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 核心提交到集群时可以改 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 !-- FlinkSQL 走这三个版本必须和 flink-version 完全一致 -- 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 dependency groupIdorg.apache.flink/groupId artifactIdflink-table-api-java-bridge/artifactId version${flink-version}/version /dependency !-- CDC 连接器资料里用的是这个坐标 -- 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打包插件用maven-assembly-plugin的jar-with-dependencies把 CDC 连接器和 MySQL 驱动一起塞进 fat jar。这里有个常被忽略的点Flink 集群的lib目录里通常没有 CDC 连接器如果打包时把它设成provided提交上去就是ClassNotFoundException: com.ververica.cdc.connectors.mysql.source.MySqlSource日志在 TaskManager 上看半天找不到。如果拉不到com.ververica坐标换org.apache.flink:flink-connector-mysql-cdc:3.0.0试一下功能等价只是 groupId 迁移过。这类坐标问题不值得死磕能编译过、能跑通就是对的。3.4 DataStream 版 MysqlSource检查点配不对断点续传就是空话DataStream 写法把每一步都摆在明面上适合需要精细控制的场景。整个类分六块环境、检查点、Source、读取、输出、执行。中间检查点那一段最长也最容易被复制粘贴时漏掉。public class FlinkCDCDataStreamTest { public static void main(String[] args) throws Exception { // 1. 流处理环境并行度先给 1方便对日志 StreamExecutionEnvironment env StreamExecutionEnvironment.getExecutionEnvironment(); env.setParallelism(1); // 2. 检查点CDC 的 binlog 位点存在状态里不开 CK 就没有断点续传 env.enableCheckpointing(3000L, CheckpointingMode.EXACTLY_ONCE); env.getCheckpointConfig().setCheckpointTimeout(60 * 1000L); env.getCheckpointConfig().setMinPauseBetweenCheckpoints(3000L); // 任务取消后保留最后一次 CK否则重启只能从头快照 env.getCheckpointConfig().enableExternalizedCheckpoints( CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION); // 一分钟内失败超过 3 次就彻底失败避免无限重启刷日志 env.setRestartStrategy(RestartStrategies.failureRateRestart( 3, Time.days(1L), Time.minutes(1L))); env.setStateBackend(new HashMapStateBackend()); // 状态目录必须放共享存储本地路径在容器化部署下重启就没了 env.getCheckpointConfig().setCheckpointStorage(hdfs://node01:8020/flink-cdc/ck); System.setProperty(HADOOP_USER_NAME, hadoop); // 3. CDC Source MySqlSourceString mySqlSource MySqlSource.Stringbuilder() .hostname(node01) .port(3306) .databaseList(test) .tableList(test.t1) .username(cdc_user) .password(Cdc2024) .deserializer(new JsonDebeziumDeserializationSchema()) .startupOptions(StartupOptions.initial()) .build(); // 4. 读取并打印 DataStreamSourceString mysqlDS env.fromSource( mySqlSource, WatermarkStrategy.noWatermarks(), MySQLSource); mysqlDS.print(); // 5. 执行 env.execute(flink-cdc-datastream-test); } }参数这一块值得逐条过一遍因为每个都对应一类线上事故。databaseList和tableList是两层过滤前者决定连哪些库后者精确到表格式是库名.表名支持正则如test\.t.*。只写databaseList(test)不写tableList会把整个库的表全抓下来表多的时候初始化能跑几个小时。startupOptions有五种取值行为差别很大。initial()是首次启动先做全量快照再接着读增量最常用earliest()不做快照从 binlog 最早的位置读前提是 binlog 没被清理过否则直接报错latest()不做快照只读连接建立之后的新变更适合做增量对齐specificOffset()和timestamp()用于从指定位点或时间点恢复做数据修复时用得上。deserializer决定输出格式。JsonDebeziumDeserializationSchema输出的是 Debezium 风格 JSON里面op字段标了操作类型r是快照读、c是插入、u是更新、d是删除。before和after分别是变更前后的行镜像删除操作只有before插入只有after。第一次看这个 JSON 会觉得字段多得离谱其实有用的就这几个。检查点里的setCheckpointStorage是最容易在测试环境被写成本地路径的一项。本地文件系统在单机 Standalone 上能跑一旦换成 Kubernetes 或者会话集群JobManager 重启换了机器状态目录就找不到了作业只能从空状态起于是重复消费一遍全量。测试环境图省事用本地路径上线前必须换成 HDFS 或对象存储。并行度这里给 1是为了输出顺序稳定、日志好对。真上生产CDC Source 的并行度取决于表的数量和主键分布initial快照阶段会按主键切 chunk 并行读增量阶段每个并行子任务各挂一个 binlog 连接并行度开太大反而会给 MySQL 造成连接压力。打包上传提交# 打包产物在 target 下 mvn clean package -DskipTests # 提交到会话集群-s 指定 savepoint 路径即可从位点恢复 bin/flink run \ -m node01:8081 \ -c com.example.cdc.FlinkCDCDataStreamTest \ ./flink-cdc-test.jar提交后在 TaskManager 日志里能看到全量快照阶段的输出op全是r。等快照跑完往test.t1里插一条日志里立刻出现opc的记录。到这一步说明 binlog 通了、权限够了、反序列化也正常。4. 从 FlinkSQL 到 CDC 3.0 Pipeline把 MySQL 整库搬到 Doris4.1 FlinkSQL 建表最省事的入门口子不想写 Java 的话FlinkSQL 的 CDC 连接器是最快的验证方式。核心就是一条CREATE TABLE把 MySQL 表映射成 Flink 表之后当普通表查。CREATE TABLE t1 ( id STRING, name STRING, PRIMARY KEY (id) NOT ENFORCED ) WITH ( connector mysql-cdc, hostname node01, port 3306, username cdc_user, password Cdc2024, database-name test, table-name t1 );PRIMARY KEY ... NOT ENFORCED这句必须写但 Flink 不会真的去校验唯一性它只是告诉优化器这列可以做 upsert 的 key。漏写的话下游 JDBC sink 或者 Doris sink 会退化成 append 模式更新就变成重复插入Doris 里同一个 id 出现好几行。这个坑我在早期项目里踩过排查了半天以为是 CDC 重复消费其实是建表语句少了一行。字段类型映射也别大意。MySQL 的TINYINT(1)在 Flink 里映射成BOOLEAN到了 Doris 又变成BOOLEAN如果下游按数值处理就会出问题DECIMAL的精度要显式声明不写默认是DECIMAL(10,0)小数部分直接被截断DATETIME和TIMESTAMP在 Flink 里是两种类型前者无时区、后者带时区跨时区同步时对不上八小时往往就出在这里。FlinkSQL 跑起来最直接# 交互式客户端里直接粘贴上面的 DDL再查一下 SELECT * FROM t1;它会先做一遍全量然后把增量变更持续打出来。适合快速验证也适合在 FlinkSQL 任务里再接一层聚合、双流 join。生产上表一多DDL 写起来就烦了这时候该上 Pipeline。4.2 CDC 3.0 Pipeline连接器放置与目录约定Pipeline 模式不需要写一行 Java但要准备好安装包和连接器 jar。把flink-cdc-3.0.0-bin.tar.gz解到/opt/module/然后确认两件事MySQL 和 Doris 的 pipeline 连接器 jar 在lib目录下作业描述文件放在任意目录习惯上建个job/。tar -zxvf flink-cdc-3.0.0-bin.tar.gz -C /opt/module/ cd /opt/module/flink-cdc-3.0.0 # 两个连接器 jar 必须在 lib 下缺一个启动就报 connector not found ls lib/ | grep pipeline-connector # flink-cdc-pipeline-connector-doris-3.0.0.jar # flink-cdc-pipeline-connector-mysql-3.0.0.jar mkdir -p jobFlink 集群侧还要开检查点Pipeline 依赖它做位点恢复。改conf/flink-conf.yaml改完分发到所有节点再启动集群。# conf/flink-conf.yaml execution.checkpointing.interval: 5000 state.backend: hashmap state.checkpoints.dir: hdfs://node01:8020/flink-cdc/ck execution.checkpointing.externalized-checkpoint-retention: RETAIN_ON_CANCELLATIONexecution.checkpointing.interval不给的话作业能跑起来也能同步数据但一旦重启就从最开始重来全量快照再跑一遍。这属于能跑但不对的那类配置测试时不容易暴露上线后宕一次机才发现。4.3 mysql-to-doris.yaml 逐项拆解整份配置就三段source、sink、pipeline。每段里的字段不多但每个都影响行为。source: type: mysql hostname: node01 port: 3306 username: cdc_user password: Cdc2024 tables: test.\.* # 正则test 库下所有表 server-id: 5400-5404 # 区间框架自动分配避免多任务冲突 server-time-zone: UTC8 sink: type: doris fenodes: node01:7030 # FE 的 HTTP 端口 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: 1tables用的是正则而不是精确表名test.\.*表示test库下全部表。注意 YAML 里反斜杠在无引号标量中会原样保留写成test\\.\\*反而会报错这点和 Java 字符串不一样。server-id写成区间是 Pipeline 模式相对 DataStream 的一个改进。Flink 内部会按并行度从区间里取不同的 id 给每个读取子任务避免一个作业内部自己跟自己冲突。但也意味着区间要留够parallelism是 4 就得至少 4 个 id。server-time-zone: UTC8决定了 Debezium 解析TIMESTAMP时按哪个时区解释。不配的话默认 UTC同步到 Doris 的DATETIME字段会整体差八小时而且这个偏差很隐蔽——全量阶段可能看不出来增量阶段一对比才发现。sink 侧的fenodes是 Doris FE 的 HTTP 端口默认 8030资料里改成 7030。注意不是 90309030 是 MySQL 协议端口填错了连接会超时报错信息也不直观。table.create.properties.replication_num: 1对单 BE 的测试环境是必须的。Doris 建表默认副本数是 3只有一个 BE 的时候建表直接失败报Failed to find enough host with storage medium and tag。这个错看着像磁盘问题实际是副本数配多了。light_schema_change: true打开轻量级 schema change这是自动建表时带上的属性。它的作用是让后续加列这类变更不用重刷全量数据配合 CDC 的 schema evolution 才能真正做到MySQL 加一列Doris 自动跟上。4.4 route 段库表改名和整库同步默认情况下源库表名和目标库表名一致落到 Doris 的test库。但test_route库也有同名表两边直接同步就会撞。这时加 route 段做改名。source: type: mysql hostname: node01 port: 3306 username: cdc_user password: Cdc2024 tables: test_route.\.* server-id: 5400-5404 server-time-zone: UTC8 sink: type: doris fenodes: node01:7030 username: root password: 000000 table.create.properties.light_schema_change: true table.create.properties.replication_num: 1 route: - source-table: test_route\.t1 sink-table: test_route.t1_route - source-table: test_route\.t2 sink-table: test_route.t2_route pipeline: name: Sync test_route to Doris parallelism: 1source-table写源端的库表正则sink-table写目标端的库表名两边都可以用正则捕获组。表多的时候用一条规则批量改source-table: test_route\.(.*)配sink-table: test_route.dwd_$1把所有表统一加前缀。这个写法在做库级隔离时很省事不用一张一张列。路由规则是从上往下匹配第一条命中就停。规则写多了要注意顺序宽泛的正则放后面否则后面的精确规则永远不生效。4.5 启动、验证与增量观察Doris 侧先把 FE 和 BE 拉起来用 MySQL 客户端连 9030 端口建库。# 启动 Doris bin/start_fe.sh --daemon bin/start_be.sh --daemon # 用 MySQL 协议连进去注意端口是 9030 mysql -uroot -p000000 -P9030 -hnode01CREATE DATABASE test; CREATE DATABASE test_route; SHOW DATABASES;然后提交 Pipeline 作业。cd /opt/module/flink-cdc-3.0.0 bin/flink-cdc.sh job/mysql-to-doris.yaml命令会打印作业 ID 和 Flink WebUI 地址。到 WebUI 上看作业拓扑是一个 source 接一个 sink并行度取自 YAML 里的parallelism。全量阶段能看到 source 端有快照读取的记录数跑完后切到增量。验证分三步一步都别省。第一步全量对账。在 Doris 里SELECT COUNT(*) FROM test.t1和 MySQL 侧的基线数字比。不一致的话先看是不是表里有大字段或者特殊字符导致解析失败sink 侧的脏数据记录一般会打 warn 日志。第二步增量插入。往test.t1插一条新数据等两三秒Doris 里查一下。能查到说明整条链路通了。第三步更新和删除。UPDATE test.t1 SET namexxx WHERE id1001看 Doris 里是不是真的改了而不是多出一行DELETE FROM test.t1 WHERE id1003看 Doris 里是不是也没了。删除这一条最能暴露表模型问题——如果 Doris 侧建的是 Duplicate 模型删除就变成了插入一条带删除标记的记录查询时还能看见。第四步加列。ALTER TABLE test.t1 ADD COLUMN age INT然后往新列写值。开了 schema evolution 的话Doris 侧会自动加列并同步数据。这一步是 3.0 相对 2.x 最实用的改进也是资料里特意提到要观察的部分。5. 避坑与排查五条血泪记录和一张上线自检表5.1 五条踩过的坑按现象、原因、解决来记现象一作业启动就报The MySQL server is not configured to use a ROW binlog_format或者能连上但收不到任何更新。原因基本锁定在binlog_format不是ROW或者log-bin压根没开再或者binlog-do-db没把目标库列进去。还有一种隐蔽情况配置改在[mysqld]之外的小节里比如放在[mysqld_safe]下MySQL 读到了但不生效。解决顺序是先SHOW VARIABLES LIKE binlog%确认运行时值再回头查配置文件的位置和段落。8.0 支持SET GLOBAL binlog_format ROW临时切但已经写出去的 STATEMENT 格式日志没法补测试环境可以这么应急生产必须改配置文件重启。现象二第二个 CDC 任务提交后第一个任务断开重连日志里反复出现A slave with the same server_uuid/server_id as this slave has connected。原因是两个作业用了同一个server-id。DataStream 写法里这个 id 由 Debezium 自动生成或者从连接参数里取Pipeline 写法里从 YAML 的区间分配。多个人在同一套库上各跑各的任务最容易撞。解决是规划好 id 段比如测试环境用 5400-5499预发用 5500-5599每个作业分一个小区间。Pipeline 的server-id写区间而不是固定值就是为了让框架自己错开。现象三作业重启后数据重复Doris 里同一个主键出现多行或者整库数据被重放了一遍。没开检查点或者检查点目录配成了本地路径。CDC 的位点是存在 Flink 状态里的状态丢了就等于失忆只能按startupOptions重新来一遍。initial()会重跑全量下游如果没有幂等去重直接翻倍。解决是把 checkpoint 间隔压到 5 秒以内状态后端和存储目录放到共享存储任务取消时RETAIN_ON_CANCELLATION。另外 Doris 侧建表一定用 Unique Key 模型主键相同自动覆盖这是最后一道后悔药。现象四Doris 建表失败报Failed to find enough host with storage medium and tag或者replication num should be less than the number of available BE。单 BE 环境下replication_num默认 3凑不齐副本。这个错和磁盘、标签都没关系纯粹是副本数配多了。解决是在 sink 段加table.create.properties.replication_num: 1或者再起两个 BE。生产环境三副本没问题测试环境别照抄生产配置。现象五时间字段整体差八小时或者DATETIME值变成1970-01-01。差八小时是server-time-zone没配或者配错Debezium 默认按 UTC 解析。变成 1970 一般是源表里的零值日期0000-00-00 00:00:00在转换时溢出或者字段类型映射成了DATE而实际存的是DATETIME。解决是 source 段固定写server-time-zone: UTC8源表尽量避免零值日期类型映射上MySQL 的DATETIME对应 Flink 的TIMESTAMP(3)对应 Doris 的DATETIME别图省事全写成STRING后面做时间聚合会哭。5.2 上线前照着过一遍的自检表检查项判断标准不通过的后果binlog_formatSHOW VARIABLES返回 ROW收不到行级变更binlog 保留时长大于全量初始化最坏耗时中途失败无法从位点恢复源表主键每张表都有显式主键无法并行快照Doris 建不了 Unique 模型checkpoint 间隔≤ 5 秒存储为共享路径重启重放全量server-id 区间各作业不重叠长度够并行度任务互相踢下线Doris 副本数单 BE 为 1多 BE 为 3建表直接失败表模型Unique Key 主键更新变重复插入时区server-time-zone: UTC8时间字段差八小时这张表我一般会贴在值班文档里每次上新链路前对着勾一遍比事后翻日志省事。5.3 出问题先看哪几个地方Flink WebUI 是第一个要看的地方不是日志。先看作业是不是 RUNNING再看 checkpoint 那一栏的最近一次成功时间和大小。checkpoint 一直失败的话后面所有数据问题都别分析了先把状态存储的问题解决掉。第二个看 source 端的指标numRecordsOut是否持续增长currentFetchEventTimeLag这类延迟指标是不是在合理范围。延迟持续上涨说明下游 sink 扛不住Doris 的导入任务在排队。第三个才是 TaskManager 日志按ERROR和WARN过滤重点看debezium和doris关键字。CDC 的日志噪音很大全量阶段满屏都是快照进度真正有用的报错往往夹在中间。Doris 侧看SHOW LOAD WHERE LABEL LIKE %cdc%导入失败的话这里能看到具体原因比如字段类型不匹配、分区不存在、内存超限。很多数据没同步过来的问题最后发现是 Doris 导入任务失败Flink 侧正常。6. 进阶Savepoint 换版本和一次能跑的对账 SQL作业写完能跑只是开始真正难的是怎么在不丢数据的前提下换版本、改逻辑、加表。CDC 作业有个天然优势位点存在状态里只要 Savepoint 在就能从断点续上不用重跑全量。换版本的标准动作是这样先在 WebUI 里触发 Savepoint等它成功别用 cancel 代替cancel 可能不触发保存确认 HDFS 上出现了savepoint-xxxx目录用新 jar 从 Savepoint 启动观察新作业追平延迟最后再停旧的。中间这几分钟两边会同时消费同一段 binlog所以下游必须是幂等的——Doris 的 Unique Key 模型天然满足这一点同主键后写的覆盖先写的不会出现双份。# 触发 savepoint记下返回的路径 bin/flink savepoint jobId hdfs://node01:8020/flink-cdc/savepoint # 从 savepoint 恢复换成新打出来的 jar bin/flink run \ -s hdfs://node01:8020/flink-cdc/savepoint/savepoint-xxxx \ -c com.example.cdc.FlinkCDCDataStreamTest \ ./flink-cdc-test-v2.jarPipeline 模式也一样bin/flink-cdc.sh支持从 savepoint 恢复只是参数位置在文档里不太显眼第一次用容易漏。另一个常被问的问题是怎么确认两边真的一致。别只比行数行数对得上但内容错了的情况太多了。我一般跑这个对账 SQL在 MySQL 和 Doris 上分别算一遍主键的哈希聚合两边比对。-- 源端和目标端各跑一次结果应完全一致 SELECT COUNT(*) AS cnt, SUM(CRC32(CONCAT_WS(|, id, IFNULL(name,)))) AS sig FROM test.t1;CRC32够快碰撞概率在这个量级可以接受字段多的话换成MD5或者SHA2慢一点但更稳。这个查询我从两边各跑一次cnt和sig都对上才算过关。差异定位时按主键区间二分比逐行 diff 快得多。还有个小技巧给 CDC 作业单独配一个告警盯currentFetchEventTimeLag这个指标超过 60 秒就报。延迟涨上去往往不是 CDC 的问题而是 Doris 那边导入排队或者 BE 磁盘满了早发现比等业务方来问强。这条是我踩过一次坑之后加的——那次 Doris 的 BE 挂了两个CDC 作业表面上还在 RUNNING数据全堵在内存里谁也没发现直到第二天报表全空。从那以后我每次上新链路都强制走一遍「插一条、改一条、删一条、加一列、重启一次」这五步少一步都不放心。希望帮到你。本文还有配套的精品资源点击获取