简介：面向Flink与达梦数据库集成场景的实时同步工程包，围绕变更数据捕获（CDC）提供可参考的连接器实现与示例代码，适合数据集成开发者和实时数仓建设者解决达梦日志解析、增量数据低延迟同步的问题。压缩包共315个文件，约341.71MB，以jar依赖、xml配置、class编译产物、java源码和docx说明为主，另含properties配置、sql脚本与工程描述文件，覆盖从依赖准备到作业配置的完整结构。已有776人学习下载。包内可参考CDC核心类、反序列化Schema、Flink SQL接入示例和统计工具，帮助理解插入、更新、删除变更的捕获与转换流程，并对照连接参数完成作业搭建与问题排查，适合作为达梦实时同步方案的学习与落地参考。
1. 从一张对不上的对账表说起：Flink CDC 接达梦的日志实时同步要解决什么
某公司的交易主库跑在达梦上，下游报表库和搜索索引靠每天凌晨的一次批量抽数同步。业务方下午改的单子，第二天上午才在报表里出现，月底对账两边差出几百条，运维只能手工补。团队想上 Flink CDC 做实时链路，翻开文档才发现一个现实问题：Flink CDC 官方连接器覆盖了 MySQL、PostgreSQL、Oracle、SQL Server 等主流库，达梦不在名单里，社区也找不到能直接拿来用的成熟实现。这套方案要解决的就是这一段空缺——把达梦的归档日志解析成变更事件流，包装成 Flink CDC 能消费的 Source，让「改一笔、秒级可见、重启不丢数」这条链路真正跑起来。它适合两类人：正在做数据同步方向毕设或课设、需要一套能自圆其说的完整设计的学生；以及手上压着一堆存量达梦库、要给它们补一条低侵入实时链路的工程师。下面按需求、选型、架构、实现、避坑、验收的顺序把这份方案拆开。
2. 需求与技术选型：为什么日志解析比定时抽数更适合达梦存量库
动手写代码之前，先把需求边界划清楚。很多同步项目最后返工，问题并不在代码写得好不好，而是一开始把「同步」当成一个笼统的词，全量、增量、结构变更混在一个任务里，出了偏差也说不清是哪一段的责任。
2.1 三类同步需求的边界划分
第一类是存量初始化。达梦库里一张订单表可能几千万行，全量阶段要考虑的是怎么在不锁死业务的前提下把数据搬完，以及搬的过程中新产生的变更怎么办。
第二类是增量捕获。业务只关心INSERT、UPDATE、DELETE三种行级变更，其中DELETE是最容易被忽略的一种——定时抽数方案根本看不见删除，只能靠业务表加is_deleted软删字段，改造成本转嫁给了业务方。
第三类是结构变更。加一列、改个字段长度、新建一张表，这些 DDL 如果不同步，下游任务会在某天凌晨突然报「字段不存在」，而且是那种查半天查不出来的错误。
划边界的原则是：增量与结构变更必须走同一条日志通道，全量单独做但不能与增量脱节。至于双向同步回写、跨库分布式事务一致性这些，明确写在方案的非目标里，别让评审老师或者甲方以为你要做的是别的项目。
2.2 三种取数方式的对比与选型结论
把常见的三条路摆在桌面上比，结论其实很明显。
| 方式 | 原理 | 延迟量级 | 对源库影响 | 能否捕获 DELETE | 适用场景 |
|---|---|---|---|---|---|
| 定时抽数 | 调度器按周期跑SELECT，靠时间戳字段过滤 | 分钟到小时 | 全表或范围扫描，压力随数据量线性上升 | 不能，除非加软删字段 | 小表、容忍延迟的报表 |
| 触发器 + 影子表 | 在业务表挂触发器，变更写入中间表 | 毫秒 | 写放大，事务提交变慢 | 能 | 表数量少、写入量低的系统 |
| 日志解析 | 读归档日志/联机日志，逻辑解析出变更 | 秒级 | 只读日志文件，几乎不碰业务表 | 能 | 大表、要求低侵入的存量库 |
触发器方案看起来最直接，实际在交易主库上是血泪经验：一张高频写入的表挂上触发器，事务提交时间涨了三成，DBA 第二天就会来找你。所以选型落在第三条路上，达梦自身提供了DBMS_LOGMNR这个日志解析包，用法和业内同类逻辑日志解析工具很像，支持把归档日志文件加进解析会话，再从内容视图里按顺序读出重做记录。
这里要强调一个容易讲错的点：日志解析拿到的不是「已经拼好的行数据」，而是重做记录。同一条UPDATE会拆成若干条列级变更，INSERT会拆成OP_CODE=1的插入和后续的列值，最终要在连接器内部按事务和行标识合并还原成一条完整的行级事件。这一步做不好，下游收到的就是一堆碎数据。
2.3 链路拓扑与组件职责划分
整体链路是四段：源端、捕获端、计算端、落地端。
- 源端：达梦主库开启归档模式，日志文件按期保留，保留窗口必须大于任务最长停机时间，否则重启后位点已经被归档清理，只能重做全量。
- 捕获端：自研的 Flink CDC Source，内部封装 LogMiner 会话，负责把重做记录还原成变更事件，并把当前位点写进 checkpoint。
- 计算端：Flink 作业，可以是 DataStream 做宽表打平，也可以是 Flink SQL 做过滤和聚合，取决于下游要的是明细还是指标。
- 落地端：Kafka 做消息分发，或者直接 upsert 到 OLAP 库。落地端的模型必须带主键，否则第二章讲的「至少一次」语义会在下游变成重复数据。
组件职责清楚了，接下来的架构设计才有落点。我一般会把这四段画成一张带位点流向的图，标清楚哪一段可能重复、哪一段可能丢，评审时这张图比代码更有说服力。
3. 总体架构设计：捕获端、计算端、落地端的三段式拆分
三段式拆分的核心思路是「把不确定性关在一个盒子里」。日志解析的不确定性最高（会话、位点、事务边界都在这里），所以全部收敛在捕获端内部，对外只暴露标准的变更事件；计算端只做无状态的转换；落地端靠主键保证最终一致。这样任何一段出问题，排查范围是清晰的。
3.1 达梦日志体系：归档日志、联机日志与 LSN 的定位关系
要读懂这段设计，先得知道达梦的日志是怎么组织的。联机重做日志是循环写的，写满一个文件就切下一个，被覆盖之前如果开了归档模式，会先复制到归档目录。这意味着两件事：任务读日志的速度必须跟得上日志产生的速度，跟不上就会遇到「要读的文件已经被覆盖」；任务重启时的起始位点必须落在归档里还存在的位置。
位点用 LSN（日志序列号）来定位比用时间戳稳。时间戳有精度问题，同一毫秒内的多个事务排序可能错乱；LSN 单调递增，天然带顺序。解析会话的基本流程是：
-- 1. 把归档日志文件加入解析会话（路径来自归档目录，按需批量添加） CALL DBMS_LOGMNR.ADD_LOGFILE('/dm/arch/ARCHIVE_LOCAL1_0x1A2B_20250101.log', DBMS_LOGMNR.NEW); -- 2. 启动解析，指定起始位点；位点来自上一次 checkpoint 保存的 LSN CALL DBMS_LOGMNR.START_LOGMNR( STARTLSN => 128340672, OPTIONS => DBMS_LOGMNR.DICT_FROM_ONLINE_CATALOG + DBMS_LOGMNR.COMMITTED_DATA_ONLY ); -- 3. 按 LSN 顺序读变更，CFLAG 仅取已提交事务，避免读到脏数据 SELECT SCN, LSN, TIMESTAMP, OPERATION, SEG_OWNER, TABLE_NAME, ROW_ID, RS_ID, CSF, DATA, SQL_REDO FROM V$LOGMNR_CONTENTS WHERE LSN > 128340672 ORDER BY LSN;几个参数值得单独说。DICT_FROM_ONLINE_CATALOG表示用当前库的数据字典做解析，好处是不用每次导字典文件，代价是解析历史日志时如果表结构已经改过，字段映射会对不上——这是后面避坑章节要讲的一个坑。COMMITTED_DATA_ONLY只输出已提交事务的记录，能过滤掉大量脏读，但它会把同一个事务的记录攒到提交那一刻才吐出来，大事务因此会形成一次性的大批量输出，内存要留够。
还有一点：V$LOGMNR_CONTENTS里OPERATION的取值区分了INSERT、UPDATE、DELETE和 DDL，同一个UPDATE语句会按列拆成多条记录，共享同一个RS_ID。还原完整行的时候要按RS_ID聚合，不能一条记录当成一条数据往下发。
3.2 全量快照与增量衔接：切分、起始位点与重复区间的处理
全量和增量的衔接是整个方案最容易出错的地方。常见的错误做法是先跑完全量再去开增量会话，中间的窗口期变更直接丢了；另一种极端是全程锁表，业务方半夜打电话过来。
我的做法是「先记位点，再跑全量，最后从位点回放」：
- 任务启动时先查一次当前最新 LSN，记作
startLsn，同时记下当前时间戳作为全量快照的一致性参考。 - 全量阶段按主键区间切分成多个 chunk，每个 chunk 用
SELECT ... WHERE id BETWEEN ? AND ? ORDER BY id分批拉取，批次大小控制在几千行，避免一次性把结果集拉进内存。 - 全量跑完后，增量会话从
startLsn开始回放。这个区间内已经全量同步过的行会被再发一次，但落地端是主键 upsert，重复写入的结果和写一次一样。
这个设计牺牲的是「不重复」，换来的是「绝对不丢」。在实时同步里，这是一个正确的取舍——丢一条数据要人工排查半天，重复一条数据 upsert 一下就过去了。
切分键的选择也有讲究。主键是自增整型最好办；如果是联合主键或者字符串主键，退化成按ROWID范围切分，或者干脆单线程分批游标读，性能差一些但不会出错。这一点在课设答辩时经常被问到，提前想好答案。
3.3 Schema 变更与元数据传递：DDL 怎么跟着数据一起过河
DDL 的处理策略分三档，按项目复杂度选：
| 策略 | 做法 | 适用 |
|---|---|---|
| 白名单 + 告警 | 只同步ADD COLUMN，其他 DDL 发告警人工介入 | 结构稳定的核心库 |
| 自动演进 | 解析 DDL 转成 SchemaChangeEvent，下游自动加列 | 迭代频繁的业务库 |
| 全量重建 | DDL 触发对应表重新快照 | 表小、变更少 |
大多数项目落在第一档就够了。实现上，解析会话里OPERATION为 DDL 的记录会带SQL_REDO，但这条语句是原始 DDL 文本，直接扔给下游没有意义，需要解析成结构化的变更描述再发出去。我一般只处理加列这一类，把它翻译成一条带表名和列定义的 JSON 事件，Kafka 侧由消费方自己决定要不要扩表。
提示：DDL 事件到达下游之前，对应的数据行可能已经在路上了。落地端如果开了严格 schema 校验，会因为「收到未知字段」直接失败。稳妥的做法是落地端对未知字段先丢弃并计数，等结构同步完再打开校验。
4. 关键模块实现：自定义 Flink CDC Source 的四个核心部分
这一章落到代码。整体实现基于 Flink 的 FLIP-27 Source 接口，配合 Flink CDC 提供的变更事件模型，把达梦的日志记录包装成标准的DataChangeEvent，这样上层的 DataStream 或 Flink SQL 作业感知不到底层是哪种数据库。
4.1 连接器配置与核心参数表
配置项集中在一个DmSourceConfig里，从 Flink 的Configuration读取，方便在 SQL 作业里用WITH子句传参。
| 参数 | 含义 | 建议值 |
|---|---|---|
dm.jdbc.url | 达梦 JDBC 连接串 | jdbc:dm://host:5236/BIZ |
dm.username/dm.password | 解析账号，需有日志解析权限 | 单独建只读账号 |
dm.logminer.start-lsn | 起始位点，为空时自动取当前最新 LSN | 断点恢复时由状态覆盖 |
dm.logminer.batch-size | 单次从内容视图拉取的记录数 | 2000 |
dm.logminer.poll-interval-ms | 无新日志时的轮询间隔 | 500 |
dm.snapshot.fetch-size | 全量阶段 JDBC 游标批量大小 | 4096 |
dm.snapshot.parallelism | 全量 chunk 并发度 | 不超过表数的 1/4 |
dm.table.include-list | 需要同步的表白名单 | 正则，逗号分隔 |
dm.logminer.batch-size这个值不要贪大。日志记录里带SQL_REDO文本，单条可能几 KB，2000 条一批大约几 MB，够用；调到两万，遇到大事务时容易把 TaskManager 的堆打满。
4.2 日志读取与事件解析的核心实现
下面这段是增量阶段的主循环，负责把日志记录还原成行级变更事件。省略了异常分支和指标埋点，保留主干。
public class DmLogMinerReader implements AutoCloseable { private final Connection conn; private final DmSourceConfig config; private long currentLsn; // 当前消费到的位点，checkpoint 时落盘 /** 从当前位点开始回放，返回本次拉到的一批变更事件 */ public List<DataChangeEvent> poll() throws SQLException { List<DataChangeEvent> events = new ArrayList<>(); String sql = "SELECT LSN, RS_ID, OPERATION, TABLE_NAME, DATA, SQL_REDO " + "FROM V$LOGMNR_CONTENTS " + "WHERE LSN > ? AND OPERATION IN ('INSERT','UPDATE','DELETE') " + "ORDER BY LSN"; try (PreparedStatement ps = conn.prepareStatement(sql)) { ps.setFetchSize(config.getBatchSize()); ps.setLong(1, currentLsn); try (ResultSet rs = ps.executeQuery()) { RowAssembler assembler = new RowAssembler(); while (rs.next()) { long lsn = rs.getLong("LSN"); String rsId = rs.getString("RS_ID"); String op = rs.getString("OPERATION"); String table = rs.getString("TABLE_NAME"); // 同一事务的多条列级记录按 RS_ID 聚合成一行，未聚合完不向下游发 assembler.accept(rsId, op, table, rs.getString("DATA"), lsn); if (assembler.isComplete(rsId)) { events.add(assembler.build(rsId)); // 产出标准 DataChangeEvent } currentLsn = lsn; // 位点只在整行聚合完成后才推进，见 4.3 } } } return events; } public long getCurrentLsn() { return currentLsn; } }几个设计点解释一下。
RowAssembler是这个实现里最关键的内部结构。日志内容视图给出的DATA是一串列序号和值的键值对，INSERT会先出现一条OP_CODE=1插入记录，再跟若干条OP_CODE=2的列值记录，DELETE则带的是删除前的行镜像。把裸记录直接往下发，下游根本没法用，必须按RS_ID聚合成完整行，再根据OPERATION判断是 upsert 还是 delete。聚合缓冲区要设上限，遇到超过阈值还没聚合完的，说明这不是普通行级操作，直接跳过并打警告，别让它无限占内存。
currentLsn的推进时机要小心。代码里写在循环末尾，看起来每行都更新，但真正提交给 checkpoint 的是「已经完整产出事件之后」的那个位点。如果把位点更新放在聚合完成之前，任务在聚合中途挂掉，重启后从聚合靠后的位置开始读，前面那半行就永远丢了。
4.3 Checkpoint 与断点续传的状态设计
Flink 的容错语义是「状态快照 + 重放」，Source 需要实现SourceReader的snapshotState，在 checkpoint 触发时把当前位点存进去。
@Override public void snapshotState(long checkpointId) throws Exception { DmSourceState state = new DmSourceState(); state.startLsn = reader.getCurrentLsn(); state.tableOffsets = snapshotTableOffsets(); // 全量阶段的 chunk 进度 OperatorStateStore store = context.getOperatorStateStore(); ListState<DmSourceState> listState = store.getListState(new ListStateDescriptor<>("dm-logminer-state", DmSourceState.class)); listState.update(Collections.singletonList(state)); }状态分两段：全量没跑完时存各 chunk 的进度，全量跑完后只存 LSN。DmSourceState需要实现Serializable，字段用基本类型，别塞进Connection这类不可序列化的对象，否则 checkpoint 时会直接抛异常——这个错误信息非常不友好，第一次遇到能查很久。
这里还有个和日志保留期的联动：checkpoint 间隔决定了最坏情况下要回放多远，如果 checkpoint 间隔是 3 分钟、任务重启加恢复要 10 分钟，那至少要回放 13 分钟的日志。归档保留策略必须覆盖这个窗口，最好留出几倍余量。
4.4 落地端幂等写入：主键模型与两阶段提交
落地端的模型决定了整条链路的最终一致性。
写 Kafka 的话，消息 key 用主键，配合enable.idempotence=true只能保证单分区不重，跨分区重复还是存在，消费端必须自己按主键去重。写 OLAP 库的话走主键模型的 upsert：
-- 落地表用主键模型，同主键多版本自动覆盖 CREATE TABLE dwd_order ( order_id BIGINT NOT NULL, user_id BIGINT, order_status VARCHAR(32), amount DECIMAL(18,2), update_time DATETIME, dm_lsn BIGINT -- 记录来源位点，排查重复时用得上 ) UNIQUE KEY(order_id) DISTRIBUTED BY HASH(order_id) BUCKETS 16; -- 写入时按位点过滤，后到的旧版本不会覆盖新版本 INSERT INTO dwd_order VALUES (?, ?, ?, ?, ?, ?) ON DUPLICATE KEY UPDATE dm_lsn = IF(VALUES(dm_lsn) > dm_lsn, VALUES(dm_lsn), dm_lsn);dm_lsn这一列是我强烈建议加的。日志回放时，同一行新旧版本到达顺序不保证（并发度大于 1 时尤其明显），只按主键 upsert 会出现「旧值覆盖新值」。带上位点做条件更新，乱序问题就被兜住了。这一列在验收阶段也很好用，能直接看出某条数据是从哪个位点来的。
5. 避坑与排查：达梦日志同步最容易翻车的五个点
这几条都是实际跑链路时踩出来的，按「现象 → 原因 → 解决」组织，能省掉不少查日志的时间。
5.1 现象：任务重启后下游数据重复三遍
现象是作业从 checkpoint 恢复后，同一批数据在下游出现两到三次，但count(distinct 主键)又是对的。
原因有两层。第一层是 Flink 的「至少一次」语义，checkpoint 之间的事件会被重放；第二层更隐蔽，日志解析会话在重启时如果用了STARTLSN之后没做去重，会把上次已消费的位点区间再读一遍。两个因素叠加，重复就被放大了。
解决办法是落地端必须幂等，这个没得商量。同时 Source 侧可以在状态里额外记一个「最近 N 个已发出事件的位点集合」，重启时跳过这些位点，能显著减少重复量。我一般把去重窗口设成 checkpoint 间隔内的平均事件量，超过就退化回纯幂等方案。
5.2 现象：增量阶段一条数据都读不到
全量跑完了，增量会话启动，日志里V$LOGMNR_CONTENTS查出来是空的，但业务明明在写。
第一个怀疑方向是归档没开。查一下归档状态，V$DATABASE或者对应的视图里ARCH_MODE不是Y的话，日志根本没落归档，解析会话当然读不到东西。切换归档模式需要在挂载状态下做，属于 DBA 操作，别自己在生产库上试。
第二个方向是起始 LSN 给错了。如果startLsn取的是「当前最新 LSN」，而全量阶段耗时较长，这期间产生的变更在 LSN 上小于最新值，会被跳过。正确的取法是全量开始之前取一次，而不是全量结束时取一次。
第三个方向，账号权限不够。解析会话需要单独的权限，普通只读账号调START_LOGMNR会报错，但错误信息不一定直白，容易误判成没数据。
5.3 现象：大事务把 TaskManager 打挂
现象是某天凌晨跑批的时候，TaskManager 频繁 OOM 重启，堆 dump 里全是字符串。
原因是COMMITTED_DATA_ONLY会把整个事务的记录攒到提交时一起输出，批量更新几百万行的事务，一次性吐出几百万条记录，batch-size再多也没用，因为它是事务级输出而不是记录级。
解决分两步。一是把解析连接器的并发度提上去，按表分发，把单表大事务的压力摊开；二是把batch-size和解析缓冲区的上限做成硬约束，超过阈值的事务走「丢弃 + 告警」的分支，让下游走一次全量重刷。听起来不优雅，但比反复重启强。
5.4 现象：时间字段全部偏移八小时
这个坑很经典，属于玄学级别。现象是源库DATETIME字段2025-01-01 10:00:00，落到下游变成2025-01-01 02:00:00。
原因是链路上有三处时区设置，JDBC 连接串没指定、Flink 作业的table.local-time-zone没指定、落地库的会话时区也没指定，任何一处默认值不同就会偏。日志解析出来的TIMESTAMP是数据库本地时间，中间经过 JVM 默认时区转换成Timestamp对象，再写出去时又按另一个时区格式化。
解决办法是把三处都显式设成同一个值：JDBC 连接串加时区参数，Flink 作业配置里固定table.local-time-zone，落地端连接也设一致。字符串类型的时间字段不走时区转换，如果业务能接受，用VARCHAR存时间反而更省心。
5.5 现象：DDL 之后作业直接报字段不存在
现象是业务加了一列之后，同步任务开始报Column not found，重启也恢复不了。
原因是解析会话用了DICT_FROM_ONLINE_CATALOG，读历史日志时用的是「当前」数据字典。历史记录里那个RS_ID对应的列数和现在的表结构对不上，聚合器按新结构去解析旧记录，直接越界。
解决办法是在 DDL 发生时主动结束当前解析会话并重建，让新旧字典的边界和 DDL 边界对齐。同时把DICT_FROM_ONLINE_CATALOG换成导出字典文件的方式，历史日志用历史字典解析，这个问题从根上消失，代价是每次结构变更都要重新导一次字典。
6. 验收与答辩：怎么证明这条链路真的实时而且不丢数
方案写得再漂亮，最后都要落在「怎么证明它是对的」。我一般准备四组验证，每组都有可观测的通过标准：
| 验收项 | 验证方法 | 通过标准 |
|---|---|---|
| 不丢数 | 源端插入 10 万条带连续编号的数据，等同步完成后比对两端count与sum(编号) | 完全一致，差值 0 |
| 不重复 | 下游按主键做count与count(distinct)对比 | 两个值相等 |
| 实时性 | 源端写入时带上毫秒时间戳，下游记录到达时间 | P99 延迟在秒级，无持续堆积 |
| 断点恢复 | 同步过程中手动 kill 一个 TaskManager | 恢复后数据完整，重复量在可接受范围 |
跑这四组的时候有个技巧：拿一张小表专门做「故障演练表」，反复 kill 任务、改 checkpoint 间隔、切换归档清理策略，把所有异常路径走一遍。评审或者甲方提问时，你手里有这张表的运行记录，比讲十遍架构图管用。
答辩环节最常被问到的三个问题，提前准备好答案。第一个是「为什么不用现成的连接器」，答案是官方没覆盖达梦，社区实现不成熟，自研才能控制位点和事务还原逻辑。第二个是「日志文件被清理了怎么办」，答案是监控归档保留时长与 checkpoint 间隔的比值，低于阈值告警，极端情况走全量重刷。第三个是「和业务方怎么约定」，答案是明确告知下游「至少一次」语义，落地端必须幂等，这个约定要写进接口文档里。
还有个容易被忽略的验收细节：把dm_lsn这一列暴露给运维。出问题时不用翻 Flink 日志，直接查下游表里某条数据的dm_lsn，再去源端归档里比对，十分钟能定位到是哪一段回放出了问题。这个小设计在真实项目里救过我好几次。
从那以后我每次做日志解析类的同步，都强制先把「位点怎么存、重启从哪读、重复怎么消」这三件事写成一句话贴在需求文档第一页，写不出来就先别写代码。希望这套拆解能帮到你。
本文还有配套的精品资源，点击获取
