企业官网建设流程全解析

简介：面向需要将达梦数据库变更实时接入 Flink 的开发与运维人员，这份 FlinkCDC 达梦日志同步工程资料围绕基于日志的 CDC 实时同步展开，可用于数据仓库同步、实时报表、监控告警等场景，帮助理解连接器配置、变更捕获与流式处理链路。压缩包共 315 个文件、约 341.71MB，以 263 个 jar 依赖为主，并含 xml、properties 配置，java/class 示例及 sql 脚本，另附 docx 说明，便于还原工程结构与调试环境。当前已有 776 人学习下载。读者可从中获得达梦 CDC 的 Java 作业与 Flink SQL 两种实现参考，结合示例类、配置参数和 SQL 文件梳理连接地址、端口、账号及任务启动流程，并借鉴日志解析、插入更新删除捕获和低延迟同步的排错思路，适合作为搭建实时同步链路的对照素材。

1. 一条 UPDATE 同步不过去，问题往往不在 Flink

第一次把 FlinkCDC 达梦数据库 基于日志实时同步接到测试环境，我在机房里坐了三个晚上。作业状态是 RUNNING，checkpoint 每 30 秒翻一页，TaskManager 日志干净得像刚装好的，目标表却一行都不动。查到最后是达梦侧没开归档，连接器拿不到日志，只能干等。这套组合的坑几乎都不在 Flink 本身：版本对位、归档开关、逻辑日志参数、connector 放哪个目录，任何一处错位，表现出来都是“作业活着但不干活”。这篇写给我这种要在一台干净机器上从零跑通达梦日志同步的人，版本怎么选、jar 怎么编、参数怎么配、崩了去哪看，都按实际落地的顺序来。

2. 装前把环境固定下来：Flink 1.18、CDC 3.0 与 DM8 的版本对位

达梦这条链路的版本关系比 MySQL 那条紧得多。达梦的日志解析依赖实例端的逻辑日志格式，connector 依赖 Flink 的 Source API，Flink CDC 3.x 又把 Source/Sink 收进了 Pipeline 体系。三者中任意一个跳版本，常见结果是编译能过、启动能过、运行到读日志那一步才报错，排查成本极高。我的习惯是先把版本写成一张表钉在 README 里，后面所有命令都从这张表取值，不临时升。

2.1 三件套版本对位与选型理由

落地组合是这样的：Flink 1.18.1、Flink CDC 3.0.1、达梦 DM8 8.1.3.x、JDK 11、Maven 3.8.6。选 Flink 1.18 是因为 Flink CDC 3.x 的 Pipeline 语法要求 1.18 起步，1.17 上只能用 2.4.x 的旧 connector 接口，配置写法完全不同；选 CDC 3.0.1 而不是更高的 3.1、3.2，是因为达梦 connector 的社区分支大多还停在 3.0 的 DataStream 接口上，往上跳要自己改EventSource的实现。JDK 用 11，Flink 1.18 对 JDK 8 还兼容但官方已不推荐，JDK 17 上部分序列化框架会出模块访问告警，11 是最稳的中间值。

达梦实例版本务必用SELECT * FROM V$INSTANCE;确认，别信安装包文件名。我遇到过一次现场装的是 8.1.2 早期小版本，逻辑日志里没有RLOG_APPEND_LOGIC全字段记录，connector 解析出来的 update 只有主键，下游永远看不到字段变化，查了两天才定位到实例版本。

2.2 JDK、Maven 与目录规划

目录规划这一步很多人跳过，后面升级必翻车。我的做法是 Flink 装一份、源码放一份、依赖仓库单独一份，三者不交叉，升级时只换 Flink 目录，源码和本地仓库不动。

# 统一基础目录，后面所有路径都从这里派生 export BASE=/opt/realtime mkdir -p ${BASE}/{app,src,repo,data,logs} # JDK 11，解压即用，不污染系统默认 java tar -zxvf jdk-11.0.20_linux-x64_bin.tar.gz -C ${BASE}/app export JAVA_HOME=${BASE}/app/jdk-11.0.20 export PATH=${JAVA_HOME}/bin:$PATH java -version # 必须看到 11.x，8 或 17 后面编译会出怪问题 # Maven 单独装，并把本地仓库指到独立目录，避免和公司私服缓存混在一起 tar -zxvf apache-maven-3.8.6-bin.tar.gz -C ${BASE}/app export MAVEN_HOME=${BASE}/app/apache-maven-3.8.6 export PATH=${MAVEN_HOME}/bin:$PATH # settings.xml 里把 localRepository 改成 ${BASE}/repo mvn -v

