简介：面向需要在Flink流处理体系中接入达梦数据库的开发者与数据平台运维人员，这份FlinkCDC达梦实时同步工程资料聚焦基于日志的变更数据捕获与低延迟同步问题，可用于实时报表、数据监控、告警及数仓同步等场景。压缩包共315个文件、约341.71MB，以jar依赖与class编译产物为主体，辅以xml配置、java示例、sql脚本及properties等配置说明，便于按工程结构查阅连接器集成与SQL、JAVA两种接入方式。已有776人学习下载，说明其在国产数据库CDC落地方面具有一定参考热度。读者可据此梳理达梦日志解析、Flink作业配置与连接参数设置思路，结合示例类与配置文件理解变更数据流转过程，并获取可复用的工程组织与排错线索。

1. 从一次「全量跑完就卡住」说起：达梦的日志要单独打开

第一次在达梦上做 FlinkCDC 实时同步的人，多半会先把 Flink CDC 的官方连接器列表从头翻到尾，翻三遍也找不到 dameng 这个词。于是顺手拿别的连接器顶上去：全量阶段读得挺欢，一到增量阶段就再也不动了，日志里连个像样的报错都没有。这不是连接器写错了，而是达梦的日志压根没为变更捕获打开——归档没开、RLOG_APPEND_LOGIC还是 0，任何声称「基于日志」的同步都拿不到变更。这篇笔记记的是一套我反复装过的组合：Flink 1.18 + flink-cdc 3.0 + DM8，从空机器装到链路跑通，适合正要把达梦业务表实时搬到 Kafka、另一个库或者湖里的开发者，也适合装到一半卡在增量阶段、想对照着查的人。

2. 环境清单与版本对齐：Flink 1.18、flink-cdc 3.0 与 DM8 的搭配

装这套东西翻车，十次里有六次不是装错，是版本没对齐。Flink、flink-cdc、达梦三者的版本是互相咬合的，随便捡一个「最新版」拼上去，往往在提交作业那一刻才炸，而且报错信息指向性很差。我一般的做法是先把版本钉死，写进一份versions.md丢到项目仓库里，后面谁再装都照这个表来。

2.1 三个组件的版本对照

这里最值得多说一句的是 flink-cdc 的 2.4 和 3.0。这两个大版本不是升级关系，是两套东西：2.4 走的是 Flink SQL / DataStream 的 Source API，jar 名是flink-sql-connector-xxx，你在 SQL Client 里CREATE TABLE ... WITH ('connector'='xxx')；3.0 走的是 Pipeline 框架，由source / sink / pipeline三段 YAML 描述，jar 名是flink-cdc-pipeline-connector-xxx，入口类是 CLI。两套 jar 混着扔进lib/，最常见的表现是启动时NoClassDefFoundError或者提示找不到 factory，看着像 jar 没放对，其实是版本体系串了。现场如果只能拿到 2.4 的包，就别照抄本文的 YAML，把第 4 章的建表语句改成 SQL Client 写法即可，参数名是一致的。

2.2 达梦实例初始化时就得定的三个参数

有三个参数是建库那一刻定死的，事后想改只能重建实例，这是我在一个模拟项目上吃过的最大一次亏：表在达梦里叫BIZ_ORDER，作业里写biz_order，连接器启动直接报对象不存在，查了半天以为是权限问题。

# 达梦初始化实例，注意 CASE_SENSITIVE 和 CHARSET 建库后不可改 dminit PATH=/dm8/data DB_NAME=DAMENG INSTANCE_NAME=DMSERVER \ PORT_NUM=5236 \ CASE_SENSITIVE=Y \ CHARSET=1 \ LENGTH_IN_CHAR=1 \ PAGE_SIZE=16 \ EXTENT_SIZE=16 \ SYSDBA_PWD="Dm@Init2024"

CASE_SENSITIVE=Y表示大小写敏感，建表时不加引号会统一转大写，此时 Flink 侧的表名必须写大写；CASE_SENSITIVE=N则相反，全部按不敏感处理。CHARSET=1是 UTF-8，0是 GB18030，中文环境下这两个选错，同步到 Kafka 就是一堆问号。LENGTH_IN_CHAR=1表示 VARCHAR 的长度按字符算而不是按字节算，跨库同步时字段长度对不上，多半是它没设。已经建好库的现场，用下面这条确认现状，别靠猜：

-- 查实例初始化参数，确认大小写与字符集 SELECT PARA_NAME, PARA_VALUE FROM V$DM_INI WHERE PARA_NAME IN ('CASE_SENSITIVE','CHARSET','LENGTH_IN_CHAR');

