简介本资源面向需要在 Flink 生态中对接达梦数据库的 Java 与 SQL 开发者聚焦基于日志的实时变更数据捕获CDC场景帮助解决达梦数据向实时数仓、事件驱动应用同步的问题。包内共 315 个文件以 263 个 jar 依赖为主辅以 xml 配置、java 源码、class 编译文件、properties 参数文件及 sql 脚本等压缩包约 341.71MB覆盖连接器依赖、作业配置与示例代码等模块。已有 765 人学习下载说明该方向具备一定关注度。资源包含 FlinkDMCDC、DMFlinkSQL 等核心类与反序列化实现读者可据此理解达梦日志解析、连接参数配置及 Java/SQL 两种同步方式的落地思路适合用于实时报表、数据监控与告警等场景的参考与二次开发。1. 达梦日志实时同步为什么 FlinkCDC 这条路值得走一遍第一次接到“把达梦数据库的变更实时同步到下游”这个需求时我脑子里第一反应是写触发器加中间表再配个定时任务轮询。这套方案能跑但延迟高、对源库压力大表结构一改就得跟着改触发器维护成本肉眼可见地涨。后来换成 FlinkCDC 基于日志的采集方式才真正把“实时”两个字落地——源库几乎无感延迟压到秒级下游拿到的是干净的变更流。达梦数据库作为国产关系型数据库里的主力选手在政务、金融、能源这些场景铺得很广但围绕它的实时同步方案一直不算丰富。FlinkCDC 社区版原生并不直接支持达梦需要走自定义 Connector 或者借助兼容模式。这篇笔记就把我踩过的路拆开讲达梦的日志机制怎么开、FlinkCDC 怎么接、增量快照怎么配、哪些参数一动就翻车。适合正在做异构数据同步、CDC 选型或者手里有达梦库要往 Kafka、Doris、Elasticsearch 灌数据的同学。2. 达梦日志机制与 FlinkCDC 接入原理先把底层逻辑捋直2.1 达梦的归档日志与逻辑日志不是一回事达梦数据库的日志体系里跟 CDC 采集最相关的是归档日志ARCHIVELOG和逻辑日志Logical Log。很多人一上来就找“达梦的 binlog”这是被 MySQL 的思路带偏了。达梦没有 binlog 这个叫法它的物理日志是 REDO 日志记录的是页级变更直接解析难度大真正对外做逻辑复制用的是逻辑日志通过配置可以输出类似 row-based 的变更记录。达梦从 DM8 开始提供了逻辑日志功能需要在数据库层面开启ENABLE_LOGIC_LOG参数并且对需要同步的表额外配置。逻辑日志记录的是行的插入、更新、删除操作包含前镜像和后镜像这正是 FlinkCDC 需要的输入格式。归档日志则是保证日志不被覆盖、支持按时间点恢复的基础做 CDC 时必须先开归档否则日志循环写会把还没采集的变更冲掉。常见做法是先确认达梦版本在 DM8 及以上然后检查ARCHIVE_MODE是否为 1ENABLE_LOGIC_LOG是否为 1。这两个开关不开后面所有 FlinkCDC 配置都是白搭。2.2 FlinkCDC 的增量快照框架怎么套到达梦上FlinkCDC 的核心是增量快照算法先对全表做一次一致性快照同时记录快照开始时的日志位点快照完成后从该位点继续消费日志保证不丢不重。这套机制在 MySQL、PostgreSQL 上由官方 Connector 实现达梦没有官方支持所以需要自己实现SourceFunction或者用 FlinkCDC 的IncrementalSource接口。我一般会走两条路一是基于 FlinkCDC 的SourceAPI 写一个达梦的Dialect和SplitReader把达梦的逻辑日志解析成SourceRecord二是如果团队不想维护 Connector可以用达梦自带的DMHS达梦数据同步软件先把日志推到 Kafka再用 FlinkCDC 的 Kafka Connector 接。前者更轻量后者更稳但多一跳。下面是一个自定义达梦 Source 的骨架代码展示怎么把逻辑日志的位点和快照切分串起来// 达梦增量快照 Source 核心骨架 public class DamengIncrementalSourceC extends IncrementalSourceC, DamengSplit { Override public void snapshotState(FunctionSnapshotContext context) throws Exception { // 把当前消费到的逻辑日志 LSN 写入状态后端 // 注意LSN 必须和快照 barrier 对齐否则恢复时会重复或丢数据 checkpointedState.clear(); checkpointedState.add(currentLsn); } Override public void initializeState(FunctionInitializationContext context) throws Exception { // 从状态恢复时拿到上次的 LSN从该位置继续消费 // 如果是首次启动LSN 设为快照开始时的最小活跃日志位点 if (context.isRestored()) { currentLsn checkpointedState.get().iterator().next(); } else { currentLsn damengDialect.getMinLogLsn(); } } }这段代码的关键在snapshotState和initializeState的对称性。达梦的逻辑日志位点用 LSN 表示每次 checkpoint 必须把当前 LSN 存下来恢复时从该 LSN 继续。如果 LSN 存早了会重复消费存晚了会丢变更。我见过有人把 LSN 存在外部 Redis 里结果 checkpoint 失败时 Redis 已经更新Flink 回滚后 LSN 对不上直接导致数据断层。所以状态必须走 Flink 自己的状态后端别自己造轮子。2.3 达梦逻辑日志的开启与验证步骤在动手写 Connector 之前先把达梦这边的日志配置跑通。以下命令在达梦的disql工具里执行-- 1. 开启归档模式需要重启数据库生效 ALTER DATABASE MOUNT; ALTER DATABASE ARCHIVELOG; ALTER DATABASE OPEN; -- 2. 开启逻辑日志 SP_SET_PARA_VALUE(1, ENABLE_LOGIC_LOG, 1); -- 3. 对需要同步的表启用逻辑日志 ALTER TABLE SYSDBA.ORDERS ENABLE LOGICAL LOG; -- 4. 验证逻辑日志是否生效 SELECT TABLE_NAME, LOGICAL_LOG FROM DBA_TABLES WHERE TABLE_NAME ORDERS;参数说明SP_SET_PARA_VALUE的第一个参数 1 表示动态参数立即生效第二个参数是参数名第三个是值。ENABLE_LOGIC_LOG开启后达梦会为配置了逻辑日志的表生成额外的日志记录。ALTER TABLE ... ENABLE LOGICAL LOG必须对每张要同步的表单独执行没有库级别的批量开关。验证时如果LOGICAL_LOG列返回Y说明表级别配置成功。然后可以往表里插一条数据去达梦的日志目录下看是否有对应的逻辑日志文件生成。达梦的逻辑日志默认和归档日志放在一起路径由LOG_PATH参数控制。注意开启归档和逻辑日志会增加磁盘 IO 和存储占用生产环境务必先评估日志保留策略别让日志把磁盘撑爆。3. FlinkCDC 达梦 Connector 的配置与调优参数怎么设才不翻车3.1 连接器依赖与 Flink 版本匹配达梦没有官方 FlinkCDC Connector所以要么自己编译要么用社区里有人放出的非官方包。我一般建议自己基于 FlinkCDC 的flink-connector-debezium模块改因为达梦的逻辑日志格式和 Debezium 支持的数据库都不一样直接套用会解析失败。依赖上Flink 版本和 FlinkCDC 版本必须对齐。比如 Flink 1.17 配 FlinkCDC 2.4Flink 1.18 配 FlinkCDC 3.0。达梦的 JDBC 驱动用DmJdbcDriver18.jar对应 JDK 8 及以上。把驱动放到 Flink 的lib目录Connector 的 jar 放到usrlib或者通过-C参数指定。!-- pom.xml 中引入 FlinkCDC 基础模块 -- dependency groupIdcom.ververica/groupId artifactIdflink-connector-debezium/artifactId version2.4.2/version /dependency dependency groupIdcom.dameng/groupId artifactIdDmJdbcDriver18/artifactId version8.1.3.62/version scopesystem/scope systemPath${project.basedir}/lib/DmJdbcDriver18.jar/systemPath /dependency达梦驱动不在 Maven 中央仓库所以用systemscope 本地引入。版本号根据你实际安装的达梦版本选8.1.3.62 是 DM8 比较常见的一个小版本。如果编译时报ClassNotFoundException: dm.jdbc.driver.DmDriver八成是驱动没打进 fat jar检查maven-shade-plugin的配置。3.2 核心配置项从 checkpoint 到日志位点一个能跑的达梦 CDC 任务配置项分三块Flink 作业级、Connector 级、达梦数据库级。下面这张表是我调优时反复改的参数参数作用推荐值踩坑点execution.checkpointing.intervalcheckpoint 间隔10s太短会增加达梦日志读取压力太长恢复时重复数据多scan.incremental.snapshot.chunk.size快照分片行数8096大表调大小表调小影响快照阶段并行度scan.snapshot.fetch.size单次拉取行数1024达梦 JDBC 对超大 fetch 支持不好别超过 4096dameng.log.mining.interval日志挖掘间隔500ms太短空转耗 CPU太长延迟高dameng.log.lsn.startup.mode启动位点模式latest可选 earliest/initial首次全量用 initialscan.incremental.snapshot.chunk.size这个参数特别关键。达梦在快照阶段如果单次查询返回行数太多容易触发游标超时或者内存溢出。我一般按表大小算小于 100 万行的表用 8096大于 1000 万行的表调到 32768同时把scan.snapshot.fetch.size控制在 2048 以内。dameng.log.lsn.startup.mode设成initial时FlinkCDC 会先做全量快照再从快照开始时的 LSN 消费增量。设成latest则跳过全量只从当前最新 LSN 开始适合已经做过全量初始化、只想接增量的场景。设错了会导致要么重复全量、要么丢历史变更。3.3 提交作业与验证数据一致性配置写完后用 Flink CLI 提交作业# 提交达梦 CDC 作业到 Flink 集群 ./bin/flink run \ -c com.example.DamengCdcJob \ -C file:///opt/flink/lib/flink-connector-dameng-cdc-1.0.jar \ -C file:///opt/flink/lib/DmJdbcDriver18.jar \ /opt/jobs/dameng-cdc-job.jar \ --dameng.host 192.168.1.100 \ --dameng.port 5236 \ --dameng.username SYSDBA \ --dameng.password SYSDBA123 \ --dameng.database DAMENG \ --table-list SYSDBA.ORDERS,SYSDBA.USERS \ --sink.kafka.bootstrap kafka:9092-C参数用来把 Connector 和驱动加到作业的 classpath避免和 Flink 自带 jar 冲突。--table-list支持逗号分隔的多表达梦的表名默认大写写小写会找不到表。提交后去 Flink Web UI 看numRecordsIn指标如果一直是 0先检查达梦的逻辑日志有没有新记录生成。验证一致性时我习惯在源库插一条数据然后去下游 Kafka 消费看能不能在秒级内拿到对应的 JSON。再更新、删除各做一次确认 op 字段分别是I、U、-D。如果更新操作只拿到后镜像没有前镜像检查达梦逻辑日志的LOGICAL_LOG_IMAGE参数是不是设成了AFTER改成BOTH才能拿到完整变更。4. 避坑与排查达梦 CDC 最常见的五类翻车现场4.1 归档日志被覆盖导致增量断层现象任务跑了一周某天重启后报LSN not found下游数据出现几小时的空洞。原因达梦的归档日志默认保留策略是按空间或时间如果日志生成速度快、保留窗口短FlinkCDC 还没消费到的日志就被删了。尤其是夜间批量作业集中写入时日志量暴涨很容易触发清理。解决把ARCH_SPACE_LIMIT调大或者设置ARCH_KEEP_TIME至少覆盖两个 checkpoint 周期。更稳妥的做法是单独挂一块盘放归档日志别和业务数据盘抢空间。重启任务前先确认起始 LSN 还在归档里不在的话只能重新全量。4.2 逻辑日志未对表启用导致变更丢失现象全量数据同步正常但后续的 insert/update 完全采集不到下游数据一直不变。原因只开了库级别的ENABLE_LOGIC_LOG忘了对具体表执行ALTER TABLE ... ENABLE LOGICAL LOG。达梦的逻辑日志是表级开关库级参数只是总闸。解决写个脚本批量对需要同步的表启用逻辑日志并且把这条检查加进上线前的 checklist。可以用SELECT TABLE_NAME FROM DBA_TABLES WHERE LOGICAL_LOG N找出漏配的表。4.3 快照阶段达梦连接数被打满现象任务启动后快照跑了一半卡住达梦监控显示连接数飙升到上限业务查询开始超时。原因scan.incremental.snapshot.chunk.size设得太小导致分片数过多每个分片开一个 JDBC 连接并行拉取。Flink 的并行度乘以分片数连接数直接爆炸。解决把 chunk size 调大减少分片数同时限制 Flink 作业的并行度别超过达梦MAX_SESSIONS的 30%。我一般会在达梦侧单独建一个 CDC 专用用户给它设连接数上限避免影响业务。4.4 时间戳类型字段解析成乱码现象下游 Kafka 里的TIMESTAMP字段变成一串数字或者null。原因达梦的TIMESTAMP类型在逻辑日志里是二进制格式自定义 Connector 如果没有正确实现DebeziumDeserializationSchema就会解析失败。达梦的 timestamp 精度到微秒和 Flink 的TimestampType默认毫秒精度不匹配。解决在反序列化时把达梦的微秒值除以 1000 转成毫秒或者用LocalDateTime直接接。别用java.sql.Timestamp的getTime()那个方法在达梦驱动里有精度丢失的坑。4.5 checkpoint 超时导致任务反复重启现象Flink UI 上 checkpoint 一直失败报Checkpoint expired before completing任务进入重启循环。原因达梦日志挖掘线程和 checkpoint 线程抢锁或者状态后端写入太慢。达梦的逻辑日志读取是单线程的如果日志量大checkpoint 时等待状态对齐的时间会超过超时阈值。解决把execution.checkpointing.timeout从默认 10 分钟调到 15 分钟同时把状态后端从MemoryStateBackend换成RocksDBStateBackend减少大状态下的 GC 压力。如果还不行检查达梦的LOG_BUFFER_SIZE是不是太小适当调大能加快日志读取。5. 进阶技巧用 FlinkCDC 做达梦到 Doris 的整库同步5.1 整库同步的表结构映射与自动建表单表同步跑通后下一步通常是整库同步。达梦到 Doris 的链路里最烦的是表结构映射达梦的VARCHAR2对应 Doris 的VARCHARNUMBER对应DECIMALCLOB对应STRING。如果手动建表几十张表能建到怀疑人生。我一般用 FlinkCDC 的SchemaEvolution能力配合 Doris 的auto-create-table参数让下游自动建表。达梦这边需要在 Connector 里实现SupportsSchemaEvolution接口把达梦的 schema 变更转成 Flink 的SchemaChangeEvent。Doris 的 Flink Connector 收到后会自动执行CREATE TABLE或ALTER TABLE ADD COLUMN。// 达梦 Schema 变更捕获示例 public class DamengSchemaChangeEvent implements SchemaChangeEvent { private final String tableName; private final String ddl; Override public String getDdl() { // 把达梦的 DDL 转成 Doris 兼容的语法 // 例如VARCHAR2(255) - VARCHAR(255) return ddl.replace(VARCHAR2, VARCHAR) .replace(NUMBER, DECIMAL); } }这段代码的关键在 DDL 转换。达梦的VARCHAR2在 Doris 里没有对应类型必须转成VARCHARNUMBER不带精度时默认转DECIMAL(38, 10)但 Doris 的 DECIMAL 最大精度是 38超过会报错。所以转换时要加一层校验精度超限的字段降级成STRING。5.2 分库分表场景下的达梦多源合并有些老系统按业务线拆了多个达梦实例每个实例里又有结构相同的分表。这种场景下FlinkCDC 可以用MySqlSource的多表合并思路把多个达梦 Source 读到的流union起来再统一写入 Doris。// 多达梦实例合并同步 DataStreamSourceRecord stream1 env.fromSource( DamengSource.builder() .host(192.168.1.101) .tableList(SYSDBA.ORDERS_01) .build(), WatermarkStrategy.noWatermarks(), dameng-source-1 ); DataStreamSourceRecord stream2 env.fromSource( DamengSource.builder() .host(192.168.1.102) .tableList(SYSDBA.ORDERS_02) .build(), WatermarkStrategy.noWatermarks(), dameng-source-2 ); stream1.union(stream2) .keyBy(record - record.getTableName()) .sinkTo(dorisSink);union之后必须keyBy表名否则同一张逻辑表的数据会被分散到不同 subtaskDoris 的批量写入会乱序。另外多个 Source 的 checkpoint 是独立对齐的如果某个实例挂了整个作业会重启所有 Source 一起回滚。对可用性要求高的场景建议每个实例单独跑一个作业下游用 Doris 的UNIQUE KEY模型做去重。5.3 监控与延迟度量别等业务投诉才发现同步断了同步链路最怕的不是报错是悄无声息地断了。我习惯在作业里加三个监控指标dameng.log.lag当前 LSN 和最新 LSN 的差值、sink.batch.interval下游写入间隔、checkpoint.durationcheckpoint 耗时。前两个用 Flink 的Gauge暴露第三个用Histogram。达梦这边可以定期查V$LOG_HISTORY看归档日志的生成速度如果生成速度突然下降可能是业务停了也可能是日志写入出了问题。下游 Doris 那边看SHOW LOAD的ETL状态如果长时间处于LOADING说明 Flink 写入批次太大或者 Doris 集群压力高。从那以后我每次上线达梦 CDC 任务都强制走一遍“插-更-删”三连验证再挂上延迟监控跑满 24 小时才敢交给业务。这套流程帮我拦住了至少三次潜在的增量断层。希望帮到你。本文还有配套的精品资源点击获取 FlinkCDC 实时同步达梦数据库：原理、配置与避坑实践
简介本资源面向需要在 Flink 生态中对接达梦数据库的 Java 与 SQL 开发者聚焦基于日志的实时变更数据捕获CDC场景帮助解决达梦数据向实时数仓、事件驱动应用同步的问题。包内共 315 个文件以 263 个 jar 依赖为主辅以 xml 配置、java 源码、class 编译文件、properties 参数文件及 sql 脚本等压缩包约 341.71MB覆盖连接器依赖、作业配置与示例代码等模块。已有 765 人学习下载说明该方向具备一定关注度。资源包含 FlinkDMCDC、DMFlinkSQL 等核心类与反序列化实现读者可据此理解达梦日志解析、连接参数配置及 Java/SQL 两种同步方式的落地思路适合用于实时报表、数据监控与告警等场景的参考与二次开发。1. 达梦日志实时同步为什么 FlinkCDC 这条路值得走一遍第一次接到“把达梦数据库的变更实时同步到下游”这个需求时我脑子里第一反应是写触发器加中间表再配个定时任务轮询。这套方案能跑但延迟高、对源库压力大表结构一改就得跟着改触发器维护成本肉眼可见地涨。后来换成 FlinkCDC 基于日志的采集方式才真正把“实时”两个字落地——源库几乎无感延迟压到秒级下游拿到的是干净的变更流。达梦数据库作为国产关系型数据库里的主力选手在政务、金融、能源这些场景铺得很广但围绕它的实时同步方案一直不算丰富。FlinkCDC 社区版原生并不直接支持达梦需要走自定义 Connector 或者借助兼容模式。这篇笔记就把我踩过的路拆开讲达梦的日志机制怎么开、FlinkCDC 怎么接、增量快照怎么配、哪些参数一动就翻车。适合正在做异构数据同步、CDC 选型或者手里有达梦库要往 Kafka、Doris、Elasticsearch 灌数据的同学。2. 达梦日志机制与 FlinkCDC 接入原理先把底层逻辑捋直2.1 达梦的归档日志与逻辑日志不是一回事达梦数据库的日志体系里跟 CDC 采集最相关的是归档日志ARCHIVELOG和逻辑日志Logical Log。很多人一上来就找“达梦的 binlog”这是被 MySQL 的思路带偏了。达梦没有 binlog 这个叫法它的物理日志是 REDO 日志记录的是页级变更直接解析难度大真正对外做逻辑复制用的是逻辑日志通过配置可以输出类似 row-based 的变更记录。达梦从 DM8 开始提供了逻辑日志功能需要在数据库层面开启ENABLE_LOGIC_LOG参数并且对需要同步的表额外配置。逻辑日志记录的是行的插入、更新、删除操作包含前镜像和后镜像这正是 FlinkCDC 需要的输入格式。归档日志则是保证日志不被覆盖、支持按时间点恢复的基础做 CDC 时必须先开归档否则日志循环写会把还没采集的变更冲掉。常见做法是先确认达梦版本在 DM8 及以上然后检查ARCHIVE_MODE是否为 1ENABLE_LOGIC_LOG是否为 1。这两个开关不开后面所有 FlinkCDC 配置都是白搭。2.2 FlinkCDC 的增量快照框架怎么套到达梦上FlinkCDC 的核心是增量快照算法先对全表做一次一致性快照同时记录快照开始时的日志位点快照完成后从该位点继续消费日志保证不丢不重。这套机制在 MySQL、PostgreSQL 上由官方 Connector 实现达梦没有官方支持所以需要自己实现SourceFunction或者用 FlinkCDC 的IncrementalSource接口。我一般会走两条路一是基于 FlinkCDC 的SourceAPI 写一个达梦的Dialect和SplitReader把达梦的逻辑日志解析成SourceRecord二是如果团队不想维护 Connector可以用达梦自带的DMHS达梦数据同步软件先把日志推到 Kafka再用 FlinkCDC 的 Kafka Connector 接。前者更轻量后者更稳但多一跳。下面是一个自定义达梦 Source 的骨架代码展示怎么把逻辑日志的位点和快照切分串起来// 达梦增量快照 Source 核心骨架 public class DamengIncrementalSourceC extends IncrementalSourceC, DamengSplit { Override public void snapshotState(FunctionSnapshotContext context) throws Exception { // 把当前消费到的逻辑日志 LSN 写入状态后端 // 注意LSN 必须和快照 barrier 对齐否则恢复时会重复或丢数据 checkpointedState.clear(); checkpointedState.add(currentLsn); } Override public void initializeState(FunctionInitializationContext context) throws Exception { // 从状态恢复时拿到上次的 LSN从该位置继续消费 // 如果是首次启动LSN 设为快照开始时的最小活跃日志位点 if (context.isRestored()) { currentLsn checkpointedState.get().iterator().next(); } else { currentLsn damengDialect.getMinLogLsn(); } } }这段代码的关键在snapshotState和initializeState的对称性。达梦的逻辑日志位点用 LSN 表示每次 checkpoint 必须把当前 LSN 存下来恢复时从该 LSN 继续。如果 LSN 存早了会重复消费存晚了会丢变更。我见过有人把 LSN 存在外部 Redis 里结果 checkpoint 失败时 Redis 已经更新Flink 回滚后 LSN 对不上直接导致数据断层。所以状态必须走 Flink 自己的状态后端别自己造轮子。2.3 达梦逻辑日志的开启与验证步骤在动手写 Connector 之前先把达梦这边的日志配置跑通。以下命令在达梦的disql工具里执行-- 1. 开启归档模式需要重启数据库生效 ALTER DATABASE MOUNT; ALTER DATABASE ARCHIVELOG; ALTER DATABASE OPEN; -- 2. 开启逻辑日志 SP_SET_PARA_VALUE(1, ENABLE_LOGIC_LOG, 1); -- 3. 对需要同步的表启用逻辑日志 ALTER TABLE SYSDBA.ORDERS ENABLE LOGICAL LOG; -- 4. 验证逻辑日志是否生效 SELECT TABLE_NAME, LOGICAL_LOG FROM DBA_TABLES WHERE TABLE_NAME ORDERS;参数说明SP_SET_PARA_VALUE的第一个参数 1 表示动态参数立即生效第二个参数是参数名第三个是值。ENABLE_LOGIC_LOG开启后达梦会为配置了逻辑日志的表生成额外的日志记录。ALTER TABLE ... ENABLE LOGICAL LOG必须对每张要同步的表单独执行没有库级别的批量开关。验证时如果LOGICAL_LOG列返回Y说明表级别配置成功。然后可以往表里插一条数据去达梦的日志目录下看是否有对应的逻辑日志文件生成。达梦的逻辑日志默认和归档日志放在一起路径由LOG_PATH参数控制。注意开启归档和逻辑日志会增加磁盘 IO 和存储占用生产环境务必先评估日志保留策略别让日志把磁盘撑爆。3. FlinkCDC 达梦 Connector 的配置与调优参数怎么设才不翻车3.1 连接器依赖与 Flink 版本匹配达梦没有官方 FlinkCDC Connector所以要么自己编译要么用社区里有人放出的非官方包。我一般建议自己基于 FlinkCDC 的flink-connector-debezium模块改因为达梦的逻辑日志格式和 Debezium 支持的数据库都不一样直接套用会解析失败。依赖上Flink 版本和 FlinkCDC 版本必须对齐。比如 Flink 1.17 配 FlinkCDC 2.4Flink 1.18 配 FlinkCDC 3.0。达梦的 JDBC 驱动用DmJdbcDriver18.jar对应 JDK 8 及以上。把驱动放到 Flink 的lib目录Connector 的 jar 放到usrlib或者通过-C参数指定。!-- pom.xml 中引入 FlinkCDC 基础模块 -- dependency groupIdcom.ververica/groupId artifactIdflink-connector-debezium/artifactId version2.4.2/version /dependency dependency groupIdcom.dameng/groupId artifactIdDmJdbcDriver18/artifactId version8.1.3.62/version scopesystem/scope systemPath${project.basedir}/lib/DmJdbcDriver18.jar/systemPath /dependency达梦驱动不在 Maven 中央仓库所以用systemscope 本地引入。版本号根据你实际安装的达梦版本选8.1.3.62 是 DM8 比较常见的一个小版本。如果编译时报ClassNotFoundException: dm.jdbc.driver.DmDriver八成是驱动没打进 fat jar检查maven-shade-plugin的配置。3.2 核心配置项从 checkpoint 到日志位点一个能跑的达梦 CDC 任务配置项分三块Flink 作业级、Connector 级、达梦数据库级。下面这张表是我调优时反复改的参数参数作用推荐值踩坑点execution.checkpointing.intervalcheckpoint 间隔10s太短会增加达梦日志读取压力太长恢复时重复数据多scan.incremental.snapshot.chunk.size快照分片行数8096大表调大小表调小影响快照阶段并行度scan.snapshot.fetch.size单次拉取行数1024达梦 JDBC 对超大 fetch 支持不好别超过 4096dameng.log.mining.interval日志挖掘间隔500ms太短空转耗 CPU太长延迟高dameng.log.lsn.startup.mode启动位点模式latest可选 earliest/initial首次全量用 initialscan.incremental.snapshot.chunk.size这个参数特别关键。达梦在快照阶段如果单次查询返回行数太多容易触发游标超时或者内存溢出。我一般按表大小算小于 100 万行的表用 8096大于 1000 万行的表调到 32768同时把scan.snapshot.fetch.size控制在 2048 以内。dameng.log.lsn.startup.mode设成initial时FlinkCDC 会先做全量快照再从快照开始时的 LSN 消费增量。设成latest则跳过全量只从当前最新 LSN 开始适合已经做过全量初始化、只想接增量的场景。设错了会导致要么重复全量、要么丢历史变更。3.3 提交作业与验证数据一致性配置写完后用 Flink CLI 提交作业# 提交达梦 CDC 作业到 Flink 集群 ./bin/flink run \ -c com.example.DamengCdcJob \ -C file:///opt/flink/lib/flink-connector-dameng-cdc-1.0.jar \ -C file:///opt/flink/lib/DmJdbcDriver18.jar \ /opt/jobs/dameng-cdc-job.jar \ --dameng.host 192.168.1.100 \ --dameng.port 5236 \ --dameng.username SYSDBA \ --dameng.password SYSDBA123 \ --dameng.database DAMENG \ --table-list SYSDBA.ORDERS,SYSDBA.USERS \ --sink.kafka.bootstrap kafka:9092-C参数用来把 Connector 和驱动加到作业的 classpath避免和 Flink 自带 jar 冲突。--table-list支持逗号分隔的多表达梦的表名默认大写写小写会找不到表。提交后去 Flink Web UI 看numRecordsIn指标如果一直是 0先检查达梦的逻辑日志有没有新记录生成。验证一致性时我习惯在源库插一条数据然后去下游 Kafka 消费看能不能在秒级内拿到对应的 JSON。再更新、删除各做一次确认 op 字段分别是I、U、-D。如果更新操作只拿到后镜像没有前镜像检查达梦逻辑日志的LOGICAL_LOG_IMAGE参数是不是设成了AFTER改成BOTH才能拿到完整变更。4. 避坑与排查达梦 CDC 最常见的五类翻车现场4.1 归档日志被覆盖导致增量断层现象任务跑了一周某天重启后报LSN not found下游数据出现几小时的空洞。原因达梦的归档日志默认保留策略是按空间或时间如果日志生成速度快、保留窗口短FlinkCDC 还没消费到的日志就被删了。尤其是夜间批量作业集中写入时日志量暴涨很容易触发清理。解决把ARCH_SPACE_LIMIT调大或者设置ARCH_KEEP_TIME至少覆盖两个 checkpoint 周期。更稳妥的做法是单独挂一块盘放归档日志别和业务数据盘抢空间。重启任务前先确认起始 LSN 还在归档里不在的话只能重新全量。4.2 逻辑日志未对表启用导致变更丢失现象全量数据同步正常但后续的 insert/update 完全采集不到下游数据一直不变。原因只开了库级别的ENABLE_LOGIC_LOG忘了对具体表执行ALTER TABLE ... ENABLE LOGICAL LOG。达梦的逻辑日志是表级开关库级参数只是总闸。解决写个脚本批量对需要同步的表启用逻辑日志并且把这条检查加进上线前的 checklist。可以用SELECT TABLE_NAME FROM DBA_TABLES WHERE LOGICAL_LOG N找出漏配的表。4.3 快照阶段达梦连接数被打满现象任务启动后快照跑了一半卡住达梦监控显示连接数飙升到上限业务查询开始超时。原因scan.incremental.snapshot.chunk.size设得太小导致分片数过多每个分片开一个 JDBC 连接并行拉取。Flink 的并行度乘以分片数连接数直接爆炸。解决把 chunk size 调大减少分片数同时限制 Flink 作业的并行度别超过达梦MAX_SESSIONS的 30%。我一般会在达梦侧单独建一个 CDC 专用用户给它设连接数上限避免影响业务。4.4 时间戳类型字段解析成乱码现象下游 Kafka 里的TIMESTAMP字段变成一串数字或者null。原因达梦的TIMESTAMP类型在逻辑日志里是二进制格式自定义 Connector 如果没有正确实现DebeziumDeserializationSchema就会解析失败。达梦的 timestamp 精度到微秒和 Flink 的TimestampType默认毫秒精度不匹配。解决在反序列化时把达梦的微秒值除以 1000 转成毫秒或者用LocalDateTime直接接。别用java.sql.Timestamp的getTime()那个方法在达梦驱动里有精度丢失的坑。4.5 checkpoint 超时导致任务反复重启现象Flink UI 上 checkpoint 一直失败报Checkpoint expired before completing任务进入重启循环。原因达梦日志挖掘线程和 checkpoint 线程抢锁或者状态后端写入太慢。达梦的逻辑日志读取是单线程的如果日志量大checkpoint 时等待状态对齐的时间会超过超时阈值。解决把execution.checkpointing.timeout从默认 10 分钟调到 15 分钟同时把状态后端从MemoryStateBackend换成RocksDBStateBackend减少大状态下的 GC 压力。如果还不行检查达梦的LOG_BUFFER_SIZE是不是太小适当调大能加快日志读取。5. 进阶技巧用 FlinkCDC 做达梦到 Doris 的整库同步5.1 整库同步的表结构映射与自动建表单表同步跑通后下一步通常是整库同步。达梦到 Doris 的链路里最烦的是表结构映射达梦的VARCHAR2对应 Doris 的VARCHARNUMBER对应DECIMALCLOB对应STRING。如果手动建表几十张表能建到怀疑人生。我一般用 FlinkCDC 的SchemaEvolution能力配合 Doris 的auto-create-table参数让下游自动建表。达梦这边需要在 Connector 里实现SupportsSchemaEvolution接口把达梦的 schema 变更转成 Flink 的SchemaChangeEvent。Doris 的 Flink Connector 收到后会自动执行CREATE TABLE或ALTER TABLE ADD COLUMN。// 达梦 Schema 变更捕获示例 public class DamengSchemaChangeEvent implements SchemaChangeEvent { private final String tableName; private final String ddl; Override public String getDdl() { // 把达梦的 DDL 转成 Doris 兼容的语法 // 例如VARCHAR2(255) - VARCHAR(255) return ddl.replace(VARCHAR2, VARCHAR) .replace(NUMBER, DECIMAL); } }这段代码的关键在 DDL 转换。达梦的VARCHAR2在 Doris 里没有对应类型必须转成VARCHARNUMBER不带精度时默认转DECIMAL(38, 10)但 Doris 的 DECIMAL 最大精度是 38超过会报错。所以转换时要加一层校验精度超限的字段降级成STRING。5.2 分库分表场景下的达梦多源合并有些老系统按业务线拆了多个达梦实例每个实例里又有结构相同的分表。这种场景下FlinkCDC 可以用MySqlSource的多表合并思路把多个达梦 Source 读到的流union起来再统一写入 Doris。// 多达梦实例合并同步 DataStreamSourceRecord stream1 env.fromSource( DamengSource.builder() .host(192.168.1.101) .tableList(SYSDBA.ORDERS_01) .build(), WatermarkStrategy.noWatermarks(), dameng-source-1 ); DataStreamSourceRecord stream2 env.fromSource( DamengSource.builder() .host(192.168.1.102) .tableList(SYSDBA.ORDERS_02) .build(), WatermarkStrategy.noWatermarks(), dameng-source-2 ); stream1.union(stream2) .keyBy(record - record.getTableName()) .sinkTo(dorisSink);union之后必须keyBy表名否则同一张逻辑表的数据会被分散到不同 subtaskDoris 的批量写入会乱序。另外多个 Source 的 checkpoint 是独立对齐的如果某个实例挂了整个作业会重启所有 Source 一起回滚。对可用性要求高的场景建议每个实例单独跑一个作业下游用 Doris 的UNIQUE KEY模型做去重。5.3 监控与延迟度量别等业务投诉才发现同步断了同步链路最怕的不是报错是悄无声息地断了。我习惯在作业里加三个监控指标dameng.log.lag当前 LSN 和最新 LSN 的差值、sink.batch.interval下游写入间隔、checkpoint.durationcheckpoint 耗时。前两个用 Flink 的Gauge暴露第三个用Histogram。达梦这边可以定期查V$LOG_HISTORY看归档日志的生成速度如果生成速度突然下降可能是业务停了也可能是日志写入出了问题。下游 Doris 那边看SHOW LOAD的ETL状态如果长时间处于LOADING说明 Flink 写入批次太大或者 Doris 集群压力高。从那以后我每次上线达梦 CDC 任务都强制走一遍“插-更-删”三连验证再挂上延迟监控跑满 24 小时才敢交给业务。这套流程帮我拦住了至少三次潜在的增量断层。希望帮到你。本文还有配套的精品资源点击获取