JAVA_HOME必须是 JDK 而不是 JRE，否则编译 connector 时tools.jar相关任务会失败。localRepository单独指定这一条看着多余，实际很关键：达梦 connector 依赖的 Debezium 版本和公司其他项目常用版本经常冲突，共用仓库会出现“昨天能编译今天不行”的怪事。环境变量建议写进/etc/profile.d/realtime.sh，因为 Flink 的start-cluster.sh是用非登录 shell 起的，只在当前终端 export 会失效。

# Flink 解压后立刻改三处配置，别等跑起来再改 tar -zxvf flink-1.18.1-bin-scala_2.12.tgz -C ${BASE}/app cd ${BASE}/app/flink-1.18.1 # 1) 绑定 JobManager 地址，standalone 模式下不配会绑到随机网卡 sed -i 's/^# jobmanager.rpc.address.*/jobmanager.rpc.address: 10.0.0.11/' conf/flink-conf.yaml # 2) 托管内存给足，达梦全量阶段会把数据攒在内存里 sed -i 's/^taskmanager.memory.process.size.*/taskmanager.memory.process.size: 4096m/' conf/flink-conf.yaml # 3) 默认并行度先设 1，达梦日志解析按单并行度跑最稳 echo "parallelism.default: 1" >> conf/flink-conf.yaml

taskmanager.memory.process.size给 4G 是底线，全量快照阶段 connector 要在内存里组批，2G 以下很容易在百万行级别的表上 OOM。parallelism.default设 1 不是保守，是达梦的日志位点在单并行度下才能保证有序，多并行度会增加乱序风险，第一次跑通前别动它。

2.3 达梦侧的三个前置开关

Flink 侧装得再对，达梦不开归档就是白干。这一步是所有翻车现场里最高频的，且报错信息极不友好——connector 只会告诉你读不到日志，不会告诉你去开归档。

-- 1) 确认当前归档状态，ARCH_MODE=Y 才算开 SELECT ARCH_MODE, STATUS$ FROM V$DATABASE; -- 2) 开归档，必须按 MOUNT -> ARCHIVELOG -> OPEN 的顺序 ALTER DATABASE MOUNT; ALTER DATABASE ARCHIVELOG; ALTER DATABASE ADD ARCHIVELOG 'DEST=/dm8/arch, TYPE=LOCAL, FILE_SIZE=1024, SPACE_LIMIT=204800'; ALTER DATABASE OPEN; -- 3) 打开逻辑日志追加，0 关闭 / 1 只记主键和变更列 / 2 记全字段 SP_SET_PARA_VALUE(1, 'RLOG_APPEND_LOGIC', 2); -- 4) 回读确认，两个参数都要生效 SELECT PARA_NAME, PARA_VALUE FROM V$DM_INI WHERE PARA_NAME IN ('RLOG_APPEND_LOGIC', 'RLOG_CHECK_SPACE');

FILE_SIZE=1024单位是 MB，SPACE_LIMIT=204800是归档空间上限 200G，这两个值要按写入量估：日志生成速度大约是全量数据变更量的 1.5 倍，归档保留时间必须大于 checkpoint 间隔的两倍，否则作业一旦停超过保留窗口，位点就再也追不回来。RLOG_APPEND_LOGIC设 2 是必须的，设 1 只记主键，下游拿到的 update 事件里没有非主键字段，看起来“同步成功”但数据是错的。这个参数是动态参数，SP_SET_PARA_VALUE第一个参数传 1 立即生效；如果实例版本较老，传 1 不生效就改成 2 并重启实例。

字符集这一步也顺手确认，达梦建库时默认可能是 GB18030，下游 Kafka 或目标库是 UTF-8，中文会乱码：

-- 0=GB18030, 1=UTF-8, 2=EUC-KR SELECT SF_GET_UNICODE_FLAG();

返回 0 的话，connector 侧要显式指定debezium.decimal.handling.mode之外的编码相关参数，或者干脆在建库阶段就用 UTF-8，省掉后面所有转码麻烦。

3. 编译并部署达梦 CDC Connector：从源码包到 lib 目录