2.3 机器、端口与目录规划

单机验证环境两台机器就够：一台跑 Flink 加 Kafka，一台跑达梦；生产上 Flink 至少三台（一个 JobManager、两个 TaskManager），达梦单独一台，别和 Flink 抢 IO，因为日志解析是 IO 密集型活。端口和目录我习惯先落到表里，装机的时候照着对一遍，比事后netstat翻半天强。

3. 装 Flink 与 CDC 连接器：从解压到第一个流作业跑起来

这一章是纯安装动作，命令都能直接抄。装之前先把 JDK 装好，java -version输出 11 再往下走，8 和 17 也能跑，但一台机器上装了三个 JDK 又忘了JAVA_HOME指哪个，是后面各种玄学报错的源头。

3.1 解压 Flink 与环境变量

# 解压到 /opt，用软链把版本号藏起来 tar -xzf flink-1.18.1-bin-scala_2.12.tgz -C /opt ln -sfn /opt/flink-1.18.1 /opt/flink chown -R flink:flink /opt/flink-1.18.1 # 环境变量写到 /etc/profile.d，避免每个 shell 都要 source cat > /etc/profile.d/flink.sh <<'EOF' export FLINK_HOME=/opt/flink export PATH=$FLINK_HOME/bin:$PATH EOF source /etc/profile.d/flink.sh # 确认版本与 Java flink --version java -version

用软链是我这几年唯一坚持下来的习惯：升级的时候解压新目录、ln -sfn换一下指向就行，配置文件另存一份覆盖过去；直接覆盖安装目录的做法，回滚的时候只能靠回忆。注意chown那一步，Flink 用非 root 账号跑是基本要求，root 起来的集群在 checkpoint 写目录时会莫名其妙地失败。

3.2 flink-conf.yaml 里必须改的几项

# /opt/flink/conf/flink-conf.yaml jobmanager.rpc.address: 10.0.0.21 jobmanager.memory.process.size: 1600m taskmanager.memory.process.size: 4096m taskmanager.numberOfTaskSlots: 4 parallelism.default: 2 # checkpoint：实时同步作业的生命线，别用默认值 execution.checkpointing.interval: 60s execution.checkpointing.timeout: 10min execution.checkpointing.min-pause: 30s execution.checkpointing.unaligned: false # 状态后端，1.18 里字段名是 state.backend.type state.backend.type: rocksdb state.checkpoints.dir: file:///data/flink/checkpoints state.savepoints.dir: file:///data/flink/savepoints # 失败重试，网络抖一下别让作业直接死 restart-strategy.type: fixed-delay restart-strategy.fixed-delay.attempts: 10 restart-strategy.fixed-delay.delay: 10s

execution.checkpointing.interval设 60 秒是个折中：太快，快照阶段和日志读取抢资源；太慢，作业挂了要回放的数据多。execution.checkpointing.timeout一定要给足，达梦侧全量读大表时 checkpoint 超时是最常见的失败原因，我给到 10 分钟。state.backend.type这个字段名是 1.18 的写法，1.16 及以前叫state.backend，直接抄旧配置过来会提示未知配置项，作业照样起，但状态后端退回内存，跑两天就 OOM——这类「配置没报错但没生效」的坑，比直接报错难查得多。验证阶段用file://就够了，生产上换成 HDFS 或者对象存储，本地盘做 checkpoint 目录，机器一挂位点全丢。

3.3 连接器 jar 落位与依赖冲突排查

cd /opt/flink/lib # 三个 jar 缺一不可：CLI、达梦 source、目标端 sink cp /root/pkgs/flink-cdc-cli-3.0.1.jar . cp /root/pkgs/flink-cdc-pipeline-connector-dameng-3.0.1.jar . cp /root/pkgs/flink-cdc-pipeline-connector-kafka-3.0.1.jar . # 确认 jar 里注册的 factory 标识，YAML 里的 type 必须和它对上 unzip -p flink-cdc-pipeline-connector-dameng-3.0.1.jar \ META-INF/services/org.apache.flink.cdc.common.factories.Factory # 看 lib 下有没有同名不同版本的依赖，冲突大多出在这 ls -1 /opt/flink/lib | grep -Ei 'guava|jackson|kafka-clients|slf4j|log4j' | sort

第二段unzip -p是我每次装完连接器必做的一步：它会把 jar 里注册的 factory 实现类打出来，类名里通常带连接器标识，YAML 的type就照它填。手上这份连接器如果是团队自己封装的，type很可能不叫dameng，硬填就是启动即报「找不到 source factory」。

