简介围绕 FlinkCDC 与达梦数据库的日志级实时同步场景这套资料面向 Flink/Java 开发者和实时数仓、数据集成工程师重点解决达梦 CDC 连接器接入、变更数据捕获、作业配置与 Java/SQL 双方式同步的实现参考问题。压缩包共 315 个文件、约 341.71MB以 jar 依赖与编译产物为主辅以 xml、properties 配置java、class 示例代码以及 sql 脚本和少量项目文件便于对照工程结构、连接参数和运行依赖。已有 776 人学习下载说明其在达梦实时同步主题下具有一定参考热度。内容包含 CDC 主作业、Flink SQL 同步示例、反序列化与统计工具等代码骨架可用于搭建本地验证环境、理解变更流转换链路并参考连接地址、端口、账号配置及常见排错思路适合需要快速落地达梦日志同步方案的中高级开发者。1. 日志同步真正的门槛不是能不能读到而是断点能不能续很多人验证 FlinkCDC 接达梦的方式很朴素源库插一条下游看有没有。这条链路能通就敢往生产上推。但生产环境的考验不在这。考验在同步跑了三周之后突然断网、在并发从 1 调到 8、在源库半夜加了个字段、在某个大表快照跑到一半时主库切了归档。这些场景下能不能读到早就不是问题位点还能不能对上才是。FlinkCDCdameng 这类组合真正值钱的地方是把全量快照和增量日志缝成一条可恢复的流水线而不是那个能跑通的 demo。下面把这条流水线拆开看。2. 达梦的日志底座REDO 与归档怎么被还原成行级变更CDC 的本质是从存储引擎的副作用里倒推出业务语义。数据库写数据时留下的不是UPDATE t SET a1而是一串页号、偏移、前像后像。要把物理日志还原成逻辑变更中间隔着一整层解码工作。这一层做得好不好直接决定了后面 Flink 算子拿到的是干净的 ChangeEvent 还是一堆需要二次猜测的残片。2.1 REDO 日志里到底存了什么为什么不能直接读达梦的 REDO 日志和大多数商业库同构记录的是数据块级别的变更向量一条业务UPDATE可能对应多条 REDO 记录UNDO 段、数据段、索引段各一份也可能被拆到不同的日志块里。更麻烦的是日志里只有段号、页号、槽位号以及变更前后的字节快照没有列名、没有表名。这意味着直接读日志文件是行不通的——你拿到的是一堆objd1024, slot7, op2这样的三元组得先反查数据字典才知道 1024 是哪张表、第 7 个槽位对应哪个列、列的类型是VARCHAR(64)还是DECIMAL(18,4)。数据字典本身也存在系统表里也会被 DDL 改所以字典还有版本概念。常见的实现思路是让数据库自己提供解码接口而不是让同步工具去逆向日志格式。这样做的好处是版本兼容性由数据库厂商兜底坏处是接口能力受限——比如某些库的解码接口不支持连续挖掘只能按区间一段段挖这直接影响实时性。2.2 LogMiner 式的解码链路从归档文件到结果集达梦提供了和同类商业库形态接近的日志挖掘能力典型流程是三步注册日志文件、启动挖掘、查询挖掘结果视图。下面是核心调用的骨架。-- 第一步把需要挖掘的日志文件登记进挖掘列表 -- NEW 表示清空并新建列表后续文件用 ADDFILE 追加 BEGIN DBMS_LOGMNR.ADD_LOGFILE( LOGFILENAME /dm/arch/ARCHIVE_LOCAL1_0x5A3F_20240515.log, OPTIONS DBMS_LOGMNR.NEW ); END; / -- 第二步指定 SCN 区间并启动挖掘 -- COMMITTED_DATA_ONLY 只返回已提交事务避免读到脏变更 -- DICT_FROM_ONLINE_CATALOG 用当前在线字典DDL 频繁的场景应改为独立字典文件 BEGIN DBMS_LOGMNR.START_LOGMNR( STARTSCN 128450331, ENDSCN 128450980, OPTIONS DBMS_LOGMNR.DICT_FROM_ONLINE_CATALOG DBMS_LOGMNR.COMMITTED_DATA_ONLY ); END; / -- 第三步读取挖掘结果 SELECT SCN, COMMIT_SCN, TIMESTAMP, OPERATION, SEG_OWNER, TABLE_NAME, ROW_ID, SQL_REDO, CSF FROM V$LOGMNR_CONTENTS WHERE SEG_OWNER BIZ ORDER BY SCN, RS_ID, SSN;这三个参数值得单独说。STARTSCN/ENDSCN是挖掘窗口窗口开太大内存扛不住开太小一条跨区间的长事务会被截断——一条事务的变更散落在窗口两侧时COMMITTED_DATA_ONLY会让它整条消失而不是报错这是最阴的一类丢数据。DICT_FROM_ONLINE_CATALOG在源表刚加过列时会解析出错误的列位置因为在线字典已经是新版本而日志是旧版本写的。V$LOGMNR_CONTENTS返回的OPERATION是INSERT/UPDATE/DELETE/DDL这类粗粒度标签SQL_REDO是拼接出来的伪 SQLCSF标记这条记录是不是继续读下一个日志块才能凑齐。真正拿来做数据同步时不建议解析SQL_REDO字符串——字段里出现单引号、换行、多字节字符都会让拼接语法崩掉。可靠做法是配合COL1、COL2… 这类列值字段按数据字典的列顺序取值。2.3 一次变更在 Flink 侧长什么样挖掘结果集到 Flink 的DataChangeEvent之间还有一层转换这层决定了后续 schema 演进好不好做。下面是一段接近真实的转换骨架。// 把一行 V$LOGMNR_CONTENTS 记录转成 Flink CDC 的变更事件 public DataChangeEvent toChangeEvent(LogmnrRow row, TableSchema schema) { // 1. 用 ROW_ID 定位到具体行作为下游 upsert 的主键兜底 String rowId row.getRowId(); // 2. 按数据字典的列顺序取列值不要按 SQL_REDO 拆 // schema.getColumns() 的顺序必须和挖掘时的字典版本一致 Object[] values new Object[schema.getColumns().size()]; for (int i 0; i values.length; i) { values[i] row.getColumn(i 1); // COL1 从 1 开始编号 } // 3. OPERATION 映射到 RowKind RowKind kind switch (row.getOperation()) { case INSERT - RowKind.INSERT; case DELETE - RowKind.DELETE; case UPDATE - RowKind.UPDATE_AFTER; // 前像在 COL*_BEFORE 里 default - throw new IllegalStateException(unsupported op: row.getOperation()); }; // 4. 用 COMMIT_SCN 作为事件时间保证乱序事务能按提交顺序重放 return new DataChangeEvent( new TableId(row.getOwner(), row.getTableName()), row.getCommitScn(), kind, GenericRowData.of(values) ); }关键点在第四步事件时间必须用COMMIT_SCN而不是SCN。一个事务里如果有 10 条变更它们的SCN各不相同但COMMIT_SCN一致。用提交点排序才能保证下游看到的顺序和源库的可见顺序一致用单条SCN排序会看到事务 A 的第 2 条排在事务 B 的第 1 条前面这种源库从未出现过的中间状态。这个细节在同步有唯一约束的表时会直接表现为下游写入失败。注意ROW_ID在行迁移、分区表、段收缩之后会失效。把它当主键兜底只适合短窗口内的去重不能作为长期唯一标识。3. 全增量一体化快照分片和增量日志怎么在同一份数据上不重不漏CDC 最难的不是读日志是把历史的存量和正在发生的增量拼起来。老方案是先全量导出、再切换到日志切换窗口内的变更要么停写业务、要么靠额外一张变更表兜底。Flink CDC 的增量快照思路是把这两件事并行起来做不需要停业务代价是要处理同一条记录既出现在快照里又出现在日志里。3.1 水位线机制用两个 SCN 圈出一个必须去重的区间增量快照的核心是两条水位线。开始快照之前先记下当前日志位点叫 low watermark所有分片读完之后再记一次当前位点叫 high watermark。这两个点之间的日志变更和快照读到的数据存在重叠。重叠的原因很直白某个分片在 T1 时刻读到了主键区间[1, 10000)这个分片读了 30 秒期间业务把主键 5000 那一行改了。日志里会有这条 UPDATE快照里读到的也是改后的值——如果两者都往下游发下游会收到两条等价的更新对 upsert 语义无害但对 append 语义的表就是重复数据。所以处理规则是low watermark 之前的日志说明对应数据已经被快照覆盖跳过low 到 high 之间的日志按主键去重high watermark 之后正常消费。这条规则听着简单落地时最容易被忽略的是——去重需要主键。没主键的表在这个模型里没有解只能退化成全表重读或者放弃同步。3.2 Split 的生命周期为什么一个分片要跨四个状态Flink CDC 把一张表切成若干分片chunk每个分片在 SourceReader 里走一遍状态机。抽象出来是四个阶段。// 分片状态机每个 chunk 独立推进失败可单独重试 public enum SplitState { PENDING, // 已分配但未开始记录下界 SNAPSHOT, // 正在做快照读记录当前读到的位置 BINLOG, // 快照读完开始消费 (low, high] 区间的日志 FINISHED // 已到达 high watermark可回收 } // 快照读的 SQL 骨架左闭右开避免相邻分片重叠 String snapshotSql SELECT * FROM table WHERE pkColumn ? AND pkColumn ? ORDER BY pkColumn;左闭右开这个约定不是随便定的。如果写成 ? AND ?相邻两个分片在主键恰好等于边界值时会各读一次下游重复如果写成 ? AND ?边界那一行永远读不到。用配配合分片器按主键的实际分布切点才能保证覆盖完整且不重叠。分片边界怎么切也有讲究。等宽切分在主键稀疏或有大量空洞时会出现严重的倾斜——某个分片只有 3 行另一个有两百万行整个作业的进度被最慢的那个拖住。常见的改进是先采样统计主键分布再按行数均分代价是多一次轻量查询。3.3 一个分片的完整时序把上面两节合起来单个分片的时间线大致是这样阶段动作记录的状态失败后行为分配SplitEnumerator 下发[lower, upper)分片边界重新入队快照按主键区间分批 SELECT已读到的最大主键从该主键续读切换快照读完切到日志消费分片的 high watermark重放 (low, high]收尾追平 high watermarkFINISHED无需处理第三行失败后重放 (low, high]是这套设计的关键取舍它不追求恰好读一次日志而是允许日志被多读靠主键去重把重复消化掉。这也是为什么整个模型强依赖主键——没有主键去重就退化成无法判断作业要么报错要么静默产生重复。注意去重状态在 Flink 里是有状态算子维护的状态大小和 (low, high] 区间的变更量正相关。如果快照跑得特别久比如几小时这个区间会很大去重状态可能把 TaskManager 内存顶满。4. Checkpoint 与位点Flink 的容错语义怎么落到 SCN 上到这一步变更已经能读出来、也能和快照拼起来了。剩下的问题是作业挂掉重启从哪里接着读答案听起来是从 checkpoint 里恢复但 checkpoint 里存的到底是什么、恢复时先补日志还是先补快照这两个问题答不上来精确一次就是一句口号。4.1 位点状态里必须存什么一个能正确恢复的位点状态至少要覆盖三类信息全局日志位点、每个分片的进度、以及尚未提交的去重窗口。少任何一项恢复后要么丢数据要么重复。// 快照进 checkpoint 的位点结构简化 public class DmOffsetState implements Serializable { // 1. 全局位点下一次挖掘的起点 SCN private long nextScn; // 2. 已提交到的位置用于判断哪些日志可以安全跳过 // RS_ID SSN 组合才能唯一定位一条日志记录 private String redoRsId; private int redoSsn; // 3. 分片进度key 是分片 idvalue 是该分片已读到的最大主键 private MapString, Object splitProgress; // 4. 去重窗口的下界low watermark早于它的日志可丢弃 private long lowWatermarkScn; }nextScn和redoRsId/redoSsn是互补关系。只用 SCN 不够——同一个 SCN 下可能有多条日志记录重放时从 SCN 起点开始会把已经发过的记录再发一遍只用 RS_ID 也不够因为日志文件切换后 RS_ID 可能复用。两个组合起来才能精确定位我消费到哪了。分片进度之所以要单独存是因为一个作业里可能同时有几十个分片在跑每个分片的进度不同。如果只存一个全局进度恢复时所有分片都得从头再来一遍快照大表场景下这个代价不可接受。4.2 两阶段提交与位点对齐Flink 的 checkpoint 是分布式的Source 拍快照、算子刷状态、Sink 预提交、最后统一提交。位点和下游事务之间要严格对齐否则会出现位点前进了但下游没写进去。常见的错误做法是在 Source 的snapshotState里直接把位点推到最新——这等于宣告我已经消费完了但下游 Sink 的事务可能还没提交。正确的做法是位点只记录已确认完成的部分也就是两阶段提交里的 pre-commit 阶段记录、commit 阶段之后才允许被后续 checkpoint 覆盖。这个语义在日志侧的具体含义是位点只能推进到最近一次成功提交的事务边界不能推进到某条中间记录。日志是按事务组织的一条事务的所有变更要么全发要么全不发位点卡在事务中间会让恢复后的重放产生半个事务。4.3 恢复流程先补日志还是先补快照恢复时的顺序不是任选的。正确的顺序是先根据 checkpoint 里的nextScn把日志挖掘窗口对齐到那个 SCN把 (lowWatermark, nextScn] 之间的日志重新挖一遍然后再检查splitProgress把还没跑完的分片从记录的进度继续。顺序反了会怎样如果先补快照快照读到的数据可能已经包含了 (low, nextScn] 区间的变更之后补日志又发一遍去重窗口已经被 checkpoint 清理掉了重复无法消除。还有个隐性约束日志文件可能在作业挂掉的这段时间里被归档策略删掉了。恢复时如果发现nextScn对应的归档文件已经不存在只能降级成全量重同步。这不是设计缺陷是日志型 CDC 的固有边界——日志的保留窗口就是你的最大故障恢复窗口。上线前把归档保留时长和业务能接受的最长中断时间对齐是必做的一步。5. 避坑与排查DM 日志同步最容易翻车的五个现场5.1 现象作业跑着跑着没报错但下游数据停止更新原因通常是挖掘窗口没有向前推进。STARTSCN/ENDSCN是调用方传入的如果作业里用当前时间减去固定偏移来算窗口源库空闲时间一长窗口内的日志没有新记录挖掘返回空结果集而代码把空结果当成没有变更继续循环位点却推进了。解决方式是让窗口跟随实际挖掘到的最大 SCN 前进而不是跟随墙钟时间。日志挖掘的结果里COMMIT_SCN的最大值才是唯一可信的推进依据没有结果就原地等待不要前移。5.2 现象大字段CLOB / BLOB同步过来是空的或者截断原因有两层。一是V$LOGMNR_CONTENTS对超大字段的返回受挖掘参数限制超过阈值的字段可能只给出占位二是 JDBC 读 LOB 需要显式流式处理用getString拿 CLOB 在某些驱动版本上会返回前 N 个字符而不报错。解决方式是判断字段长度超限时用getCharacterStream/getBinaryStream读并在转换层保留原始类型不降级成字符串。同时在上线前用一张含 1MB 文本的表做一次端到端验证别等业务反馈才发现。5.3 现象加了新列之后同步过来的数据整体错位原因是用在线字典解析旧日志。DDL 加列后在线字典的列顺序变了而日志里旧事务写入的列位置还是老顺序两者一错位后面的列全部串行。解决办法是挖掘时使用与日志版本匹配的字典快照或者干脆在 DDL 发生后主动触发一次 schema 重新加载并短暂重放。更稳妥的做法是把 DDL 变更也当成一种事件往下游发让下游先应用 schema 变更再应用数据变更。5.4 现象并发调大后主库出现明显的锁等待原因是快照读用了加锁读或者长事务隔离级别。某些实现为了拿到一致性快照会在分片读上加FOR UPDATE或者开可重复读事务分片越多、事务越长主库上的锁持有时间越久。解决办法是把快照读降级成普通读靠 low/high watermark 的去重机制来保证语义而不是靠数据库锁。这也是增量快照相比老式全量导出的优势——它把一致性责任从数据库转移到了 Flink 的有状态算子。5.5 现象重启后下游出现大量重复数据原因一般出在去重窗口的清理时机。如果 checkpoint 在去重状态还没刷盘时就把 low watermark 往前推了恢复后这段窗口的日志会被重新消费而无从去重。解决办法是把 low watermark 的推进和去重状态的快照绑在同一个 checkpoint 里两者要么一起生效要么一起回滚。另外在 Sink 侧做一层幂等兜底按主键 upsert能挡住绝大多数这类问题。6. 设计取舍为什么这套机制做不到完美黑匣子把前面几章串起来看会发现整套设计里有几个绕不开的取舍它们不是实现得不够好而是问题本身没有完美解。第一个取舍是实时性和事务完整性。挖掘窗口开得小延迟低但跨窗口的长事务会被切断窗口开得大事务完整但内存占用和延迟都上去了。实际系统里通常用一个可配置的窗口上限超过就强制提交接受极少数长事务需要人工介入。第二个取舍是状态大小和恢复速度。去重窗口越大恢复时能跳过的日志越多但常驻状态越大。反过来窗口小则状态轻恢复时要重放更多日志。这个平衡点和大表的更新频率强相关没有通用值。第三个取舍是日志保留窗口和故障恢复能力。这可能是最容易被低估的一个归档保留可容忍的最长中断代价1 天1 天内恢复可续传磁盘占用小7 天一周内恢复可续传需要额外存储规划30 天月度级故障可续传存储成本显著上升超过保留窗口的故障只能全量重同步。这不是工具的能力边界是日志只记录有限历史这个物理事实决定的。任何声称能无限续传的方案本质上都在别处偷偷存了一份历史。至于那句CDC 只需要配一下就能跑我踩过的坑足够让我不信它。从那以后我每次接一个新的源库都强制走一遍流程先确认归档模式开着且保留时长够用再确认目标表都有主键然后用一张带大字段和频繁更新的表压一遍快照加增量最后手动 kill 一次作业验证恢复后不重不漏。这三步走完才敢往生产上放。希望帮到你。本文还有配套的精品资源点击获取