Flink CDC 主仓库的 source 列表里有 MySQL、Postgres、Oracle、SQL Server、MongoDB、TiDB、Db2，没有达梦。达梦的连接器在社区和厂商侧单独维护，所以第一步不是pom里加个依赖那么简单，得自己拉源码编译。这一步决定了后面能不能跑起来，也是最容易在依赖上翻车的地方。

3.1 为什么官方仓库里没有，以及怎么拿到能编译的源码

Flink CDC 的 connector 分两类：一类在主仓库里，跟着版本一起发；一类由各数据库厂商或社区维护，接口按 Flink CDC 的Source规范实现，但发版节奏独立。达梦属于后者，常见做法是从社区的flink-connector-dameng-cdc仓库拉取，选与 Flink CDC 3.0 对齐的分支。拉下来第一件事不是编译，是看pom.xml里的flink.version和flink-cdc.version两个属性，必须和集群版本一致，不一致先改这两行。

cd ${BASE}/src # 用 tag 而不是 main 分支，main 上经常跟着未发布的 CDC 版本 git clone -b v3.0.1 https://example.com/flink-connector-dameng-cdc.git cd flink-connector-dameng-cdc # 先核对版本属性，这是编译前唯一必须做的事 grep -n -E "flink.version|flink-cdc.version|debezium.version" pom.xml

flink.version要改成 1.18.1，flink-cdc.version改成 3.0.1，debezium.version保持仓库默认值不动——Debezium 的版本和达梦日志解析逻辑绑定，跟着仓库走比追新安全。改完版本再编译，否则打出来的 jar 在集群上会因为 API 不匹配抛NoSuchMethodError，这类错误堆栈指向 Flink 内部，很容易误判成集群问题。

3.2 Maven 打包与 shade 插件的取舍

打包的核心问题是“依赖要不要一起打进去”。connector 依赖 Debezium 核心包和达梦 JDBC 驱动，这些在集群的lib目录里未必有，全打进 fat-jar 体积会到几十兆，部署时上传慢；不打进去就得手动往lib里补依赖，容易漏。

# 跳测试：集成测试要连真实达梦实例，CI 环境里必然超时 mvn clean package -DskipTests \ -Dcheckstyle.skip=true \ -Drat.skip=true \ -Dmaven.javadoc.skip=true # 看产物体积，正常在 20~40MB 之间 ls -lh target/*.jar

跳checkstyle和rat是省时间，这两个插件在源码包里经常因为许可证头缺失直接失败，跟功能无关。maven.javadoc.skip同理，JDK 11 下生成 javadoc 对中文注释不友好，报错信息还特别长。

pom.xml里 Flink 本体的依赖必须标provided，这是铁律：

<!-- Flink 本体由集群提供，打进 jar 会和集群自带的类冲突 --> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-streaming-java</artifactId> <version>${flink.version}</version> <scope>provided</scope> </dependency> <!-- Debezium 和达梦驱动必须打进去，集群 lib 里没有 --> <dependency> <groupId>io.debezium</groupId> <artifactId>debezium-core</artifactId> <version>${debezium.version}</version> </dependency> <dependency> <groupId>com.dameng</groupId> <artifactId>DmJdbcDriver18</artifactId> <version>8.1.3.62</version> </dependency>

标provided的边界要卡准：Flink 的flink-core、flink-streaming-java、flink-table-common全部provided；Debezium 全系、达梦 JDBC、Jackson 这些必须打进去。判断标准很简单——集群lib目录里已经有的，provided；没有的，打进去。判断错了的表现是ClassNotFoundException或NoSuchMethodError，两者指向的问题完全不同，前者是缺包，后者是版本冲突。

如果打出来的 jar 里 Debezium 没被 shade 进去（体积明显偏小），检查 shade 插件是否被 profile 跳过：

