达梦日志解析接入Flink：基于LOGMNR的实时增量同步

简介：面向使用 FlinkCDC 将达梦数据库变更接入 Flink 流处理链路的开发者，这套资料围绕日志级实时同步的连接器部署与示例验证展开，适合需要搭建实时数据管道、又希望降低对源库性能影响的中高级读者。压缩包共 5 个文件、约 35.48MB，包含连接器与 JDBC 驱动 jar、示例工程压缩包、SQL 客户端初始化脚本和中文用户手册，分别对应依赖准备、程序参考、环境初始化及配置排错。已有 2091 人学习下载。读者可据此理解基于日志解析捕获插入、更新、删除变更的接入流程，对照 JAVA 作业与 SQL 两种同步方式，掌握数据库地址、端口、账号等连接参数设置，并为实时报表、数据监控、告警等事件驱动场景搭建可复用的同步样例，同时借助手册梳理版本适配与常见问题排查思路。

1. 达梦按日志同步到 Flink：现成 connector 为什么接不上，缺口在哪

接手一个达梦数据库的实时同步需求时，多数人的第一反应是找个 FlinkCDC 连接器配上去就完事，翻一遍官方支持列表就会发现这条路走不通：达梦不在 Flink CDC 的源端清单里，也没有社区版 connector 能直接吃达梦的行级变更。真正能拿到 INSERT、UPDATE、DELETE 三种语义的只有一条路——基于日志解析，把达梦的归档日志用DBMS_LOGMNR挖出来，再喂给 Flink 做流式处理。

这套方案的落点是秒级延迟、对源库只读不写、删除也能捕获，代价是要自己写一个 Source，还要把 SCN 位点、SQL_REDO 文本解析、归档保留窗口这些琐碎的事一并管住。适合已经上了达梦、又被业务逼着要准实时同步的团队；不适合只想跑个 demo、连归档模式都不愿意动的人。

2. 三条同步路线的选型：为什么最后落到 LOGMNR 而不是轮询

2.1 三种方案摆在一起比，差距在删除和位点

做达梦到下游的实时同步，工程上能落地的其实只有三条路：按时间戳或自增 ID 轮询、用触发器把变更写进中间表、读归档日志做解析。前两条在小数据量、无删除场景下能跑，一旦表上有物理删除或者更新频繁，就会开始出问题。

轮询最大的逻辑漏洞是「更新丢中间态」和「删除拿不到」。一张订单表如果在两次轮询之间被改了三次，你只能看到最后一次的结果，下游如果要做累计或者状态机，直接就错了。触发器方案能解决语义问题，但把压力转嫁到业务事务里，高峰期影子表膨胀、触发器报错回滚主事务的事故并不少见，而且每加一张表就要改一次 DDL。

日志解析读的是归档文件，源库那边只多一个归档写入的常规开销，业务侧完全无感。这就是我一般会选它的理由——不是因为时髦，是因为只有它在「删除」和「位点」这两件事上给得出确定答案。

2.2 达梦日志解析链路：从 REDO 到 V$LOGMNR_CONTENTS

达梦的日志体系和 Oracle 很像，走的也是 REDO 日志加归档的路线。当事务提交时，变更先写联机重做日志，归档模式下再由归档进程把写满的日志文件复制到归档目录。DBMS_LOGMNR是内置包，作用是把这些归档日志当成「可查询的数据源」，解析成一行行带 SCN、操作类型、SQL 文本的记录，暴露在V$LOGMNR_CONTENTS视图里。

整条链路是这样跑的：

1. DBMS_LOGMNR.ADD_LOGFILE把待分析的归档文件加进当前挖掘会话，可以加多个；
2. DBMS_LOGMNR.START_LOGMNR指定 SCN 窗口和字典来源，会话开始解析；
3. 查V$LOGMNR_CONTENTS，按SCN、RS_ID、SSN排序读出来；
4. DBMS_LOGMNR.END_LOGMNR结束会话，释放归档文件句柄。

字典来源这一步很关键。DICT_FROM_ONLINE_CATALOG用当前在线数据字典，好处是不用额外导出字典文件，坏处是如果表结构在挖掘窗口内发生过 DDL，解析出来的列名可能对不上历史数据。表结构变更频繁的库，常见做法是每次 DDL 之后导出一份字典快照，挖掘时用DICT_FROM_REDO_LOGS或者指定字典文件。这一点在项目初期不显眼，跑上三个月加了个字段就会翻车。

