1. 为什么要在 11g 上折腾 OGG 到 Kafka 的实时同步先说说这个项目到底在解决什么问题。很多公司的核心业务库还在 Oracle 11g 上跑着短期内不可能整体迁移到新版本但下游的数据团队、风控团队、实时大屏团队又急着要数据而且要求是秒级延迟不是那种 T1 跑批能糊弄过去的。这时候摆在面前的路其实就那么几条要么写触发器加中间表要么用 LogMiner 自己解析归档日志要么上商业化的 CDC 工具。触发器方案对源库性能影响大而且维护起来是噩梦自己解析归档日志光是处理各种数据类型和事务边界就能把人逼疯。所以 GoldenGate 这套东西虽然老但在 Oracle 生态里做增量捕获确实是稳的尤其是 11g 这个版本很多企业环境里跑得最久、踩坑记录最全。这个项目的核心目标很明确把 Oracle 11g 里的增量变更通过 OGG 的抽取和投递能力实时推到 Kafka 里让下游消费者能拿到近乎实时的数据流。适合谁来参考我觉得是那些手里有 11g 环境、又不想大动干戈改架构的 DBA 和数据工程师。你不需要把整个库都同步过去只需要挑几张核心业务表把 insert、update、delete 这些操作捕获出来序列化成 JSON 或者 Avro 丢进 Kafka下游爱怎么消费怎么消费。听起来简单但真做起来从环境准备到参数调优再到故障排查每一步都有坑。我见过太多人一上来就问“OGG 到 Kafka 怎么配”结果连源库的归档模式都没开或者补全日志没加抽取进程一启动就报错。所以这篇东西我会从最底层的原理讲起把每一步为什么这么做说清楚再把我自己踩过的坑和盘托出。你照着做不一定能一次成功但至少知道出问题该往哪个方向查。2. 整体架构设计与核心组件拆解2.1 数据流向的完整链路先把这个链路捋清楚不然后面配参数的时候容易晕。整个数据流向是这样的Oracle 11g 源库产生 redo 日志Extract 进程从 redo 或者归档日志里把变更抓出来写成 trail 文件然后 Pump 进程把本地 trail 文件通过网络传到目标端的 trail 文件最后 Replicat 进程读取目标端 trail 文件通过 Kafka Handler 把数据投递到 Kafka 集群的指定 Topic 里。这里有个关键点OGG for Big Data 的版本和传统 OGG 不太一样它自带了一套 Kafka Handler不需要你额外写 Java 程序去消费 trail 文件配置好之后 Replicat 直接就能往 Kafka 写。为什么用这种两段式 trail 文件的设计因为抽取和投递解耦了。源库压力大的时候Extract 只管写本地 trailPump 可以慢慢传目标端 Kafka 挂了Replicat 停掉就行不会影响源库的抽取。这种设计在跨网络、跨机房的场景下特别有用你可以在源端和目标端分别部署中间走专线互不干扰。2.2 为什么选 OGG 而不是其他方案有人会问现在 Debezium 这么火为什么不用它Debezium 确实好用但它对 Oracle 11g 的支持是通过 XStream 或者 LogMiner 适配器实现的XStream 需要额外的 licenseLogMiner 适配器在 11g 上对某些数据类型的处理不够稳定尤其是 LOB 字段和压缩表。OGG 在这方面积累了很多年对 11g 的各种奇葩场景兼容性更好。而且 OGG 的 Replicat 可以直接对接 Kafka省去了中间消息队列的转换层链路更短延迟更低。另一个考虑是运维成本。Debezium 跑在 Kafka Connect 里虽然部署简单但一旦出问题排查起来要同时看 Connect 的日志、Kafka 的日志、还有 Oracle 的日志链路太长。OGG 虽然配置繁琐但每个进程的日志很清晰ggserr.log 里能看到每一步在干什么对于习惯了 Oracle 生态的 DBA 来说上手更快。2.3 版本选型与兼容性矩阵这里要特别强调版本问题。标题里写的是“11g 最新版”指的是 Oracle 数据库 11.2.0.4 这个版本这是 11g 系列里最稳定、补丁最全的版本。OGG 的版本选择上建议用 12.3.0.1 或者 19.1 的 Microservices 版本但如果你习惯用经典架构12.2.0.1 也够用。关键是 OGG for Big Data 的版本要和 Kafka 的版本匹配比如 OGG 19.1 官方认证的 Kafka 版本是 2.x如果你用的是 Kafka 3.x虽然大概率能跑但遇到问题官方不背锅。组件推荐版本备注Oracle 数据库11.2.0.4必须开启归档和补全日志OGG 核心12.3.0.1 或 19.1经典架构或微服务架构均可OGG for Big Data与 OGG 核心版本一致包含 Kafka HandlerKafka2.8.x 或 3.x需确认 Handler 兼容性JDK1.8OGG 对 JDK 版本敏感注意OGG 19.1 的 Kafka Handler 在 JDK 11 上跑会有反射相关的警告虽然不影响功能但日志里一堆报错看着心烦建议还是用 JDK 8。3. 源端环境准备与关键配置3.1 数据库层面的必做功课在装 OGG 之前源库这边有几件事必须做少一件后面都会报错。第一开启归档模式这个不用多说没有归档日志 OGG 没法抽取。第二开启最小补全日志也就是ALTER DATABASE ADD SUPPLEMENTAL LOG DATA;这一步是为了让 redo 日志里包含足够的信息来唯一标识一行记录。第三对你需要同步的表开启表级补全日志ALTER TABLE schema.table ADD SUPPLEMENTAL LOG DATA (ALL) COLUMNS;注意这里用 ALL COLUMNS 而不是 PRIMARY KEY因为如果只记录主键update 的时候 OGG 拿不到变更前的完整行信息投递到 Kafka 的 JSON 里就只有主键和变更列下游做宽表关联的时候会缺字段。还有一个容易被忽略的点ENABLE_GOLDENGATE_REPLICATION参数。在 11.2.0.4 上这个参数默认是 false需要手动改成 true否则 OGG 的 Extract 进程连不上数据库。改完之后要重启数据库别想着动态生效这个参数是静态的。-- 检查归档模式 ARCHIVE LOG LIST; -- 开启最小补全日志 ALTER DATABASE ADD SUPPLEMENTAL LOG DATA; ALTER DATABASE ADD SUPPLEMENTAL LOG DATA (PRIMARY KEY) COLUMNS; -- 对具体表开启全列补全日志 ALTER TABLE scott.emp ADD SUPPLEMENTAL LOG DATA (ALL) COLUMNS; -- 检查补全日志状态 SELECT SUPPLEMENTAL_LOG_DATA_MIN, SUPPLEMENTAL_LOG_DATA_PK, SUPPLEMENTAL_LOG_DATA_UI FROM V$DATABASE;3.2 OGG 安装与目录规划OGG 的安装其实很简单解压就行但目录规划有讲究。我习惯把 OGG 装在独立的文件系统上不要和 Oracle 的 ORACLE_HOME 混在一起因为 trail 文件会不断增长万一把根目录撑爆了数据库也跟着挂。目录结构大概是这样/ogg是安装目录下面分dirprm放参数文件dirdat放 trail 文件dirrpt放报告dirlog放日志。每个进程的 trail 文件最好单独放一个目录比如dirdat/ex给 Extract 用dirdat/pu给 Pump 用这样清理旧文件的时候不容易误删。环境变量方面LD_LIBRARY_PATH要包含 OGG 的 lib 目录和 Oracle 的 lib 目录顺序不能反否则会加载到错误的库。ORACLE_HOME和ORACLE_SID也要设好因为 OGG 的 Extract 进程需要调用 Oracle 的客户端库来连接数据库。3.3 创建 OGG 专用数据库用户不要用 sys 或者 system 跑 OGG权限太大容易出事。建一个专用用户比如ogguser给它必要的权限。在 11g 上需要授予CREATE SESSION、ALTER SESSION、SELECT ANY DICTIONARY、SELECT ANY TABLE、FLASHBACK ANY TABLE、EXECUTE ON DBMS_FLASHBACK这些权限。如果要做 DDL 同步还需要CREATE TABLE、CREATE SEQUENCE之类的权限但一般增量数据同步不涉及 DDL所以可以省掉。CREATE USER ogguser IDENTIFIED BY ogguser DEFAULT TABLESPACE users; GRANT CREATE SESSION, ALTER SESSION TO ogguser; GRANT SELECT ANY DICTIONARY TO ogguser; GRANT SELECT ANY TABLE TO ogguser; GRANT FLASHBACK ANY TABLE TO ogguser; GRANT EXECUTE ON DBMS_FLASHBACK TO ogguser;提示如果源库有大量表SELECT ANY TABLE的权限检查会拖慢 Extract 的启动速度可以改成只授予具体表的 SELECT 权限但维护起来麻烦自己权衡。4. 配置 Extract 与 Pump 进程4.1 Extract 进程的参数文件详解Extract 进程是整个链路的源头它的参数文件决定了抓什么、怎么抓。我一般把参数文件命名为ext1.prm放在dirprm目录下。核心参数包括EXTRACT ext1 SETENV (ORACLE_HOME /u01/app/oracle/product/11.2.0/dbhome_1) SETENV (ORACLE_SID orcl) USERID ogguser, PASSWORD ogguser EXTTRAIL ./dirdat/ex TABLE scott.emp; TABLE scott.dept;这里有几个细节值得展开。SETENV是为了确保 Extract 进程能找到正确的 Oracle 库尤其是在同一台机器上有多个 Oracle Home 的时候。EXTTRAIL指定了本地 trail 文件的前缀OGG 会自动生成类似ex000000这样的文件。TABLE参数指定要同步的表可以用通配符比如TABLE scott.*;但生产环境不建议这么干万一同步了不该同步的表下游会收到一堆垃圾数据。还有一个隐藏参数TRANLOGOPTIONS用来控制 Extract 如何读取 redo 日志。在 11g 上建议加上TRANLOGOPTIONS EXCLUDEUSER ogguser把 OGG 自己的操作排除掉避免循环捕获。如果源库有 RAC 集群还需要加THREADS参数指定线程数。4.2 Pump 进程的配置要点Pump 进程负责把本地 trail 传到目标端参数文件pump1.prm大概长这样EXTRACT pump1 RMTHOST 192.168.1.100, MGRPORT 7809 RMTTRAIL ./dirdat/pu PASSTHRU TABLE scott.emp; TABLE scott.dept;RMTHOST指定目标端的 IP 和 Manager 端口RMTTRAIL指定目标端的 trail 文件前缀。PASSTHRU表示不做任何转换直接把数据传过去这样可以减少源端的 CPU 消耗。如果网络带宽有限可以加COMPRESS参数压缩传输但会增加 CPU 开销自己权衡。注意Pump 进程的TABLE参数必须和 Extract 进程的一致否则会出现数据丢失或者重复。我见过有人改了 Extract 的表列表忘了改 Pump结果新加的表一直没同步过去查了半天才发现。4.3 启动与状态检查配置好之后用ggsci进入命令行依次启动 Manager、Extract、Pump。启动顺序不能乱Manager 必须先起来否则其他进程连不上。启动命令是START MANAGER、START ext1、START pump1。启动之后用INFO ALL查看状态正常应该是RUNNING。如果看到ABENDED赶紧去看ggserr.log里面会有详细的报错信息。# 进入 ggsci ./ggsci # 启动 Manager START MANAGER # 启动 Extract START ext1 # 启动 Pump START pump1 # 查看状态 INFO ALL5. 目标端 Kafka 投递配置5.1 Replicat 进程与 Kafka Handler目标端的 Replicat 进程负责把 trail 文件里的数据写到 Kafka。这里要用到 OGG for Big Data 提供的 Kafka Handler它的本质是一个 Java 类Replicat 通过 JVM 调用它来发送消息。参数文件rep1.prm大概是这样REPLICAT rep1 TARGETDB LIBFILE libggjava.so SET property./dirprm/kafka.props GROUPTRANSOPS 1000 MAXTRANSOPS 10000 MAP scott.emp, TARGET scott.emp; MAP scott.dept, TARGET scott.dept;TARGETDB这一行是关键它加载了 Kafka Handler 的共享库和属性文件。GROUPTRANSOPS控制每多少条操作提交一次设太小会导致 Kafka 消息太碎设太大会增加延迟。MAXTRANSOPS是单个事务的最大操作数超过这个数会强制拆分防止大事务把内存撑爆。5.2 kafka.props 属性文件详解kafka.props是 Kafka Handler 的核心配置文件里面定义了 Kafka 集群地址、Topic 映射、序列化格式等。一个典型的配置如下gg.handlerlistkafkahandler gg.handler.kafkahandler.typekafka gg.handler.kafkahandler.kafkaProducerConfigFile./dirprm/producer.properties gg.handler.kafkahandler.topicMappingTemplateogg_${tableName} gg.handler.kafkahandler.keyMappingTemplate${primaryKeys} gg.handler.kafkahandler.formatjson gg.handler.kafkahandler.modeop gg.handler.kafkahandler.SchemaTopicNameogg_schematopicMappingTemplate定义了 Topic 的命名规则${tableName}会被替换成实际的表名比如scott.emp会变成ogg_emp。formatjson表示消息体是 JSON 格式这是最常用的因为下游消费起来方便。modeop表示只投递变更操作不投递完整行这样消息体更小但下游需要自己根据主键去关联历史数据。producer.properties是 Kafka 原生的生产者配置里面至少要配bootstrap.servers、acks、key.serializer、value.serializer。acks建议设成all虽然延迟会稍微高一点但能保证消息不丢。key.serializer和value.serializer用 Kafka 自带的StringSerializer就行因为 OGG 已经把数据序列化成 JSON 字符串了。5.3 消息格式与下游消费OGG 投递到 Kafka 的消息格式默认是 JSON包含table、op_type、op_ts、current_ts、pos、before、after这些字段。op_type有I、U、D三种分别代表 insert、update、delete。before和after分别表示变更前和变更后的行数据。下游消费的时候根据op_type决定怎么处理比如 insert 就直接写目标表update 就做 upsertdelete 就删掉对应记录。这里有个坑如果源表没有主键OGG 默认会用所有列作为 key但这样会导致消息的 key 特别长而且 update 的时候无法准确定位到具体行。所以强烈建议所有需要同步的表都有主键实在没有的话加一个唯一索引也行。6. 实操过程中的常见问题与排查技巧6.1 Extract 启动报错 OGG-00446这个错误太常见了基本上第一次配 OGG 的人都会遇到。报错信息大概是“Could not find archived log for sequence...”意思是 Extract 找不到需要的归档日志。原因通常是归档日志被删了或者 Extract 的检查点落后太多。解决办法是先确认归档日志的保留策略确保 Extract 需要的日志还在。如果确实被删了只能重新初始化 Extract也就是删掉检查点从当前时间点重新开始抓。# 在 ggsci 里删除 Extract 的检查点 DELETE EXTRACT ext1 # 重新添加 ADD EXTRACT ext1, TRANLOG, BEGIN NOW提示重新初始化之前一定要确认下游能不能接受数据丢失或者有没有全量同步的兜底方案。生产环境千万别随便删检查点。6.2 Replicat 投递到 Kafka 报错 Connection refused这个一般是 Kafka 集群地址配错了或者网络不通。先检查producer.properties里的bootstrap.servers是不是写对了注意端口号是 9092 还是别的。然后在 OGG 服务器上用telnet或者nc测一下能不能连上 Kafka 的端口。如果网络通但还是报错看看 Kafka 的advertised.listeners配置有时候 Kafka 返回给客户端的地址是内网地址OGG 服务器解析不了。6.3 数据重复或丢失的排查思路数据重复通常是因为 Replicat 在投递过程中失败了重启之后从上一个检查点重新投递导致部分消息重复。Kafka 的acksall只能保证消息不丢不能保证不重复。要解决重复问题下游消费的时候需要做幂等处理比如根据主键做 upsert或者用消息里的pos字段去重。数据丢失的原因比较多可能是 Extract 的TABLE参数漏了某张表也可能是 Pump 的TABLE参数不一致还可能是 Replicat 的MAP参数写错了。排查的时候先在源库做一条测试数据的变更然后看 Extract 的 trail 文件里有没有这条记录再看目标端的 trail 文件最后看 Kafka 的 Topic 里有没有。一步一步缩小范围总能找到问题。问题现象可能原因排查方法Extract 启动报 OGG-00446归档日志缺失检查归档保留策略必要时重新初始化Replicat 报 Connection refusedKafka 地址错误或网络不通检查 producer.properties测试端口连通性数据重复Replicat 重启导致重复投递下游做幂等处理用主键 upsert数据丢失表列表不一致或 MAP 写错逐段检查 Extract、Pump、Replicat 的参数延迟越来越高大事务或 Kafka 吞吐瓶颈调整 GROUPTRANSOPS检查 Kafka 分区数6.4 性能调优的几个关键参数如果发现同步延迟越来越大可以从这几个方面调优。第一增大 Extract 的TRANLOGOPTIONS里的READSIZE默认是 1048576可以调到 2097152加快 redo 日志的读取速度。第二调整 Replicat 的GROUPTRANSOPS如果 Kafka 吞吐够可以调到 5000 甚至 10000减少网络往返次数。第三增加 Kafka Topic 的分区数让 Replicat 可以并行投递但 OGG 的 Kafka Handler 默认是单线程的要并行的话需要配多个 Replicat 进程每个负责不同的表。还有一个容易被忽略的点Kafka 的linger.ms和batch.size。linger.ms设成 10 到 50 毫秒batch.size设成 16384 或者 32768可以让生产者攒一批消息再发提高吞吐量但会增加一点延迟。这个需要根据业务对延迟的容忍度来权衡。7. 一些个人经验与后续扩展思路这个方案我前后在几个环境里部署过最深的体会是OGG 的配置本身不难难的是对源库的理解和对异常情况的预判。比如有一次源库做了一次大批量的数据清理删了几百万行结果 Extract 直接卡死因为单个事务太大trail 文件瞬间暴涨。后来加了MAXTRANSOPS参数把大事务拆成小事务才解决了问题。所以你在配的时候一定要根据源库的业务特点去调参数不能照搬别人的配置。后续如果想扩展有几个方向可以考虑。一是把 Kafka 换成 Pulsar 或者 RocketMQOGG for Big Data 也支持这些消息队列配置思路类似只是 Handler 换一下。二是增加数据脱敏的环节在 Replicat 投递之前用 OGG 的COLMAP或者自定义的 Java 类对敏感字段做加密或掩码。三是做双向同步也就是 Oracle 到 Kafka 的同时也把 Kafka 的变更写回 Oracle这个复杂度会高很多需要处理冲突检测和循环捕获没有强需求不建议碰。最后再分享一个小技巧OGG 的ggserr.log会越来越大如果不清理几个月就能撑满磁盘。可以配一个定时任务每天把超过 7 天的日志归档或者删掉。还有 trail 文件也要定期清理尤其是测试环境经常是跑几天就几个 G 的 trail 文件堆在那里。生产环境的话建议用 OGG 自带的PURGEOLDEXTRACTS参数让 Manager 自动清理过期的 trail 文件省心很多。