<plugin> <groupId>org.apache.maven.plugins</groupId> <artifactId>maven-shade-plugin</artifactId> <version>3.4.1</version> <executions> <execution> <phase>package</phase> <goals><goal>shade</goal></goals> <configuration> <!-- 服务文件必须合并，否则 Flink 找不到 Factory 的 SPI 注册 --> <transformers> <transformer implementation="org.apache.maven.plugins.shade.resource.ServicesResourceTransformer"/> </transformers> <!-- 排除签名文件，JDK 11 下带签名会抛 SecurityException --> <filters> <filter> <artifact>*:*</artifact> <excludes> <exclude>META-INF/*.SF</exclude> <exclude>META-INF/*.DSA</exclude> <exclude>META-INF/*.RSA</exclude> </excludes> </filter> </filters> </configuration> </execution> </executions> </plugin>

ServicesResourceTransformer这一条最容易被忽略。Flink 的 connector 靠META-INF/services下的 SPI 文件注册工厂类，shade 时如果只保留第一个同名文件，其他 connector 的注册信息就被覆盖，表现出来是“达梦 source 的 type 认不出来”。排除签名文件是 JDK 11 的老问题，不排会在启动时抛Invalid signature file digest。

3.3 放进 lib 目录并重启集群

jar 打好之后放哪里，直接决定作业能不能提交。lib和usrlib的区别要记清：lib是集群启动时加载的，所有作业共享；usrlib是单作业加载的。connector 必须放lib，因为 TaskManager 侧也要能加载到。

cd ${BASE}/app/flink-1.18.1 # 放进 lib，权限跟其他 jar 保持一致 cp ${BASE}/src/flink-connector-dameng-cdc/target/dm-cdc-connector-3.0.1.jar lib/ chmod 644 lib/dm-cdc-connector-3.0.1.jar # standalone 模式必须整套重启，只重启 TaskManager 不生效 bin/stop-cluster.sh bin/start-cluster.sh # 确认 jar 被加载进 classpath ls -l lib/ | grep -i dm # 在启动日志里找 SPI 注册痕迹 grep -i -m 5 "dameng\|dm-cdc" log/flink-*-taskexecutor-*.log

stop-cluster.sh之后要等进程真正退出再启动，中间隔 5 秒以上，否则 TaskManager 的端口可能还没释放，新进程起来会因为BindException直接退出。启动日志里如果搜不到dameng关键字，说明 SPI 文件没进 jar，回到上一步检查ServicesResourceTransformer。

注意：lib目录里不要放两个版本的同一个 connector，Flink 加载时会按文件名排序取第一个，版本冲突的报错信息完全看不出是 jar 重复导致的。

多节点集群里每个 TaskManager 节点的lib都要放同一份 jar，不能只放 JobManager 那台。我见过一次只在提交节点放了 jar，作业能提交、能调度，一到 TaskManager 就抛ClassNotFoundException，日志里还看不到完整堆栈。

4. 写作业：Pipeline YAML 与 DataStream API 两条路

到这一步环境已经齐了，接下来是把同步作业写出来。Flink CDC 3.0 给了两条路：Pipeline YAML 适合整库同步、配置驱动、不用写代码；DataStream API 适合要做字段转换、过滤、多路分流的场景。两条路的 checkpoint 语义一致，差别在可控性。

4.1 Pipeline YAML：整库同步最快的一条路

YAML 的写法在 3.0 定型，source 和 sink 都用type声明连接器，参数平铺。达梦 source 的type取决于 connector 里 SPI 注册的名字，编译完可以在 jar 里确认。

############################################################################## # 达梦 DM8 -> Kafka 整库同步管道 # 提交：bin/flink-cdc.sh dm-orders-pipeline.yaml -c conf/ ############################################################################## source: type: dameng name: DmSource hostname: 10.0.0.21 port: 5236 username: SYSDBA password: 'Dm123456789' # 库名和表名用点号分隔，正则里点号要转义，否则 . 会匹配任意字符 tables: APP_DB\.ORDERS|APP_DB\.ORDER_ITEM # DDL 变更跟随，达梦侧加字段不用重启作业 schema-change.enabled: true # 首次跑用 initial 做全量+增量，追平后改 latest-offset 避免重复全量 scan.startup.mode: initial server-time-zone: Asia/Shanghai sink: type: kafka name: KafkaSink properties.bootstrap.servers: 10.0.0.31:9092 topic: dm_orders_rt # 单并行度写入，保证同一个主键的变更严格有序 partition.strategy: all-to-zero pipeline: name: dm-orders-to-kafka parallelism: 1 local-time-zone: Asia/Shanghai

tables的正则转义是最常被写错的一处，APP_DB.ORDERS里的点号不转义会匹配到APP_DBXORDERS之类的库表，症状是“配了却同步不到”。scan.startup.mode第一次必须用initial，它会把存量数据做一次快照再切增量；切完增量后改成latest-offset，否则每次重启都重新全量，大表上能把作业卡死。schema-change.enabled打开后 DDL 会跟随，但要确认目标端也支持，Kafka 侧只是把 DDL 事件写进去，消费方要自己处理。

parallelism: 1在 Pipeline 里写死，是因为达梦日志解析的分片能力有限，多并行度容易出现同一张表被两个 subtask 重复消费。整库同步场景下这个值不要调大。

4.2 DataStream API：要做字段处理时更可控

需要脱敏、过滤、多表分流到不同 topic 的场景，YAML 就不够用了，得用 DataStream 写。核心是构造DamengSource并开 checkpoint，checkpoint 是达梦位点唯一的持久化手段，不开等于每次重启都从头来。

public class DmOrdersSyncJob { public static void main(String[] args) throws Exception { StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); // 达梦位点只存 checkpoint，间隔必须小于归档保留时间的一半 env.enableCheckpointing(30_000L, CheckpointingMode.EXACTLY_ONCE); env.getCheckpointConfig().setCheckpointStorage("hdfs://nn:8020/flink/ckp/dm-orders"); env.getCheckpointConfig().setMinPauseBetweenCheckpoints(10_000L); env.getCheckpointConfig().setCheckpointTimeout(120_000L); // 取消作业时保留位点，这是唯一的后悔药 env.getCheckpointConfig().setExternalizedCheckpointCleanup( ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION); DamengSource<String> source = DamengSource.<String>builder() .hostname("10.0.0.21") .port(5236) .username("SYSDBA") .password("Dm123456789") .databaseList("APP_DB") .tableList("APP_DB.ORDERS") // 直接出 Debezium 标准 JSON，下游自己解析 op 字段 .deserializer(new JsonDebeziumDeserializationSchema()) .startupOptions(StartupOptions.initial()) .build(); env.fromSource(source, WatermarkStrategy.noWatermarks(), "dm-orders") .filter(json -> !json.contains("\"op\":\"d\"")) // 过滤删除事件，按业务需要取舍 .sinkTo(kafkaSink()) .name("kafka-orders"); env.execute("dm-orders-sync"); } }

enableCheckpointing的 30 秒不是随便定的：checkpoint 间隔乘以 2 必须小于达梦归档的保留窗口，否则作业停机时间一长，位点对应的日志已经被归档清理，恢复时直接抛“日志找不到”。RETAIN_ON_CANCELLATION一定要开，手动取消作业时位点会被一起删掉，下次启动就只能重新全量，大表上这一下就是几小时。WatermarkStrategy.noWatermarks()是因为 CDC 流本身按 LSN 有序，不需要事件时间语义，加了反而增加状态。

4.3 提交参数与并行度的取舍

作业提交时的参数决定了资源占用和稳定性，几个值必须显式给，不能靠默认。

# 提交到达梦同步作业，参数顺序不要乱 bin/flink run \ -d \ -m 10.0.0.11:8081 \ -c com.example.DmOrdersSyncJob \ -p 1 \ -yD taskmanager.memory.process.size=4096m \ -yD execution.checkpointing.interval=30s \ -yD state.backend=rocksdb \ -yD state.checkpoints.dir=hdfs://nn:8020/flink/ckp/dm-orders \ ${BASE}/src/dm-sync/target/dm-sync-1.0.jar

-d是分离模式，不加的话终端一断作业就停。-p 1和 YAML 里的parallelism是同一个含义，达梦 source 保持 1。state.backend=rocksdb在全量阶段会存大量快照状态，默认的 heap backend 在百万行级别就会 OOM，RocksDB 把状态放磁盘，代价是略微慢一点。state.checkpoints.dir必须是所有节点都能访问的共享存储，本地路径在多节点集群上会导致恢复失败。

5. 排查：达梦日志同步最常见的五个翻车现场

这条链路的报错有个共同点——Flink 侧的信息永远指向“读不到数据”，真正的原因在达梦实例或依赖版本上。下面五条是我实际踩过的，按出现频率排。

5.1 五个现场：现象、原因、解决

现场一：作业 RUNNING，checkpoint 正常，目标端零数据。现象是 TaskManager 日志里反复出现“no more records”之类的空转信息，numRecordsIn 一直是 0。原因是达梦没开归档，或者RLOG_APPEND_LOGIC是 0，connector 连上了库但读不到任何变更。解决办法是回到 2.3 节，SELECT ARCH_MODE FROM V$DATABASE确认是 Y，RLOG_APPEND_LOGIC确认是 2，改完参数重启实例再重启作业。这个坑的血泪经验是：connector 不会主动告诉你归档没开，它只会安静地等。

现场二：提交时报ClassNotFoundException: io.debezium.config.Configuration。现象是作业提交阶段就失败，堆栈指向 Debezium 的配置类。原因是打 jar 时 Debezium 被标成了provided，或者 shade 插件被 profile 跳过，jar 体积只有几百 KB。解决是检查pom.xml里 Debezium 依赖的scope必须是默认的compile，打完看体积应该在 20MB 以上，不够就重新打包。

现场三：恢复作业时报“日志位点已失效”或一直卡在 restore。现象是从 checkpoint 恢复时抛异常，或者恢复后不消费新数据。原因是 checkpoint 间隔乘 2 超过了达梦归档保留窗口，停机期间归档被清理，位点对应的日志文件没了。解决办法是临时改scan.startup.mode为initial重新全量一次，然后把归档SPACE_LIMIT调大、checkpoint 间隔调小。根治手段是把归档保留时间和 checkpoint 间隔做成一个固定比例，我一般按 10:1 留。

现场四：目标端字段全是 null，或者 update 事件只有主键。现象是 insert 正常，update 过来的字段值缺失。原因是RLOG_APPEND_LOGIC设成了 1，这个值只记录主键和变更列，没变的列不会出现在日志里。解决是改成 2 并重启实例。另一种可能是建库字符集是 GB18030，中文字段解析成乱码后被当成 null，用SF_GET_UNICODE_FLAG()确认。

现场五：全量阶段卡在某个百分比不动，或者直接 OOM。现象是 snapshot 阶段 numRecordsIn 停在一个数字不再增长，TaskManager 内存曲线一路上涨。原因通常有两个：目标表没有主键，connector 无法做分片，只能单线程全表扫；或者taskmanager.memory.process.size给得太小，全量数据在内存里组批时撑爆。解决是给表加主键（达梦上可以用唯一索引代替），内存给到 4G 以上，超大表用scan.snapshot.fetch.size调小批次。

5.2 日志该去哪看：三层定位法

出问题时的定位顺序是固定的，跳过任何一层都会浪费时间。

# 从一堆日志里精准捞出 connector 相关行 grep -i -E "dameng|dm-cdc|RLOG|LSN" log/flink-*-taskexecutor-*.log | tail -50 # 确认达梦归档日志是否在增长，隔 10 秒对比两次 SELECT FILE_NAME, NEXT_CHANGE# FROM V$RLOG;

Web UI 的 Exceptions 页只显示最外层异常，真正的根因在Caused by链的最后一层。TaskManager 日志里搜LSN能看到 connector 当前解析到哪个位点，和V$RLOG里的日志序号对比，就能判断是“读得慢”还是“根本没读”。这三层走完，绝大多数问题能在十分钟内定位。

6. 装完怎么验证：从一条 UPDATE 到端到端对齐

环境装完、作业跑起来，最怕的是“看起来在同步”。验证不能只看作业状态，要从源头造一条变更，一路跟到目标端。

我的验证流程分四步。第一步，在达梦侧确认位点在推进：连续两次查V$RLOG的NEXT_CHANGE#，值在增长说明日志在写；同时在 Web UI 看 source 的numRecordsIn是否跟着涨。第二步，造一条可控变更，用一条带明确标记的 UPDATE：

-- 用一个不会和业务数据冲突的标记值 UPDATE APP_DB.ORDERS SET REMARK = 'CDC_VERIFY_20240101' WHERE ORDER_ID = 1001; COMMIT;

第三步，去目标端查这条标记，同时核对变更前后的完整字段。只查到标记不够，要确认op字段是u、before和after都有值。如果after里其他字段是 null，说明逻辑日志只记了变更列，回到 5.1 的现场四。第四步，故意停一次作业再恢复，验证位点持久化：flink cancel之后重新提交，观察是否从上次位点继续而不是重新全量，这一步能验证RETAIN_ON_CANCELLATION和 checkpoint 存储配置是否真的生效。

延迟的计算方式也要固定下来：目标端写入时间减去达梦COMMIT时间，正常在秒级，超过 10 秒就要看 checkpoint 是不是在长时间阻塞。这套验证流程跑通一次大约二十分钟，比事后查“为什么数据不对”便宜太多。

从那以后我每次装完达梦 CDC，都强制走一遍这四步加那张表，尤其是取消再恢复那一步，它验证的是位点持久化，而不是“作业能不能起来”。这套东西从环境对位到验证清单我都整理在一起了，照着顺序走能少绕几个晚上。希望帮到你。

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