第三段是排依赖冲突的土办法，但管用。Flink 的lib/是按文件名顺序加载的，guava-31.jar和guava-19.jar同时存在时，先被扫到的那个赢，另一个版本的类在运行期才炸，报错是NoSuchMethodError，堆栈还指在你的代码上。规矩只有一条：连接器自带的依赖必须是 shaded 版，非 shaded 的第三方包一律不放lib/，需要的话丢到用户 jar 里。

3.4 起集群并提交第一个作业

# 启动集群，standalone 模式下 JobManager 和 TaskManager 在一台机上 /opt/flink/bin/start-cluster.sh # 确认两个进程都在 jps | grep -E 'StandaloneSessionClusterEntrypoint|TaskManagerRunner' # 提交 pipeline 作业，入口类以 jar 里实际存在的为准 /opt/flink/bin/flink run \ -c org.apache.flink.cdc.cli.CliFrontend \ /opt/flink/lib/flink-cdc-cli-3.0.1.jar \ /opt/jobs/dm-to-kafka.yaml # 查看作业状态和日志 /opt/flink/bin/flink list tail -f /opt/flink/log/flink-*-taskexecutor-*.log

第一次提交建议把 sink 先设成打印或者写一个测试 topic，确认 source 能吐出数据再换正式目标。入口类名如果不对，flink run会直接告诉你找不到主类，用unzip -l flink-cdc-cli-3.0.1.jar | grep CliFrontend查一下真实路径再填，别照抄网上的类名。作业提交后 Web UI 上应该能看到RUNNING，如果几秒内变成RESTARTING，别急着改配置，先看 TaskManager 日志的最后 50 行，八成是连接器参数或者达梦连不上。

4. 打通达梦日志源：归档、逻辑日志与连接器参数怎么配

Flink 侧装完只是把锅架好了，真正决定能不能实时的是达梦这一侧。达梦默认跑在非归档模式，日志写满就覆盖，变更捕获没有任何可读的素材；归档开了但逻辑日志开关没开，读出来的日志里也没有主键和变更类型，连接器只能干瞪眼。

4.1 CDC 专用账号与最小权限

-- 单独建一个 CDC 账号，不要拿 SYSDBA 跑同步作业 CREATE USER CDC_USER IDENTIFIED BY "Cdc@2024#dm"; GRANT RESOURCE TO CDC_USER; -- 业务表的读权限，按实际表逐个授，别 GRANT SELECT ANY TABLE GRANT SELECT ON DAMENG.BIZ_ORDER TO CDC_USER; GRANT SELECT ON DAMENG.BIZ_PAY TO CDC_USER; -- 查日志与归档状态需要的视图权限 GRANT SELECT ON V$ARCHIVED_LOG TO CDC_USER; GRANT SELECT ON V$RLOG TO CDC_USER; GRANT SELECT ON V$DM_INI TO CDC_USER;

用 SYSDBA 跑同步作业是最省事也最危险的做法：密码会明文躺在 YAML 里，作业日志一打全就泄了；而且权限过大，连接器内部拼错一个表名都可能读到不该读的东西。我一般单独建账号，只授需要的表，配置里的密码用环境变量注入，YAML 里写${DM_PASSWORD}这种占位符由提交脚本替换。

4.2 开归档与逻辑日志

-- 1) 切到 MOUNT 状态才能改归档模式 ALTER DATABASE MOUNT; ALTER DATABASE ARCHIVELOG; -- 2) 加一个本地归档目录，大小按日增量估 ALTER DATABASE ADD ARCHIVELOG 'DEST=/dm8/arch, TYPE=LOCAL, FILE_SIZE=1024, SPACE_LIMIT=204800'; -- 3) 回到 OPEN 状态 ALTER DATABASE OPEN; -- 4) 打开逻辑日志，这是「基于日志同步」的前提 SP_SET_PARA_VALUE(2, 'RLOG_APPEND_LOGIC', 1); -- 5) 确认参数已落盘，然后重启实例 SELECT PARA_NAME, PARA_VALUE FROM V$DM_INI WHERE PARA_NAME = 'RLOG_APPEND_LOGIC';

FILE_SIZE单位是 MB，单个归档文件 1G；SPACE_LIMIT也是 MB，这里给 200G 上限，写满之后达梦会停写而不是覆盖，所以磁盘监控必须跟上——归档盘写满导致业务库挂起，是我见过代价最高的一次事故。

