四种数据库一套实时同步MySQL、PostgreSQL、Oracle、MongoDB CDC 实战实时同步真正难的不是“连上数据库”而是“持续拿到变更”。MySQL 看 BinlogPostgreSQL 读 WALOracle 挖日志MongoDB 订阅 Change Streams。四种数据库四套机制、四种位点、四组前置条件。如果每接一个源都要再引入一套 Debezium Kafka Flink同步这件事的成本可能比数据本身还高。01 批量同步解决不了的问题很多团队的同步任务一开始都是这么做的每天凌晨 2 点 → 全量查一遍源库 → 覆盖写入目标库数据量小的时候没问题。但只要业务开始实时化这套方案很快会露出三个短板延迟高早上 9 点看到的报表数据截止在凌晨 2 点压力大每次全量扫描都会给源库带来明显负载表越大越明显抓不到中间态一条订单从创建、支付到取消批量同步只能看到最终结果过程全部丢失。CDCChange Data Capture变更数据捕获解决的就是这个问题不反复查全表而是直接读取数据库的变更日志把每一次 INSERT、UPDATE、DELETE 都变成一条数据流。这件事在概念上很简单落到工程上却不简单因为不同数据库的“变更日志”完全不是一回事数据库变更来源采集方式位点MySQLBinlog伪装从库订阅binlog 文件 positionPostgreSQLWAL逻辑复制 pgoutputLSNOracleRedo LogLogMiner 日志挖掘SCNMongoDBOplogChange StreamsresumeToken这篇文章讲清楚一件事把这四种数据库的实时变更收敛成同一种数据流、同一套下游写法。02 四种数据库四种 CDCMySQL订阅 BinlogMySQL 的 Binlog 本来就是为复制设计的CDC 只是把“从库”换成了一个数据同步任务。前提条件很明确开启 Binlogbinlog_formatROW采集账号具备REPLICATION SLAVE、REPLICATION CLIENT权限。-- 检查SHOWVARIABLESLIKElog_bin;SHOWVARIABLESLIKEbinlog_format;-- 授权GRANTREPLICATIONSLAVE,REPLICATIONCLIENTON*.*TOdatay%;DataY 的MySQLBinlogInput会把自己伪装成一个从库实时解析WRITE_ROWS、UPDATE_ROWS、DELETE_ROWS等事件。几个容易被忽略的点serverId必须唯一。多个任务连同一个 MySQL 实例时冲突会导致互相踢下线相邻的同库同表同类型事件会自动合并最多 10,000 条减少下游写入次数MySQL 是四种源里唯一会把 DDL 一起传下去的ALTER TABLE可以自动同步到目标端Binlog 有保留时间任务停太久再启动可能已经追不回来了。PostgreSQL读取 WALPostgreSQL 没有 Binlog对应的是 WALWrite-Ahead Log。CDC 走的是逻辑复制-- 必须重启生效ALTERSYSTEMSETwal_levellogical;-- 采集账号需要复制权限ALTERROLE datayWITHREPLICATION;PostgresCDCInput底层用 pgjdbc 的复制 API 读取 WAL内置pgoutput协议解析器不需要额外部署 Debezium 或 Kafka。组件会自动维护两样东西publication声明哪些表需要发布replication slot记录消费到了哪个 LSN。这两者既是便利也是运维点。尤其是 slot任务长期停摆而 slot 一直存在时PostgreSQL 会保留对应的 WAL磁盘可能被慢慢吃满。下线任务时记得清理 slot。和 MySQL 相比PostgreSQL CDC 有几个明确差异不传输 DDL逻辑复制只发数据变更源表加字段不会自动同步TOAST 列UPDATE 时未变更的大字段不会携带值TRUNCATE当前版本只告警不做同步初始数据默认只捕获启动后的增量存量需要开启snapshot或单独做一次全量。OracleLogMiner 日志挖掘Oracle 的变更在 Redo Log 里DataY 用的是数据库自带的DBMS_LOGMNR不依赖 GoldenGate、XStream 这类额外组件。前置条件比前两种更重-- 1. 归档模式需要 SYSDBA重启到 MOUNTSHUTDOWNIMMEDIATE;STARTUP MOUNT;ALTERDATABASEARCHIVELOG;ALTERDATABASEOPEN;-- 2. 补充日志ALTERDATABASEADDSUPPLEMENTAL LOGDATA;ALTERDATABASEADDSUPPLEMENTAL LOGDATA(ALL)COLUMNS;-- 3. 授权GRANTLOGMININGTOdatay;GRANTSELECTANYTRANSACTIONTOdatay;GRANTEXECUTEONSYS.DBMS_LOGMNRTOdatay;GRANTEXECUTEONSYS.DBMS_LOGMNR_DTOdatay;OracleCDCInput会按 SCN 区间挖掘在线与归档日志把已提交事务转换成 INSERT / UPDATE / DELETE 事件。这里的工程约束要提前知道必须开归档模式否则没有日志可挖建议开启全列补充日志否则 UPDATE 事件里未变更的列可能拿不到LOB 列暂不支持CLOB/BLOB/LONG/ROWID会被忽略并告警不传输 DDL源表结构变更需要单独处理长事务有极小概率丢失不要让事务长时间不提交归档日志会持续增长需要配合备份与清理策略。MongoDB订阅 Change StreamsMongoDB 的 CDC 最“原生”直接用 Change Streams 监听集合变更官方驱动就支持。但它有一个硬前提必须运行在副本集或分片集群上单机 mongod 不可用。本地验证可以起一个单节点副本集dockerrun-d--namedatay-mongo-cdc-p27017:27017 mongo:7\--replSetrs0--bind_ip_alldockerexecdatay-mongo-cdc mongosh--eval\rs.initiate({_id:rs0,members:[{_id:0,host:localhost:27017}]})dockerexecdatay-mongo-cdc mongosh--evalrs.status().okMongoCDCInput把变更流事件映射成统一事件Change Streams 事件映射为下发数据insertINSERTfullDocumentupdate/replaceUPDATEfullDocument默认updateLookupdeleteDELETEdocumentKey通常只有_iddrop/rename/dropDatabase忽略仅告警同样有边界DELETE 前镜像有限默认只能拿到_id要完整前镜像需在集合上启用 pre-image集合无固定 schema字段由采样推断类型冲突时按 JSON 处理不传输结构变更删集合、改集合名不会自动同步。03 统一事件模型四种源一种流四种采集组件看起来完全不同但输出被统一成了同一个模型BinlogEvent。一条变更进入数据流后长这样{eventType:UPDATE,database:source,table:orders,timestamp:1792396517078,beforeData:{id:1001,status:PENDING},afterData:{id:1001,status:PAID}}FlowFile 上同时带上下游需要的元数据属性含义_eventTypeINSERT/UPDATE/DELETE/DDL_table表名MongoDB 为集合名_database库名 / schema_tableMetadata表结构用于自动建表与类型转换_timestamp事件时间这一步是整个方案的关键。从这一行开始MySQL 的 Binlog、PostgreSQL 的 WAL、Oracle 的 SCN、MongoDB 的 resumeToken对下游都不再有区别。下游只需要认识_eventType就能决定怎么写INSERT → 插入UPDATE → 按主键更新DELETE → 按主键删除DDL → 执行结构变更语句。StreamJdbcOutput的modelauto做的就是这件事。目标表不存在时它还会根据_tableMetadata自动建表并做跨库类型映射。也就是说同一条链路可以这样复用MySQL Binlog ─┐ PostgreSQL WAL ─┼─▶ 统一 CDC 事件 ─▶ StreamJdbcOutput(modelauto) ─▶ MySQL / PG / Oracle / DuckDB / Doris Oracle LogMiner ─┤ MongoDB Change ─┘04 四种源怎么配四个输入组件的配置结构高度一致差异主要在“采集范围”和“前置条件”。MySQL{.id:mysql_cdc,.name:MySQLBinlogInput,sourceId:{type:mysql,url:jdbc:mysql://127.0.0.1:3306/source,username:datay,password:******,dbschema:source},databaseNamePattern:source,tableNamePattern:orders|customers,serverId:1001,snapshot:true}serverId多任务部署时务必错开。PostgreSQL{.id:pg_cdc,.name:PostgresCDCInput,sourceId:{type:postgresql,url:jdbc:postgresql://127.0.0.1:5432/source,username:datay,password:******,dbschema:public},schema:public,tableNamePattern:orders|customers,publicationName:datay_pub_orders,slotName:datay_slot_orders,snapshot:true}固定publicationName和slotName任务重启后才知道从哪里续。Oracle{.id:oracle_cdc,.name:OracleCDCInput,sourceId:{type:oracle,url:jdbc:oracle:thin://127.0.0.1:1521/FREEPDB1,driver:oracle.jdbc.OracleDriver,username:datay,password:******,dbschema:DATAY},schema:DATAY,tableNamePattern:ORDERS|CUSTOMERS,pollIntervalMs:1000,snapshot:true}表名过滤不区分大小写采集范围始终限定在指定 schema 内。MongoDB{.id:mongo_cdc,.name:MongoCDCInput,sourceId:{type:mongodb,url:mongodb://127.0.0.1:27017,dbschema:test,extraParams:{replicaSet:rs0,directConnection:true,authSource:admin}},database:test,collectionNamePattern:orders|customers,snapshot:true}MongoDB 的连接参数放在extraParams里副本集名称和directConnection是最常出问题的两项。05 全量 增量快照不是可选项CDC 默认只捕获“任务启动之后”的变更。这意味着如果源库里已经有 500 万条历史数据直接开 CDC目标端只会拿到新数据。四种输入组件都提供了snapshot{snapshot:true,snapshotFetchSize:5000}首次运行没有断点、没有snapshotDone标记时顺序是先确定增量起点MySQL Binlog 位点 / PG LSN / Oracle SCN / Mongo operationTime对采集范围内的表做一次性全量读取按 INSERT 事件分批下发再从上述位点启动增量同步全量完成后写入snapshotDone重启不会重复全量。这里采用的是一致性策略叫“简单一致性”增量位点取在快照之前快照期间发生的变更会由 CDC 再补一遍。好处是实现轻、切换快代价是可能出现少量重复。对于按主键幂等写入或者存量基本稳定的场景完全够用。如果业务要求严格不重不漏就需要更重的导出快照方案。06 断点续传四种位点一个原则实时同步任务一定会重启发布、扩容、机器故障、数据库维护。所以每个 CDC 组件都必须回答一个问题重启之后从哪里接着读源位点保存方式MySQLbinlogFilebinlogPosition随 FlowFile 持久化PostgreSQLLSN复制槽确认位点 任务状态Oracle已提交事务的最大 SCN随 FlowFile 持久化MongoDBresumeToken随 FlowFile 持久化四种实现原则只有一个位点跟着数据走数据写成功之后再确认位点。这也是为什么 CDC 场景要强调幂等。至少一次投递、故障恢复重放、快照与增量重叠都会带来重复事件。最稳妥的做法是目标表必须有主键写入使用modelauto或按主键 upsert不要用“自增 ID append”去接 CDC 流。07 目标端auto 模式与自动建表源端统一之后目标端就简单了。StreamJdbcOutput的modelauto会根据_eventType自动选择写入策略事件写入策略INSERTINSERT / 批量追加UPDATE按主键 UPDATEDuckDB 为 INSERT OR REPLACEDELETE按主键 DELETEDDL执行 DDL 语句仅 MySQL 源会产生目标表不存在时组件会读取_tableMetadata经过DatabaseConverter做类型映射后自动建表。常见映射包括MySQLDATETIME→ PostgreSQLTIMESTAMP/ OracleTIMESTAMPMySQLTINYINT→ 目标端小整数类型OracleNUMBER(p,s)→ 目标端DECIMAL(p,s)MongoDB 动态字段 → JSON / 文本列。一个完整的跨库链路任务定义只有两个节点{units:[{.id:mysql_cdc,.name:MySQLBinlogInput,sourceId:{type:mysql,url:jdbc:mysql://127.0.0.1:3306/source,username:datay,password:******,dbschema:source},tableNamePattern:orders,snapshot:true},{.id:pg_sink,.name:StreamJdbcOutput,sourceId:{type:postgresql,url:jdbc:postgresql://127.0.0.1:5432/target,username:datay,password:******,dbschema:public},schema:public,table:,model:auto}],connections:[{sourceId:mysql_cdc,targetId:pg_sink,sourcePort:0}],version:1.0.0}把MySQLBinlogInput换成另外三个组件链路其余部分不用动。08 选型与限制对照维度MySQLPostgreSQLOracleMongoDB采集机制Binlog逻辑复制 / WALLogMinerChange Streams延迟毫秒级毫秒级秒级轮询毫秒级位点file positionLSNSCNresumeTokenDDL 同步支持不支持不支持不支持全量快照支持支持支持支持额外部署无无无无主要前提ROW 模式 复制权限wal_levellogical 复制权限归档 补充日志 LOGMINING副本集主要限制Binlog 过期TOAST / TRUNCATE / 槽位膨胀LOB 列 / 长事务 / 归档空间DELETE 前镜像有限 / 无固定 schema选型时可以记住三句话关系型库之间做实时同步优先 CDC不要轮询。延迟和源库压力都不在一个量级。只有 MySQL 能自动同步 DDL。其余三种源的表结构变更要有单独的上线流程。MongoDB 的难点不在接入在数据建模。无固定 schema 意味着目标端要么用 JSON 列要么先做字段收敛。09 上线前必须检查的几件事① 权限最小化采集账号只给复制/挖掘所需权限不要直接用业务账号或 DBA 账号跑同步。② 位点与目标表主键没有主键的表UPDATE / DELETE 无法可靠落地。CDC 上线前先检查采集范围内所有表的主键。③ 日志与磁盘MySQLBinlog 保留时间要覆盖最长停机时间PostgreSQL关注 replication slot 造成的 WAL 堆积Oracle归档日志增长与清理策略MongoDBoplog 窗口要足够长否则 resumeToken 会失效。④ 重复与幂等默认语义是至少一次。目标端按主键写入重复事件最多覆盖一次不会产生脏数据。⑤ 观测至少要看三个指标采集位点、同步延迟、目标端写入失败数。没有这三个指标实时同步就是黑盒。10 DataY 是什么上面这套能力跑在DataY上。它是一个开源、轻量的一体化数据平台一个 JAR、内嵌 DuckDB2 核 4G 就能跑通「数据集成 → 数据建模 → 数据开发 → 任务调度 → 数据应用」的完整链路。模块定位说明DataY Core数据集成引擎轻量、可嵌入、可扩展、流批一体JSON 定义任务、自动建表、CDC 实时同步DataY Web可视化数据平台拖拽式任务设计、任务调度、维度建模、指标管理、智能问数和实时同步相关的能力四种 CDC 输入MySQL Binlog、PostgreSQL WAL、Oracle LogMiner、MongoDB Change Streams统一的变更事件模型INSERT / UPDATE / DELETE 前后镜像 表结构元数据全量与增量一体snapshot首次先全量后增量完成后自动标记断点续传Binlog 位点、LSN、SCN、resumeToken 全部随任务状态持久化自动建表与类型转换跨 MySQL / PostgreSQL / Oracle / DuckDB / Doris 写入可视化编排CDC 之后可以接实时 SQL、属性路由、字段清洗、大模型判断再落到目标端快速开始gitclone https://gitee.com/yuyuejia/datay.gitcddatay mvn clean package-DskipTestsjava-jardatay-web/target/*.jar--DATAY_AI_API_KEYsk-xxxxxxxxx# 浏览器打开 http://localhost:8080默认账号 admin/admin产品首页http://datay.yuyuejia.com.cn/项目地址https://github.com/yuyuejia/datay11 写在最后实时同步的复杂度从来不在“写一个循环不断查表”。真正的复杂度藏在四种数据库各自的变更日志里怎么开、怎么读、从哪续、哪些变更拿不到、失败了怎么办。把这些差异收敛掉之后剩下的就是一条普通的数据流捕获变更 → 统一事件 → 清洗路由 → 幂等写入 → 可观测。当 MySQL、PostgreSQL、Oracle、MongoDB 都只是“一个输入组件”的差别实时同步才从一次性项目变成可以长期维护的基础设施。