另一个参数是COMMITTED_DATA_ONLY。加上它之后只返回已提交事务的记录，下游不用自己做事务归并；不加的话会看到大量未提交、后来回滚的中间记录，得靠 XID 自己分组判断。达梦不同小版本对这个常量的支持略有差异，如果START_LOGMNR报常量不识别，就退回到不加该选项，在应用层按 XID 收集、遇到 COMMIT 记录再统一发出。这是我压箱底的兼容写法。

2.3 Flink CDC 这边该落在哪一层

确定了日志解析路线，下一个问题是：代码写在 Flink CDC 的哪个抽象上。

如果目标是接进 Flink CDC Pipeline（也就是flink-cdc-pipeline-connector那套 YAML 编排），需要实现EventSource、EventSourceProvider和一个配套的DataSinkFactory，把达梦的变更转成DataChangeEvent。这套接口的好处是能复用 CDC Pipeline 的 schema 演进和路由能力，坏处是版本之间 API 还在动，升级一次要改一遍。

如果只是要一份能跑、能维护的同步链路，直接落在 DataStream API 上更省事：自定义一个RichSourceFunction发ChangeRecord，下游keyBy主键做幂等写入。这样做的好处是所有细节都在自己手里，checkpoint 语义、位点推进、解析规则想怎么改就怎么改；坏处是要自己处理并行度和顺序性——日志解析源天然是单并行度的，因为 SCN 是全局单调的，拆并行就得按表或者按主键哈希二次分发。

我一般会走第二条路，然后在 Source 后面立刻接一次keyBy把并行度打上去。单并行度读日志、多并行度写下游，是这个场景下最稳的结构。

3. 达梦侧准备：归档、补充日志与一次跑通的 LOGMNR 会话

3.1 开归档：dmarch.ini 加三条 DDL

没有归档就没有日志解析，这是前置条件里最容易漏的一步。先在库里确认当前模式：

-- 返回 Y 表示已开归档，N 表示还在非归档模式 SELECT ARCH_MODE FROM V$DATABASE; -- 顺便看一眼归档相关参数有没有配 SELECT PARA_NAME, PARA_VALUE FROM V$DM_INI WHERE PARA_NAME IN ('ARCH_INI', 'RLOG_APPEND_LOGIC');

如果ARCH_MODE是 N，就得改配置。达梦的归档走的是dm.ini里的ARCH_INI开关加上独立的dmarch.ini配置文件，改完要重启实例：

# dmarch.ini，放在实例目录下，和 dm.ini 同级 [ARCHIVE_LOCAL1] ARCH_TYPE = LOCAL ARCH_DEST = /dmdata/DAMENG/arch ARCH_FILE_SIZE = 1024 ARCH_SPACE_LIMIT = 102400

ARCH_FILE_SIZE单位是 MB，单文件写满 1 GB 就切下一个；ARCH_SPACE_LIMIT是整个归档目录的上限，单位也是 MB，这里给 100 GB。这个上限一定要设，不设的话归档会把磁盘吃满，然后整库挂起——这是运维层面最常见的翻车点。

配好之后用系统过程打开开关再重启：

-- 1 表示静态参数，需要重启实例生效 SP_SET_PARA_VALUE(1, 'ARCH_INI', 1); -- 重启后复查 SELECT ARCH_MODE FROM V$DATABASE;

3.2 补充日志：让 UPDATE 带全整行

这一步决定了你下游拿到的 UPDATE 记录是「只有被改的列」还是「完整一行」。默认情况下，日志里只记录变化字段的前后值，没动的列根本不出现。如果下游是整行覆盖写，那些没出现的列就会被写成 null，数据静默损坏。

-- 库级打开补充日志 ALTER DATABASE ADD SUPPLEMENTAL LOG DATA; -- 表级指定所有列，粒度更细，也更好排查是哪张表没开 ALTER TABLE "BIZ"."ORDERS" ADD SUPPLEMENTAL LOG DATA (ALL) COLUMNS; ALTER TABLE "BIZ"."ORDER_ITEM" ADD SUPPLEMENTAL LOG DATA (ALL) COLUMNS;

