简介面向具备一定 Flink 基础的大数据研发人员这份 docx 文档围绕 Flink CDC 3.0 展开聚焦 MySQL 数据实时捕获并同步至 Doris 的完整链路帮助读者理解变更数据捕获原理并落地 Streaming ETL 场景。内容从 CDC 概念切入对比基于查询与基于 Binlog 两种方式在延迟、全量捕获和数据库压力上的差异并结合 flink-cdc-connectors 说明 MySQL、PostgreSQL 等数据源的读取方式。包内共 1 个 docx 文件压缩包约 145KB文档给出数据准备、DataStream 与 Flink SQL 应用、MySQL 到 Doris 的环境搭建、配置编写、任务启动和测试步骤并提示开启 Binlog、设置检查点等关键细节。已有 1789 人学习适合需要掌握部署流程、保证同步一致性与稳定性的工程师参考实践。1. 从一张凌晨还在跑的对账表说起对账报表凌晨两点还在跑批量任务业务方早上八点就要看昨天下午的订单明细——这种需求下用 Flink CDC 3.0 把 MySQL 的变更实时同步到 Doris 做 OLAP 查询基本绕不过去。它直接读 MySQL 的 binlog把 insert/update/delete 按发生顺序变成一条流经过 Flink 加工后写进 Doris能做到秒级可见。适合谁手里有 Flink 基础、要搭一条 MySQL 到 Doris 实时同步链路的后端或大数据开发。这份资料从 CDC 概念讲到 DataStream、FlinkSQL 两种写法再落到 3.0 的 Pipeline yaml 配置属于能照着敲完就跑起来的那种。2. 架构与选型为什么是 Binlog Flink Doris 这条链路动手写代码前得先想清楚为什么不直接用 DataX 定时抽一把。选型错了后面调参数全是白费力气。2.1 基于查询和基于 Binlog差的不是一点延迟CDC 这个概念本身不复杂核心是监测数据库的变动把插入、更新、删除按发生顺序完整记下来写到消息中间件或直接投给下游。难的是“怎么拿到这些变动”业内只有两条路。一条是基于查询定时跑一条select * from t where update_time ?靠时间戳或自增 ID 捞增量。Sqoop、DataX 走的就是这条路执行模式是 Batch也就是一批一批来。它的毛病很直接物理删除捞不到一条记录在两次查询之间改了三次你只能看到最后一次中间态全丢了为了不漏数据查询间隔不敢拉太长库的压力就上来了。另一条是基于 Binlog订阅 MySQL 的二进制日志数据库写什么你就收到什么。Canal、Maxwell、Debezium 都是这个路子执行模式是 Streaming延迟低删除操作天然带op d标记不会丢。代价是要开 binlog、要处理位点、要占一个 server-id。维度基于查询的 CDC基于 Binlog 的 CDC代表产品Sqoop、DataXCanal、Maxwell、Debezium执行模式BatchStreaming能否捕获全部变更否删除和中间态会丢是变化延迟高取决于调度周期低接近实时对源库压力有每次查询都是真实读小读 binlog 不碰业务表Flink CDC 属于后者。Flink 社区把 Debezium 嵌进了 Source 组件里flink-cdc-connectors可以直接从 MySQL、PostgreSQL 读全量数据和增量变更全量阶段按主键分片并行扫扫完自动切到 binlog 阶段接着读这个“无锁快照 增量衔接”是整个组件最值钱的地方。提示全量阶段默认会对表加轻量级锁来拿一致性位点大表上要留意snapshot.select.statement.overrides和锁等待超时别在业务高峰期第一次起任务。2.2 Flink CDC 3.0 的三条落地路径同一套 connectorFlink CDC 提供了三种写法很多人一开始就选错写到一半发现路走不通再返工。第一种是DataStream API。手工MySqlSource.builder()构造 Sourceenv.fromSource()接进流里后面能接任意算子做过滤、打宽、聚合。自由度最高但整库同步几十张表的时候你得自己写循环、自己拼 sink代码量上去了。第二种是FlinkSQL。一句CREATE TABLE ... WITH (connector mysql-cdc, ...)就把 Source 声明出来了配合tableEnv.sqlQuery或者直接INSERT INTO写目标端。适合字段映射规整、逻辑能用 SQL 表达的场景改起来最快。第三种是 3.0 才正式成型的Pipeline。不写 Java 代码写一份 yaml描述 source、sink、pipeline 三段用bin/flink-cdc.sh提交。整库同步、表路由、加列同步这些原本要写代码的事配置里两三行搞定。这也是这份资料里 MySQL 到 Doris 那一节用的方式。方式代码量适合场景改造灵活度DataStream大需要自定义算子、复杂清洗最高FlinkSQL小字段映射规整、逻辑可 SQL 化中等Pipeline yaml几乎为零整库同步、表级路由、快速上线靠配置和自定义插件我一般的判断标准是只做搬运选 Pipeline搬运途中要洗数据选 FlinkSQL要接自定义 UDF、要和多条流 join老老实实写 DataStream。2.3 端到端链路里每个组件在干什么把链路拆开看一共四段。MySQL 侧负责产生 binlogbinlog_format必须是row否则 Debezium 解析不出变更前后的完整行Flink CDC Source 负责把 binlog event 反序列化成 Flink 的内部 RowData全量阶段和增量阶段在同一个 Source 里完成切换位点信息以状态形式存进 Checkpoint中间的 Flink 算子做清洗和路由最后 Doris Sink 把数据以 Stream Load 的方式批量灌进去。Doris 这一端有个前提容易被忽略目标表必须是唯一键模型Unique Key才能把op u的更新和op d的删除变成 upsert 和 delete。如果建成明细模型同一条记录改十次你就得到十行业务方一对账就发现数字翻倍。建表时UNIQUE KEY(id)加上DISTRIBUTED BY HASH(id) BUCKETS 1小表这么写就够了大表按数据量调 bucket 数。还有一层是状态存储。Flink CDC 靠 Checkpoint 记住“读到哪个 binlog 文件的哪个位置”所以必须配一个可靠的 Checkpoint 存储HDFS 或对象存储都行。这份资料里用的是 HDFS 路径恢复的时候从 Savepoint 起能接着上次的位置读不重不漏。3. 环境与数据准备Binlog、依赖与工程骨架环境这一步最容易翻车因为报错信息往往和真实原因隔了十万八千里。按顺序来一步别跳。3.1 建库建表与造数先准备两个库test做基础同步验证test_route专门用来验证表路由。每个库里三张表t1、t2结构一致t3故意换个字段名这样在路由和字段映射出问题时能第一时间发现。-- 源库基础同步验证 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) ); -- 故意用不同字段验证 schema 是否被正确识别 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);路由验证库同理建一个test_route表结构照抄但t2里塞几条 ID 不同的数据比如 1004、1005、1006。这么做的目的是同步跑起来后你看 Doris 里test.t2和doris_test_route.t2的行数不一样就能确认路由规则真的生效了而不是所有表都被灌进了同一个库。注意主键用 VARCHAR 而不是自增 BIGINT是为了模拟业务里常见的字符串主键场景。Flink CDC 全量阶段按主键切分 chunk字符串主键也能切但排序和切分效率不如整型大表上建议还是用整型或雪花 ID。3.2 开启 Binlog 的三个必改参数MySQL 8 默认binlog_format是 ROW但log-bin和server-id不一定开。改配置文件加完重启。# /etc/my.cnf server-id 1 log-bin mysql-bin binlog_format row # 只对指定库产生 binlog降低 binlog 体积 binlog-do-db test binlog-do-db test_routeserver-id是给复制协议用的标识CDC 客户端本质上就是一个伪装成 Slave 的消费者没有它连不上。log-bin决定 binlog 文件前缀和是否启用。binlog_format row是硬性要求mixed 和 statement 都会让 Debezium 拿不到变更前后的完整行镜像。binlog-do-db是可选优化不写就是全库记录。改完重启登进去确认一下mysql -uroot -p -e show variables like binlog_format; show variables like log_bin;两个值分别是ROW和ON才算过关。另外账号要有REPLICATION SLAVE、REPLICATION CLIENT、SELECT三个权限缺一个都会在连接阶段被拒。3.3 Maven 依赖与打包插件工程用 Flink 1.18.0 打底CDC connector 用 3.0.0JDK 8。依赖分三组Flink 运行时、Table API、CDC connector外加 MySQL 驱动。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 运行时本地跑和数据流 API 都要 -- dependency groupIdorg.apache.flink/groupId artifactIdflink-streaming-java/artifactId version${flink-version}/version /dependency !-- 提交到集群时提供 Client 能力 -- dependency groupIdorg.apache.flink/groupId artifactIdflink-clients/artifactId version${flink-version}/version /dependency !-- FlinkSQL 路线需要这一组缺一个都会在 createTable 时报 NoMatchingTableFactory -- 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 核心连接器 -- 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把所有依赖打进一个 fat jar。集群提交时只传这一个文件省得在lib/目录里到处塞包。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这里有个血泪经验scope千万别给 Flink 那几个包加provided。本地 IDE 跑的时候不会报错打成 fat jar 上集群就抛ClassNotFoundException因为集群自带的版本和你编译时用的可能不一致。用 assembly 全打进去虽然 jar 大几十兆但能省掉一整类排查。提示3.0 之后官方发行版逐步把包名迁到了org.apache.flink.cdc.*命名空间。这份资料的代码用的是com.ververica.cdc.*如果你手上下的是 Apache 命名空间的包import 要整体换掉两套混着引会在运行时报方法签名不匹配编译器还查不出来。4. 核心链路拆解DataStream、FlinkSQL 与 3.0 Pipeline三种写法逐个过一遍重点是搞清每个参数改了什么、不改会怎样。4.1 DataStream 主程序从环境到 Source这段代码是整个链路的地基Checkpoint 和 Source 两块的配置尤其不能省。public class FlinkCDCDataStreamTest { public static void main(String[] args) throws Exception { // 1. 流处理环境并行度先给 1便于观察日志顺序 StreamExecutionEnvironment env StreamExecutionEnvironment.getExecutionEnvironment(); env.setParallelism(1); // 2. Checkpoint 七件套缺一项都可能影响断点续传 env.enableCheckpointing(3000L, CheckpointingMode.EXACTLY_ONCE); env.getCheckpointConfig().setCheckpointTimeout(60 * 1000L); env.getCheckpointConfig().setMinPauseBetweenCheckpoints(3000L); env.getCheckpointConfig().enableExternalizedCheckpoints( CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION); env.setRestartStrategy(RestartStrategies.failureRateRestart( 3, Time.days(1L), Time.minutes(1L))); env.setStateBackend(new HashMapStateBackend()); env.getCheckpointConfig().setCheckpointStorage(hdfs://node01:8020/flinkCDC); // 写 HDFS 需要用户名否则 Permission denied System.setProperty(HADOOP_USER_NAME, flink); // 3. 构造 MySQL CDC Source MySqlSourceString mySqlSource MySqlSource.Stringbuilder() .hostname(node03) .port(3306) .databaseList(test) .tableList(test.t1) .username(root) .password(000000) .deserializer(new JsonDebeziumDeserializationSchema()) .startupOptions(StartupOptions.initial()) .build(); // 4. 接入流并打印 DataStreamSourceString mysqlDS env.fromSource( mySqlSource, WatermarkStrategy.noWatermarks(), MySQLSource); mysqlDS.print(); env.execute(); } }逐项说。enableCheckpointing(3000L, EXACTLY_ONCE)里的 3000 是间隔毫秒数意思是每 3 秒做一次快照生产上一般给 30 秒到 1 分钟本地验证给 3 秒是为了快速看到效果。EXACTLY_ONCE保证恢复时不重复处理代价是两阶段提交Doris 侧要支持事务性写入用 Stream Load 的话靠 label 去重。setCheckpointTimeout给 60 秒大状态快照慢的时候不至于被误判超时。RETAIN_ON_CANCELLATION很关键任务手动 cancel 时保留最后一次 Checkpoint否则你点一下停止位点就没了再起来只能从头全量扫。databaseList和tableList支持正则写test表示整个库写test.t1表示精确到表。想一次抓多张表可以写test.t1|test.t2或者直接test\..*。JsonDebeziumDeserializationSchema把 SourceRecord 转成 JSON 字符串输出里带before、after、op、ts_ms四个关键字段调试阶段直接 print 最直观。4.2 启动模式的五个取值选错就是全量重扫StartupOptions是 Flink CDC 里最容易被忽略、又最容易出事的一个参数。// initial首次启动先做全量快照然后接着读最新 binlog最常用 .startupOptions(StartupOptions.initial()) // earliest不做快照直接从 binlog 最早的位置读要求 binlog 完整保留 // .startupOptions(StartupOptions.earliest()) // latest不做快照只读任务启动之后的变更历史数据完全没有 // .startupOptions(StartupOptions.latest()) // specificOffset从指定的 binlog 文件位点开始 // .startupOptions(StartupOptions.specificOffset(mysql-bin.000003, 154)) // timestamp从指定时间戳开始会从头遍历 binlog 跳过更早的事件 // .startupOptions(StartupOptions.timestamp(1690000000000L))initial是绝大多数场景的答案第一次跑把存量数据全捞过来捞完自动切增量。latest只在“目标端已经有存量数据、只想要增量”时用用错了会发现 Doris 里空空如也。earliest最危险它假设 binlog 里有完整历史而 MySQL 的 binlog 默认只保留几天过期就被 purge 了这时候你读到的是残缺数据而且不会报错。timestamp看着好用实际会从头扫 binlogbinlog 体积大的时候启动要等很久。4.3 Checkpoint 与 Savepoint断点续传怎么验证光配了没用得真的验证一遍。这份资料的验证路径很标准值得抄下来当 checklist。# 1. 提交任务 bin/flink run -m node01:8081 -c com.example.cdc.FlinkCDCDataStreamTest ./flink-cdc-test.jar # 2. 观察日志应该先打出一批全量数据的 JSON然后安静下来等 binlog # 3. 手动触发 Savepoint bin/flink savepoint JobId hdfs://node01:8020/flinkCDC/save # 4. WebUI 里 Cancel 任务 # 5. 去 MySQL 里对 t1 做几条 insert / update / delete # 6. 从 Savepoint 恢复 bin/flink run -s hdfs://node01:8020/flinkCDC/save/savepoint-xxxxx \ -c com.example.cdc.FlinkCDCDataStreamTest ./flink-cdc-test.jar恢复后看日志如果只打出了第 5 步那几条变更说明位点恢复成功如果又把 1001 到 1003 全量打了一遍说明 Checkpoint 没生效或者状态后端不可写。第二种情况我遇到过原因是 HDFS 路径的写权限没给对HADOOP_USER_NAME设了个不存在的用户Checkpoint 静默失败任务照跑日志里只有一行 warning。注意enableExternalizedCheckpoints不做设置的话任务 cancel 时 Checkpoint 会被清掉。这个坑我在测试环境踩过一次改完参数忘了重启任务直接 cancel位点没了只能从头再扫一遍全量。4.4 FlinkSQL 写法一行 DDL 换来什么SQL 路线把 Source 声明成一张表后面接标准 SQL。StreamExecutionEnvironment env StreamExecutionEnvironment.getExecutionEnvironment(); env.setParallelism(1); StreamTableEnvironment tableEnv StreamTableEnvironment.create(env); // 声明 CDC Source 表字段名要和 MySQL 对上 tableEnv.executeSql( CREATE TABLE t1 (\n id STRING PRIMARY KEY NOT ENFORCED,\n name STRING\n ) WITH (\n connector mysql-cdc,\n hostname node03,\n port 3306,\n username root,\n password 000000,\n database-name test,\n table-name t1\n )); // 查询并打印结果表会持续输出 changelog tableEnv.sqlQuery(SELECT * FROM t1).execute().print();PRIMARY KEY NOT ENFORCED是必须的FlinkSQL 靠它判断这是 upsert 流而不是 append 流下游做聚合时行为完全不一样。connector mysql-cdc是工厂标识Flink 靠它找到对应的 TableFactory。和 DataStream 相比SQL 版本少了 Checkpoint 配置本地跑没问题上集群一定要在外层补上否则一样没法续传。两种写法的取舍很实际字段要做CASE WHEN、要做时间和字符串转换、要和维表 joinSQL 写着舒服要接自定义反序列化、要把一条变更拆成多条下游消息、要动态决定写哪张目标表只能上 DataStream。4.5 3.0 Pipeline一份 yaml 搞定整库同步3.0 最大的变化在这儿。装好flink-cdc-3.0.0-bin之后把flink-cdc-pipeline-connector-mysql-3.0.0.jar和flink-cdc-pipeline-connector-doris-3.0.0.jar扔进lib/目录然后写配置。source: type: mysql hostname: node01 port: 3306 username: root password: 123456 tables: test.\.* # 正则匹配整库注意转义点号 server-id: 5400-5404 # 给个区间多个任务不会撞车 server-time-zone: UTC8 # 不写会出现时间差 8 小时 sink: type: doris fenodes: node01:7030 # 按实际 FE 的 http_port 填 username: root password: 000000 table.create.properties.light_schema_change: true table.create.properties.replication_num: 1 # 单 BE 测试环境必须给 1 pipeline: name: Sync MySQL Database to Doris parallelism: 1source段的tables支持正则test.\.*表示test库下所有表。server-id写成区间Flink 会从中分配多个同步任务并存时不至于互相踢下线。server-time-zone不配的后果是 datetime 字段整体偏移这个坑很隐蔽因为数据量看着对得上只有对账时才发现时间戳全错。sink段的light_schema_change开启后源端加列能自动同步到 Doris不用手工改表replication_num在单节点环境必须写 1默认是 3副本数不够会建表失败。提交和原生 Flink 任务不一样用的是 CDC 自带的脚本# Flink 侧先开 Checkpoint3.0 的 Pipeline 依赖它做位点恢复 # conf/flink-conf.yaml 里加一行 # execution.checkpointing.interval: 5000 bin/flink-cdc.sh job/mysql-to-doris.yaml跑起来之后去 Doris 里show tables应该能看到和 MySQL 同名的库表。然后回 MySQL 对几张表做增删改再刷新 Doris数据应该同步过来。这一步验证通过说明整条链路是通的。路由变更的配置思路一样只是加一段routeroute: - source-table: test_route\.t1 sink-table: doris_test_route.t1source-table写库名加表名的正则sink-table指定落到哪个库哪张表。整库改名的常见写法是sink-table: doris_route.承接源表名这样几十张表一条规则就迁完了不用一张一张列。5. 避坑与排查那几条让我加班到凌晨的记录下面五条都是我或者身边同事真实撞过的按现象、原因、解决三段写遇到了直接对号入座。5.1 只跑全量不进增量日志停在 snapshot 阶段现象任务启动后哗哗打出一批全量数据然后就没动静了。去 MySQL 里改数据Flink 日志一条新的都没有但任务状态是 RUNNING。原因九成是 binlog 没配对。要么binlog_format还是MIXED要么log-bin没开要么账号缺REPLICATION SLAVE权限。还有一种情况是tableList写的是库名而不是表名全量能扫到增量订阅时匹配不上。解决先show variables like binlog_format和show variables like log_bin确认再show grants for root%看权限。都正常的话把日志级别调到 DEBUG看有没有BinlogSplitReader相关的报错。我遇到过最离谱的一次是 MySQL 重启后 binlog 文件被轮转客户端还拿着旧位点连这时候清掉 Checkpoint 重跑最快。5.2 从 Savepoint 恢复后数据重复或直接重扫全量现象任务 cancel 再恢复Doris 里出现大量重复行或者日志显示又从第一条数据开始扫。原因Checkpoint 没真正落盘。可能是状态后端配置和存储路径不匹配可能是 HDFS 权限不足导致快照静默失败也可能是 cancel 时忘了配RETAIN_ON_CANCELLATION快照被清理掉了。解决先在 WebUI 的 Checkpoints 页签确认最近一次快照的 state 是 COMPLETED再去 HDFS 上看目录里有没有chk-xxx文件夹。恢复时用-s参数指定具体路径别用-s :latest那个在 1.18 上偶尔会解析到已失效的目录。数据重复的话Doris 侧如果是唯一键模型重新灌一遍会自动去重算是兜底。5.3 源端加列之后 Doris 侧建表失败现象Pipeline 跑得好好的开发在 MySQL 上ALTER TABLE ADD COLUMN过一会儿 Doris 侧报 schema change 相关的错同步中断。原因Doris 的 light schema change 默认关闭加列需要走重量级的 schema change耗时长而且和写入冲突。另外如果目标表不是唯一键模型加列行为也不一样。解决配置文件里必须显式写table.create.properties.light_schema_change: true这个参数只对新建表生效已经建好的表要手工改。加列操作建议放在低峰期改完观察 Doris 的SHOW ALTER TABLE COLUMN状态直到 FINISHED。5.4 多个同步任务互相踢日志报 server-id 冲突现象第二个同步任务起来之后第一个任务突然断开重连两个任务反复抢。原因每个 CDC 客户端在 MySQL 眼里都是一个 Slaveserver-id就是身份标识。两个任务用了同一个 IDMySQL 会认为后连的把前一个顶掉。解决配置里用区间比如server-id: 5400-5404Flink 会从中分配不同的值。任务多了就把区间放大别偷懒写死一个数字。这个问题在单任务测试时永远不会暴露一上生产就炸。5.5 时间字段整体差 8 小时现象数据都同步过来了行数也对但 Doris 里的create_time比 MySQL 里晚了 8 小时。原因MySQL 的datetime类型本身不带时区binlog 里存的是服务器本地时间客户端默认按 UTC 解释一进一出就差 8 小时。这个错误不报异常最容易被忽略。解决配置里加server-time-zone: UTC8。已经同步过去的数据改配置也不会自动修正要重新灌一遍。上线前我一般会挑一条带时间戳的记录在两边各查一次做比对比看日志靠谱。6. 二次开发改造从能跑到敢放进生产链路跑通只是第一步。真正上生产有几个地方我基本都会动。换掉 JSON 反序列化。JsonDebeziumDeserializationSchema调试好用生产环境序列化开销不小而且下游还得再解析一遍。常见的做法是自己实现DebeziumDeserializationSchemaT直接输出自定义 POJO 或者 RowData把before、after、op打平成业务需要的结构。public class FlatRowDeserializer implements DebeziumDeserializationSchemaRowData { Override public void deserialize(SourceRecord record, CollectorRowData out) { // 从 record 里取出 Struct按 op 类型分别处理 Struct value (Struct) record.value(); String op value.getString(op); // cinsert, uupdate, ddelete, rsnapshot Struct after value.getStruct(after); // 只保留业务需要的字段按目标端顺序组装成 RowData out.collect(RowData.create(...)); } Override public TypeInformationRowData getProducedType() { return ...; // 和下游算子声明的类型必须严格一致 } }这里的重点是op的四个取值c是插入u是更新d是删除r是全量快照阶段读出来的行。删除事件的after是 null字段要从before里取这点如果写漏了删除操作会直接 NPE。加一层打宽。原始 binlog 是一条一条变更业务侧要的是宽表。常见做法是拿 CDC 流去 join 一张维表用 FlinkSQL 的 temporal join维表用 JDBC connector 或者 HBase 缓存。改造完之后 Pipeline 就用不了了得回到 DataStream 或 SQL 路线。监控指标。Pipeline 模式下能看到的只有 Flink 的吞吐和延迟业务侧感知不到。我一般会在 Sink 前后各挂一个Counter记录处理条数和失败条数暴露到 Prometheus。验证方法是先看总条数是否和源端变更次数对得上再看端到端延迟的 P99 有没有突然抬升。改造点改动位置验证方式自定义反序列化DeserializationSchema 实现类抽样比对 JSON 与 POJO 字段宽表打宽FlinkSQL temporal join维表更新后新数据是否生效监控埋点算子内 Counter / Gauge条数对账 延迟曲线表路由调整yaml 的 route 段目标库表名、行数逐表核对改完别急着发生产先在测试库跑一轮完整的“全量 增删改 加列 恢复”组合拳。我踩过最贵的一次坑是只测了插入没测删除上线第一周业务方发现删掉的订单在报表里还活着排查了半天才发现是op d分支写错了字段来源。从那以后我每次改反序列化逻辑都会强制走一遍包含删除和加列的回归清单哪怕只改了一行。希望帮到你。本文还有配套的精品资源点击获取