SP_SET_PARA_VALUE第一个参数是作用域：1表示动态生效（内存加配置文件），2表示只写配置文件、需要重启实例才生效。RLOG_APPEND_LOGIC属于静态参数，必须重启。取值上，0是关闭，1会记录主键与变更类型，部分小版本支持更大的取值以记录变更前后的列值镜像，做整行同步时更省事，代价是日志量成倍上涨。具体支持到哪个值，以现场版本的参数手册为准，别直接抄。

提示：开归档和改静态参数都要重启实例，生产上排进变更窗口，重启前先确认归档目录有写权限，否则实例可能起不来。

4.3 连接器参数与作业 YAML

# /opt/jobs/dm-to-kafka.yaml source: type: dameng # 与 jar 内注册的 factory 标识一致 name: dm-source hostname: 10.0.0.31 port: 5236 username: CDC_USER password: ${DM_PASSWORD} tables: DAMENG\.BIZ_.* # 库名.表名，支持正则；注意大小写 startup: mode: initial # initial=先全量再增量，latest=只追增量 sink: type: kafka name: kafka-sink properties.bootstrap.servers: 10.0.0.21:9092 topic: ods_dm_biz format: json pipeline: name: dm-biz-to-kafka parallelism: 2 local-time-zone: Asia/Shanghai

tables的写法是最容易出错的一处：达梦在大小写敏感模式下，未加引号建的表名会存成大写，所以正则里要写DAMENG\.BIZ_.*而不是小写。startup.mode决定起点，initial会先做一次全量快照再切增量，latest只从当前位点往后追。第一次上线用initial，作业重启或者换版本时改成latest并配合 savepoint，否则每次重启都重跑一遍全量，目标端数据翻倍。

4.4 位点管理与增量快照的起点

全量切增量的那一刻是整条链路最脆的时候。快照阶段读的是表，增量阶段读的是日志，两者靠一个位点衔接，衔接点没记准就会出现「全量最后几条重复」或者「切换瞬间丢几条」。我的做法是：作业上线前先在达梦侧记下当前归档序列号，作业跑起来后对比目标端条数与源端COUNT(*)，差值在个位数以内再放业务方使用。

# 每次升级或改配置前，先打 savepoint /opt/flink/bin/flink stop -p file:///data/flink/savepoints/dm-biz-v1 \ $(/opt/flink/bin/flink list | grep dm-biz-to-kafka | awk '{print $4}') # 从 savepoint 恢复，位点不丢，也不会重跑全量 /opt/flink/bin/flink run \ -s file:///data/flink/savepoints/dm-biz-v1 \ -c org.apache.flink.cdc.cli.CliFrontend \ /opt/flink/lib/flink-cdc-cli-3.0.1.jar /opt/jobs/dm-to-kafka.yaml

flink stop和flink cancel的区别要记牢：stop会先做一次 savepoint 再优雅停，cancel是直接杀。改并行度、换 jar、调 checkpoint 间隔都必须走stop，直接用cancel等于把位点扔了，下次启动只能重头来。

5. 排查与避坑：达梦实时同步里最容易翻车的五个点

这一章的五条都是我或者同事实际踩过的，按「现象、原因、解决」记，遇到的时候直接对号入座比重装一遍快。

5.1 现象：全量读完了，增量一条不动

现象很典型：作业状态RUNNING，指标里numRecordsIn停在全量条数不再增长，日志没有任何 ERROR，只有零星 WARN。

原因是达梦侧只开了归档没开逻辑日志，或者RLOG_APPEND_LOGIC改了但实例没重启。归档日志里记录的是页级变更，没有主键和变更类型，连接器解析不出可用于回放的记录，只能静静地等。

解决办法是先确认参数真的生效，再确认归档在写：

-- 参数是否已生效 SELECT PARA_NAME, PARA_VALUE FROM V$DM_INI WHERE PARA_NAME='RLOG_APPEND_LOGIC'; -- 归档是否在推进，隔一分钟查两次对比 SELECT SEQUENCE#, FIRST_TIME, NEXT_TIME, ARCHIVED FROM V$ARCHIVED_LOG ORDER BY SEQUENCE# DESC LIMIT 5;

参数是 0 就SP_SET_PARA_VALUE(2,'RLOG_APPEND_LOGIC',1)后重启；归档序列号不推进，说明业务压根没写或者归档目录挂了。两个都对还不动，去达梦侧随便改一条测试表，再看源端连接器的读取位点指标有没有跳。

5.2 现象：启动就报对象不存在，但表明明在

报错里带着表名，去达梦客户端SELECT * FROM 表名一切正常，偏偏连接器说找不到。