库级开关是总闸，表级是逐表声明。我的习惯是两张都开，库级保证新加的表默认带上，表级保证关键业务表一定全列。如果某个达梦版本对表级语法支持不好，退路是在库级打开后重建一次表或者做一次全量重放，让字典把全列信息登记进去。

提示：补充日志会增加归档日志的体积，写入量大的表大概会涨 20% 到 50%。ARCH_SPACE_LIMIT要按这个增量重新算，别沿用没开补充日志时的数值。

3.3 手工跑通一次 LOGMNR 会话

在写 Flink 代码之前，先用 SQL 客户端把整条挖掘链路手工跑一遍。这一步不做，后面调 Flink 的时候分不清是日志没挖出来还是解析写错了。

-- 第一步：把待分析的归档文件加进会话 BEGIN DBMS_LOGMNR.ADD_LOGFILE( LOGFILENAME => '/dmdata/DAMENG/arch/ARCHIVE_LOCAL1/ARCHIVE_20240115_0031.log', OPTIONS => DBMS_LOGMNR.NEW ); END; / -- 第二步：开一个 SCN 窗口开始挖掘 BEGIN DBMS_LOGMNR.START_LOGMNR( STARTSCN => 8800000, ENDSCN => 8810000, OPTIONS => DBMS_LOGMNR.DICT_FROM_ONLINE_CATALOG + DBMS_LOGMNR.COMMITTED_DATA_ONLY ); END; / -- 第三步：看结果，重点看 CSF 和 SQL_REDO 两列 SELECT SCN, TIMESTAMP, XID, OPERATION, SEG_OWNER, TABLE_NAME, CSF, RS_ID, SSN, SQL_REDO FROM V$LOGMNR_CONTENTS WHERE TABLE_NAME IN ('ORDERS', 'ORDER_ITEM') AND OPERATION IN ('INSERT', 'UPDATE', 'DELETE') ORDER BY SCN, RS_ID, SSN; -- 第四步：一定要结束会话，不然归档文件一直被占着 BEGIN DBMS_LOGMNR.END_LOGMNR(); END; /

执行完看三样东西：OPERATION里三种类型是不是都能出现、SQL_REDO是不是可读的insert into/update/delete文本、CSF有没有出现 1。CSF = 1表示这一行是上一条 SQL 的续行，通常是大字段或者超长语句被切成了多行，后面解析的时候必须按RS_ID + SSN归并回去，不然你会看到一堆语法不完整的半截 SQL。

STARTSCN和ENDSCN的取值直接决定了挖掘范围。手工测试时随便给个区间就行，生产上这两个值等价于「上次消费到的位点」和「当前可挖到的最新位点」，是整个同步链路的核心状态。

3.4 归档保留窗口与清理

归档文件是这条链路的唯一数据源，被删掉就等于数据丢了。达梦本身不做自动清理，ARCH_SPACE_LIMIT满了之后行为是停止写归档还是覆盖旧文件，取决于具体配置，两种都不是你想要的——前者会让源库挂起，后者会直接丢数据。

我的做法是外部加一个定时清理任务，但清理之前必须先确认消费位点：

#!/bin/bash # 保留最近 48 小时的归档，且不删除 SCN 大于消费位点的文件 ARCH_DIR=/dmdata/DAMENG/arch/ARCHIVE_LOCAL1 LAST_SCN=$(cat /opt/cdc/last_scn.txt) find "$ARCH_DIR" -name "ARCHIVE_*.log" -mtime +2 | while read -r f; do # 解析文件名里的最大 SCN 做二次校验，低于消费位点才允许删 if check_archive_scn "$f" "$LAST_SCN"; then rm -f "$f" fi done

这里的check_archive_scn是自定义脚本，思路是把每个归档文件的尾部 SCN 记下来，只有整段日志都落在已消费位点之前的才删。生产上更稳的做法是把消费位点上报到监控系统，清理脚本读监控接口判断，别只依赖本地文件——本地文件丢一次，清理逻辑就失去保护。

保留窗口怎么定？看你的最长故障恢复时间。如果一次 Flink 故障排查加重启要两个小时，那归档至少留 24 小时，留 48 小时更保险。窗口太短，一次通宵故障之后你会发现要同步的数据已经不存在了，只能整库重来。

