Oracle 11g 通过 OGG 实时同步增量数据到 Kafka 实战指南

1. 为什么要在 11g 上折腾 OGG 到 Kafka 的实时同步

先说说这个项目到底在解决什么问题。很多公司的核心业务库还在 Oracle 11g 上跑着，短期内不可能整体迁移到新版本，但下游的数据团队、风控团队、实时大屏团队又急着要数据，而且要求是秒级延迟，不是那种 T+1 跑批能糊弄过去的。这时候摆在面前的路其实就那么几条：要么写触发器加中间表，要么用 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 只管写本地 trail，Pump 可以慢慢传；目标端 Kafka 挂了，Replicat 停掉就行，不会影响源库的抽取。这种设计在跨网络、跨机房的场景下特别有用，你可以在源端和目标端分别部署，中间走专线，互不干扰。

2.2 为什么选 OGG 而不是其他方案

有人会问，现在 Debezium 这么火，为什么不用它？Debezium 确实好用，但它对 Oracle 11g 的支持是通过 XStream 或者 LogMiner 适配器实现的，XStream 需要额外的 license，LogMiner 适配器在 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，虽然大概率能跑，但遇到问题官方不背锅。

注意：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 ALL

5. 目标端 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.handlerlist=kafkahandler gg.handler.kafkahandler.type=kafka gg.handler.kafkahandler.kafkaProducerConfigFile=./dirprm/producer.properties gg.handler.kafkahandler.topicMappingTemplate=ogg_${tableName} gg.handler.kafkahandler.keyMappingTemplate=${primaryKeys} gg.handler.kafkahandler.format=json gg.handler.kafkahandler.mode=op gg.handler.kafkahandler.SchemaTopicName=ogg_schema

topicMappingTemplate定义了 Topic 的命名规则，${tableName}会被替换成实际的表名，比如scott.emp会变成ogg_emp。format=json表示消息体是 JSON 格式，这是最常用的，因为下游消费起来方便。mode=op表示只投递变更操作，不投递完整行，这样消息体更小，但下游需要自己根据主键去关联历史数据。

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 就做 upsert，delete 就删掉对应记录。

这里有个坑：如果源表没有主键，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 的acks=all只能保证消息不丢，不能保证不重复。要解决重复问题，下游消费的时候需要做幂等处理，比如根据主键做 upsert，或者用消息里的pos字段去重。

数据丢失的原因比较多，可能是 Extract 的TABLE参数漏了某张表，也可能是 Pump 的TABLE参数不一致，还可能是 Replicat 的MAP参数写错了。排查的时候，先在源库做一条测试数据的变更，然后看 Extract 的 trail 文件里有没有这条记录，再看目标端的 trail 文件，最后看 Kafka 的 Topic 里有没有。一步一步缩小范围，总能找到问题。

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 或者 RocketMQ，OGG for Big Data 也支持这些消息队列，配置思路类似，只是 Handler 换一下。二是增加数据脱敏的环节，在 Replicat 投递之前，用 OGG 的COLMAP或者自定义的 Java 类对敏感字段做加密或掩码。三是做双向同步，也就是 Oracle 到 Kafka 的同时，也把 Kafka 的变更写回 Oracle，这个复杂度会高很多，需要处理冲突检测和循环捕获，没有强需求不建议碰。

最后再分享一个小技巧：OGG 的ggserr.log会越来越大，如果不清理，几个月就能撑满磁盘。可以配一个定时任务，每天把超过 7 天的日志归档或者删掉。还有 trail 文件，也要定期清理，尤其是测试环境，经常是跑几天就几个 G 的 trail 文件堆在那里。生产环境的话，建议用 OGG 自带的PURGEOLDEXTRACTS参数，让 Manager 自动清理过期的 trail 文件，省心很多。