原因基本是两个：一是大小写，实例CASE_SENSITIVE=Y而作业里写了小写；二是 schema 没带全，达梦默认 schema 是登录用户同名 schema，CDC_USER登录时查BIZ_ORDER实际找的是CDC_USER.BIZ_ORDER，而表在DAMENG下。

解决是 YAML 里把库名.表名写全并且大小写对齐：

tables: DAMENG\.BIZ_ORDER # 大写实例 + 显式 schema

不确定表名到底怎么存的，直接查数据字典：

SELECT OWNER, TABLE_NAME FROM ALL_TABLES WHERE TABLE_NAME LIKE '%BIZ%';

5.3 现象：作业跑一阵就 RESTARTING，日志里是 checkpoint 超时

Web UI 上 checkpoint 一栏全是 failed，作业反复重启，日志里Checkpoint expired before completing。

原因通常有三个叠加：checkpoint 目录指到了本地盘且磁盘满；全量阶段单并行度读大表，60 秒做不完一次快照；execution.checkpointing.timeout用了默认值，压根没给全量留时间。

解决按顺序来：先把state.checkpoints.dir换成共享存储并确认可写，再把 timeout 抬到 10 分钟，最后把全量阶段的并行度提上去。快照阶段内存吃紧的，taskmanager.memory.process.size从 4G 加到 8G，RocksDB 的托管内存比例也调一下：

taskmanager.memory.process.size: 8192m taskmanager.memory.managed.fraction: 0.5 execution.checkpointing.timeout: 10min

5.4 现象：换了连接器版本后NoSuchMethodError

堆栈指向某个第三方包，类名眼熟但方法签名对不上，重启几次位置还变。

原因是lib/下同时存在两个版本的同一个依赖，Flink 按文件名顺序加载，先扫到的赢。连接器升级时旧 jar 没删干净是最常见的情形。

解决是把lib/清一遍再放：

# 列出疑似重复依赖 ls -1 /opt/flink/lib | sed -E 's/-[0-9].*//' | sort | uniq -d # 确认某个类到底从哪个 jar 加载的 for j in /opt/flink/lib/*.jar; do unzip -l "$j" 2>/dev/null | grep -q 'com/google/common/collect/ImmutableList.class' && echo "$j" done

第二条命令能直接告诉你是哪个 jar 提供了这个类，删掉旧的那个就行。规矩还是那条：非 shaded 的第三方依赖不进lib/。

5.5 现象：重启后目标端数据翻倍

作业重启一次，Kafka 里同一批订单出现两份，业务方找上门。

原因有两层：重启时用了cancel而不是stop，位点回到上一个 checkpoint；目标端又不是幂等的，重复写就变成真重复。CDC 的语义是至少一次，端到端精确一次要靠 sink 侧去重。

解决分两步：操作上强制走 savepoint，任何变更前先flink stop -p；链路上给 Kafka 消息带主键，下游按主键做 upsert 或者用 Flink 的幂等写入。临时救急可以按主键去重后重放，但根子上还是要把「先 stop 再改」写进操作规范。

6. 装完怎么验证：三段打点确认链路真的实时

装完不作验证就交给业务，是我早期最常犯的错。后来固定成三段打点：源端条数、作业位点、目标端条数，三段对得上才算跑通。

第一段在达梦侧，取基准值：

-- 记下当前条数和最大更新时间，作为对账基准 SELECT COUNT(*) AS CNT, MAX(UPDATE_TIME) AS MAX_TS FROM DAMENG.BIZ_ORDER;

第二段在 Flink 侧，看作业是不是真的在动。Web UI 的numRecordsIn和numRecordsOut隔一分钟刷新一次，两个数都在涨才算增量生效；再看 checkpoint 面板，最近三次都是 completed 且耗时稳定，才说明这条链路扛得住。

第三段在目标端，用 Kafka 自带脚本抽几条看内容：

# 抽 5 条看字段是否完整、中文有没有乱码、时间字段时区对不对 /opt/kafka/bin/kafka-console-consumer.sh \ --bootstrap-server 10.0.0.21:9092 \ --topic ods_dm_biz --max-messages 5 \ --property print.key=true

三段之外还有一条最直接的路子：在达梦上手工改一条测试数据，记下时间，然后掐表看目标端多久能查到。我一般要求 10 秒内可见，超过这个数就得回头看 checkpoint 间隔是不是设太大。

从那以后我每次装完这套环境，都强制先跑一遍源端COUNT(*)对账、再看 checkpoint 面板、最后才敢把业务表加进tables正则里，顺序反了就容易在半夜被叫起来。这套组合用到的配置样例、建库脚本和检查命令我都整理在了资料里，三个地址改成你自己环境的就能直接跑，希望帮到你。

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