4. Flink 侧落地：全量 bootstrap 与增量 LOGMNR Source 的两段式实现

4.1 全量 bootstrap 与基准 SCN 对齐

全量和增量的衔接是这类同步最经典的难点。做法上有两种：一种是在全量开始前对源表加锁，保证全量期间没有变更；另一种是不加锁，先取一个基准 SCN，全量读完之后从这个 SCN 开始重放增量，重复的部分靠下游幂等吸收。前者影响业务，后者只影响下游写入的幂等性要求，我选后者。

取基准 SCN 有两种常见做法：查V$DATABASE里暴露的当前 SCN 字段（不同小版本字段名可能不一样），或者起一次最小的 LOGMNR 会话取MAX(SCN)。我一般用后者，因为它跟你后面真正读日志的口径完全一致，不会出现「取基准用的是一套 SCN、读日志用的是另一套」的偏差。

-- 起一个只读窗口，拿到当前可挖到的最大 SCN 作为基准 BEGIN DBMS_LOGMNR.ADD_LOGFILE( LOGFILENAME => '/dmdata/DAMENG/arch/ARCHIVE_LOCAL1/CURRENT_ARCH.log', OPTIONS => DBMS_LOGMNR.NEW ); END; / BEGIN DBMS_LOGMNR.START_LOGMNR(OPTIONS => DBMS_LOGMNR.DICT_FROM_ONLINE_CATALOG); END; / SELECT NVL(MAX(SCN), 0) AS BASE_SCN FROM V$LOGMNR_CONTENTS; BEGIN DBMS_LOGMNR.END_LOGMNR(); END; /

拿到BASE_SCN之后开始全量拉取。表不大的话直接用 Flink JDBC connector 的分片读取，按主键做均匀切分：

JdbcSource<RowData> fullSource = JdbcSource.<RowData>builder() .setUrl("jdbc:dm://10.0.0.11:5236") .setUsername("CDC_USER") .setPassword("******") .setTableName("BIZ.ORDERS") // 用主键做分片列，整数主键最理想，字符串主键要先算哈希 .setPartitionColumnName("ID") .setPartitionLowerBound(1L) .setPartitionUpperBound(50_000_000L) .setNumPartitions(8) // 达梦 JDBC 默认 fetchSize 偏小，大表拉取会明显变慢 .setFetchSize(5000) .setDriverOptions(Map.of("dm.jdbc.fetchsize", "5000")) .build();

.setNumPartitions(8)是并行度，按表的数据量和下游承接能力调；.setFetchSize(5000)这个值不要设太大，达梦 JDBC 驱动在超大 fetchSize 下内存占用会明显上升，5000 是个比较稳的折中。分片列必须是有索引的整型主键，用字符串主键做分片会导致每个分片都要全表扫，全量阶段能跑几个小时。

全量发出去的每条记录都打上BASE_SCN标记，下游写入的时候如果遇到同主键的增量记录且 SCN 更大，就以增量为准。这样全量期间发生的变更会在增量阶段重放一次，业务上表现为「短暂看到历史值再被覆盖」，最终一致。

4.2 增量 Source 的 run 循环与 checkpoint 语义

增量部分的核心是一个循环：定位归档文件、开挖掘会话、分页拉取、解析发包、结束会话、推进位点。位点推进的时机是整套实现里最容易写错的地方。

public class DmLogMinerSource extends RichSourceFunction<ChangeRecord> implements CheckpointedFunction, CheckpointListener { private final DmLogMinerOptions options; private volatile boolean running = true; /** 已经 emit 出去的最大 SCN，随 checkpoint 一起进快照 */ private volatile long emittedScn = 0L; /** checkpoint 恢复出来的安全位点，故障后从这里重放 */ private transient ListState<Long> scnState; private transient LogMinerSession session; @Override public void open(Configuration parameters) { // 连接池和归档定位器放这里初始化，别在构造函数里建连接 this.session = new LogMinerSession(options); } @Override public void run(SourceContext<ChangeRecord> ctx) throws Exception { while (running) { // 找 SCN 之后还没消费完的归档文件，单轮最多拿 N 个，防止一次开太多句柄 List<String> logs = ArchiveLocator.locate(options, emittedScn, options.maxLogPerRound); if (logs.isEmpty()) { Thread.sleep(options.pollIntervalMs); continue; } try (LogMinerSession s = LogMinerSession.open(options, logs, emittedScn)) { List<RawChange> batch; do { batch = s.fetch(options.batchSize); // 关键：emit 必须在 checkpointLock 内，保证快照的 SCN 和已发出的数据严格对齐 synchronized (ctx.getCheckpointLock()) { for (RawChange raw : batch) { ChangeRecord record = SqlRedoParser.parse(raw, options); ctx.collect(record); emittedScn = Math.max(emittedScn, raw.getScn()); } } if (batch.size() == options.batchSize) { Thread.sleep(options.batchSleepMs); } } while (batch.size() == options.batchSize && running); } // 一轮结束必须 END_LOGMNR，否则归档文件一直被占着无法清理 } } @Override public void snapshotState(FunctionSnapshotContext context) throws Exception { scnState.clear(); scnState.add(emittedScn); } @Override public void initializeState(FunctionInitializationContext context) throws Exception { scnState = context.getOperatorStateStore() .getListState(new ListStateDescriptor<>("dm-logminer-scn", Types.LONG)); if (context.isRestored()) { for (Long scn : scnState.get()) { emittedScn = Math.max(emittedScn, scn); } } } @Override public void notifyCheckpointComplete(long checkpointId) { // 下游是幂等 upsert，这里只记录，不推进位点 LOG.info("checkpoint {} 完成，安全位点 SCN={}", checkpointId, emittedScn); } @Override public void cancel() { running = false; } }

几个参数和设计点值得展开说。

pollIntervalMs决定空闲时多久探一次新归档，设 1000 毫秒对应秒级延迟。这个值别设成 100 以下，每次探测都要做一次归档目录扫描，太频繁会白烧 IO。

batchSize是单次从V$LOGMNR_CONTENTS拉多少行，2000 到 5000 之间比较合适。太小的话每批都有一次网络往返，大事务场景下延迟会被放大；太大则单次解析的内存峰值高，而且synchronized块持有锁的时间变长，会影响 checkpoint 的完成时间。

maxLogPerRound控制单轮最多 ADD 几个归档文件。设成 4 的意思是单轮最多消费 4 个日志文件，消费完 END 掉再来一轮。不设上限的话，一次故障恢复可能要 ADD 上百个文件，LOGMNR 会话内存会直接打满。

位点语义上，emittedScn是「已经发出去的最大 SCN」，快照存的是它。恢复时从这个值重放，意味着最后一批数据会被重复消费一次。这是有意为之——因为 flink 的 checkpoint 完成到notifyCheckpointComplete之间有窗口期，严格不重放需要两阶段提交，成本太高。下游按主键做幂等 upsert，重复一次没有任何副作用，比丢数据划算得多。

装配层大概长这样：

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); env.enableCheckpointing(30_000L, CheckpointingMode.EXACTLY_ONCE); env.getCheckpointConfig().setMinPauseBetweenCheckpoints(10_000L); env.getCheckpointConfig().setCheckpointTimeout(120_000L); // 位点状态丢了就得重头来，这里配成保留最近 3 个 env.getCheckpointConfig().setExternalizedCheckpointCleanup( ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION); DataStream<ChangeRecord> changes = env .addSource(new DmLogMinerSource(options)) .name("dm-logminer-source") // uid 必须显式指定，不然改了类名或结构，状态恢复直接失败 .uid("dm-logminer-source") // 日志解析源天然单并行度，SCN 全局单调，不能拆 .setParallelism(1); changes.keyBy(ChangeRecord::getTableKey) .process(new IdempotentUpsertFunction(options)) .name("idempotent-upsert") .uid("idempotent-upsert") .setParallelism(8);

.uid()这一行是血泪经验。不写 uid 的话 Flink 会用类名自动生成，某天你重构把类名改了，从 checkpoint 恢复时状态映射不上，直接起不来，只能丢掉位点重新全量。写上之后这个映射就固定了。

4.3 把 SQL_REDO 解析成 ChangeRecord

V$LOGMNR_CONTENTS给的SQL_REDO是给人看的 SQL 文本，不是什么结构化格式。解析它是整套方案里最脆的一环，得按「宁可解析不出来报错，也不要静默解析错」的原则写。

public final class SqlRedoParser { private static final Pattern INSERT = Pattern.compile( "^insert\\s+into\\s+\"?([\\w$#]+)\"?\\.\"?([\\w$#]+)\"?\\s*\\((.*?)\\)\\s*values\\s*\\((.*)\\)\\s*$", Pattern.CASE_INSENSITIVE | Pattern.DOTALL); private static final Pattern UPDATE = Pattern.compile( "^update\\s+\"?([\\w$#]+)\"?\\.\"?([\\w$#]+)\"?\\s+set\\s+(.*?)\\s+where\\s+(.*)$", Pattern.CASE_INSENSITIVE | Pattern.DOTALL); private static final Pattern DELETE = Pattern.compile( "^delete\\s+from\\s+\"?([\\w$#]+)\"?\\.\"?([\\w$#]+)\"?\\s+where\\s+(.*)$", Pattern.CASE_INSENSITIVE | Pattern.DOTALL); private static final Pattern ASSIGN = Pattern.compile( "\"?([\\w$#]+)\"?\\s*=\\s*('(?:[^']|'')*'|null|[-+]?\\d+(?:\\.\\d+)?)", Pattern.CASE_INSENSITIVE); public static ChangeRecord parse(RawChange raw, DmLogMinerOptions opt) { String sql = raw.getSqlRedo().trim(); Matcher m = INSERT.matcher(sql); if (m.matches()) { Map<String, Object> row = new LinkedHashMap<>(); List<String> cols = splitTopLevel(m.group(3)); List<String> vals = splitTopLevel(m.group(4)); if (cols.size() != vals.size()) { // 列数对不上说明被截断了，直接抛出去，别猜 throw new IllegalStateException("列值数量不一致: " + sql); } for (int i = 0; i < cols.size(); i++) { row.put(unquote(cols.get(i)), literal(unquote(vals.get(i)))); } return ChangeRecord.insert(raw.getScn(), raw.getTableName(), row); } // UPDATE 和 DELETE 走同一套 where 解析逻辑，代码略 throw new UnsupportedOperationException("未识别的 SQL_REDO: " + sql); } /** 按顶层逗号切分，忽略引号和括号内部的分隔符 */ private static List<String> splitTopLevel(String s) { List<String> out = new ArrayList<>(); int depth = 0; boolean inQuote = false; StringBuilder buf = new StringBuilder(); for (int i = 0; i < s.length(); i++) { char c = s.charAt(i); if (c == '\'') inQuote = !inQuote; if (!inQuote && c == '(') depth++; if (!inQuote && c == ')') depth--; if (!inQuote && depth == 0 && c == ',') { out.add(buf.toString().trim()); buf.setLength(0); continue; } buf.append(c); } if (buf.length() > 0) out.add(buf.toString().trim()); return out; } }

splitTopLevel为什么必须存在？因为values (...)里面可能有函数调用或者带逗号的字符串字面量，比如('a,b', 3)或者(to_date('2024-01-15','yyyy-mm-dd'), 1)。直接split(",")会把这一个值切成两个，列数立刻对不上。按引号状态和括号深度扫描是最稳的做法。

literal方法处理字面量到 Java 类型的转换：null转 null，带引号的去引号并把连续两个单引号还原成一个，数字走BigDecimal而不是double——金额字段用 double 会丢精度，这个坑在财务场景下代价很高。

注意：一旦解析失败，不要 try-catch 之后跳过继续跑。跳过意味着一条变更静默消失，下游数据就此错了，而且你不知道错了多少条。正确做法是抛异常让作业失败，人工介入修解析规则或者补数据。

4.4 写入端：幂等 upsert 与主键设计

下游写入以主键为单位做覆盖。UPDATE 和 INSERT 都转成 upsert，DELETE 转成按主键删除。表的唯一键就是上游的主键列，这个映射关系要显式配置，别指望自动推断。

写入端用批量 upsert 比逐条快一个数量级，批量大小 500 到 1000 比较合适，事务边界对齐 Flink 的 checkpoint 周期。别用单条自动提交，秒级延迟下每秒几千条变更

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