Oracle 11g OGG 到 Kafka 增量同步实战：配置、调优与避坑指南

1. 为什么增量同步这件事值得单独拿出来讲做数据平台的人都有一个共识全量同步是体力活增量同步才是技术活。全量同步无非是导一份快照、灌一次目标端数据量大就多等几个小时逻辑上没什么弯弯绕。但增量同步不一样它要求你在源库持续写入的前提下把每一条变更准确、有序、低延迟地搬到目标端中间任何一个环节出问题都可能造成数据丢失或者重复。Oracle 到 Kafka 的增量同步是这类需求里最典型的场景之一。源端是关系型数据库里最重的那一类事务机制复杂、日志体系庞大目标端是分布式消息队列讲究的是高吞吐和顺序性。把这两端接起来中间需要一个既懂 Oracle 事务、又能稳定输出的采集组件。Oracle GoldenGate下面统一简称 OGG就是干这个的它在 Oracle 生态里做日志级增量采集已经很多年了成熟度和稳定性都经过大量生产环境验证。这篇内容面向的是正在做或者准备做 Oracle 到 Kafka 实时同步的开发和运维同学尤其是还在用 11g 版本、又不想大动源库架构的团队。我会把整套方案的思路、配置细节、踩过的坑都摊开讲尽量做到你照着做就能跑起来。需要说明的是文中涉及的具体参数和步骤一部分来自官方文档的标准做法一部分是我在实际项目里根据常见实践补充的经验两者我会尽量区分清楚。2. 整体方案设计与技术选型思路2.1 为什么是 OGG 而不是其他采集方式在 Oracle 增量采集这个领域能选的方案其实不多大致有这么几类基于触发器的采集、基于时间戳轮询的采集、基于日志解析的采集。前两种在真实生产里基本可以排除原因很直接。触发器方案需要在源表上挂触发器每次 DML 都会额外触发一段 PL/SQL 逻辑对源库的性能影响是实打实的而且触发器维护起来极其麻烦表结构一变就得跟着改。时间戳轮询更不靠谱它依赖表里有可靠的更新时间字段物理删除根本捕获不到而且轮询间隔和延迟之间永远在打架。日志解析方案是唯一能做到对源库几乎零侵入、又能捕获全部变更类型包括删除的方式。Oracle 自己的日志体系里归档日志和在线重做日志记录了所有数据变更OGG 的 Extract 进程就是通过读取这些日志来还原变更的。它不需要在源表上加任何东西对业务代码完全透明这是它最大的价值。至于为什么不用 Oracle 自带的 LogMiner 直接对接是因为 LogMiner 本身更偏向诊断和审计用途做持续高吞吐的增量抽取时资源消耗和稳定性都不如 OGG 专门优化的抽取进程。OGG 在日志解析之上还做了事务重组、检查点管理、断点续传这些工程化的事情这些才是生产环境真正需要的。2.2 整体链路拆解整套链路可以拆成四段我用一个表格把每段的职责和关键组件列清楚链路阶段核心组件主要职责源端日志准备源库归档模式、补充日志保证变更被完整记录变更抽取Extract 进程从日志中解析出变更记录变更传输Data Pump、Server Collector把变更投递到目标端变更投递Replicat 进程、Kafka Handler写入 Kafka Topic这里有个容易混淆的点OGG 传统架构里Extract 抽出来的数据要先落到本地的 Trail 文件再由 Data Pump 进程通过网络传到目标端的 Trail 文件最后 Replicat 读取目标端 Trail 写入最终目标。这个两段式 Trail设计是为了解耦抽取和投递源端和目标端可以各自独立启停网络抖动也不会直接丢数据。但对接 Kafka 的时候情况有点特殊。OGG 从 12c 开始提供了 Kafka Handler可以让 Replicat 直接把变更写成 Kafka 消息。在 11g 上官方的 Kafka Handler 支持相对有限所以很多团队会选择用 Replicat 写到文件或者 JMS再由外部程序转发到 Kafka。不过如果版本和补丁到位11g 也能通过配置实现直接投递。下面我按能直接投递的思路来讲同时把降级方案也说清楚。2.3 关键设计决策事务一致性和顺序性Kafka 的一个 Topic 分区内是有序的但跨分区不保证顺序。而 Oracle 的事务可能涉及多张表、多行记录如果这些变更被分散到不同分区消费端就可能看到半个事务。这是设计阶段必须想清楚的问题。我的建议是按主键或者业务主键做分区键保证同一业务实体的变更落到同一分区。对于跨表事务如果业务上要求强一致那就得把所有相关表的变更路由到同一个分区代价是并行度下降。这个取舍没有标准答案取决于你的业务能不能接受最终一致。另一个决策点是提交时机。OGG 的 Replicat 默认是按事务提交的也就是说一个 Oracle 事务的所有变更会作为一个批次写入 Kafka。这个行为要确认清楚因为它直接决定了消费端看到的数据粒度。3. 源端准备把地基打牢3.1 归档模式和补充日志的开启这一步是所有日志解析方案的前提没做好后面全是坑。源库必须运行在归档模式ARCHIVELOG否则在线重做日志被覆盖后变更就永久丢失了。检查命令很简单-- 查看当前归档模式 ARCHIVE LOG LIST; -- 如果没开需要重启到 mount 状态开启 SHUTDOWN IMMEDIATE; STARTUP MOUNT; ALTER DATABASE ARCHIVELOG; ALTER DATABASE OPEN;补充日志Supplemental Logging是另一个必须项。默认情况下Oracle 的重做日志只记录改了哪些列不记录改之前是什么值而增量同步往往需要知道变更前后的完整行尤其是做更新和删除的时候。补充日志分几个层级-- 数据库级最小补充日志必须开 ALTER DATABASE ADD SUPPLEMENTAL LOG DATA; -- 数据库级全列补充日志会记录所有列的前像 ALTER DATABASE ADD SUPPLEMENTAL LOG DATA (ALL) COLUMNS; -- 表级补充日志只对关键表开减少日志量 ALTER TABLE schema_name.table_name ADD SUPPLEMENTAL LOG DATA (ALL) COLUMNS;提示数据库级全列补充日志会让日志量显著增长生产环境建议只对需要同步的表开表级补充日志不要图省事全库开。这里有个经验主键补充日志是必须的因为 OGG 要靠主键来定位要更新的行。如果表没有主键OGG 会退而求其次用所有列做定位效率低还容易出错。所以同步的表最好都有主键没有的话建议加上或者至少建一个唯一索引。3.2 OGG 源端安装与目录规划OGG 的安装本身不复杂解压即用但目录规划很讲究。我一般会这样分OGG 安装目录放软件本身升级时不动数据Trail 文件目录放抽取出来的变更文件要预留足够空间参数文件目录放各种配置文件检查点目录放断点信息这个绝对不能丢磁盘空间要重点算一下。Trail 文件的增长速度取决于源库的变更量高峰期可能每小时几个 G。如果目标端消费慢Trail 会堆积磁盘满了 Extract 就会挂。我的做法是给 Trail 目录单独挂一块盘并且配监控告警使用率超过 70% 就报警。3.3 创建 OGG 专用数据库用户不要用 sys 或者业务用户跑 OGG权限太大不安全权限不够又跑不起来。标准做法是建一个专用用户授予必要的权限CREATE USER ogg_user IDENTIFIED BY your_password; GRANT CREATE SESSION, ALTER SESSION TO ogg_user; GRANT RESOURCE TO ogg_user; GRANT SELECT ANY DICTIONARY TO ogg_user; GRANT SELECT ANY TABLE TO ogg_user; GRANT FLASHBACK ANY TABLE TO ogg_user; GRANT EXECUTE ON DBMS_FLASHBACK TO ogg_user;如果是 11g还需要额外授予EXECUTE ON DBMS_CAPTURE_ADM之类的权限具体取决于你要同步的对象范围。权限这块的原则是给够但不多给。给多了是安全隐患给少了 Extract 启动就报错。4. 核心配置Extract 与 Replicat 的实操细节4.1 Extract 进程配置Extract 是整条链路的源头它的配置决定了抽什么、怎么抽。一个典型的 Extract 参数文件长这样EXTRACT ext_kafka SETENV (ORACLE_SID ORCL) SETENV (NLS_LANG AMERICAN_AMERICA.AL32UTF8) USERID ogg_user, PASSWORD your_password EXTTRAIL ./dirdat/ea TRANLOGOPTIONS EXCLUDEUSER ogg_user TABLE schema_name.table_name;逐行解释一下关键点。SETENV里的字符集要和源库一致否则中文会乱码这个坑我踩过不止一次。EXTTRAIL指定 Trail 文件的前缀OGG 会自动在后面加序号。TRANLOGOPTIONS EXCLUDEUSER用来排除 OGG 自己的操作避免自己抽自己形成循环。TABLE那行是核心它决定了抽取范围。可以写具体的表也可以用通配符。如果只同步部分列可以写成TABLE schema.table (col1, col2, col3);。这里要注意如果表结构变了比如加了列Extract 参数不更新的话新列不会被同步。启动 Extract 之前一定要先注册ADD EXTRACT ext_kafka, TRANLOG, BEGIN NOW ADD EXTTRAIL ./dirdat/ea, EXTRACT ext_kafkaBEGIN NOW表示从当前时间点开始抽如果要抽历史数据可以指定 SCN 或者时间戳。注册完之后用START ext_kafka启动。4.2 Data Pump 与网络传输如果源端和目标端不在同一台机器中间需要 Data Pump 进程把本地 Trail 传到目标端。配置如下EXTRACT pump_kafka RMTHOST target_host, MGRPORT 7809 RMTTRAIL ./dirdat/ra PASSTHRU TABLE schema_name.table_name;PASSTHRU表示不做任何转换直接透传这样效率最高。RMTHOST指定目标端地址和 Manager 端口。Data Pump 的好处是它只负责搬运不做解析资源消耗小而且源端 Trail 可以配置成传完就删节省空间。4.3 Replicat 对接 Kafka 的配置这是整条链路最关键的一环。Replicat 要把变更写成 Kafka 消息需要配置 Kafka Handler。参数文件大致如下REPLICAT rep_kafka TARGETDB LIBFILE libggjava.so SET propertydirprm/kafka.props MAP schema_name.table_name, TARGET schema_name.table_name;真正的 Kafka 连接信息在kafka.props里gg.handler.kafkahandler.KafkaProducerConfigFilekafka-producer.properties gg.handler.kafkahandler.topicMappingTemplateogg_${schemaName}_${tableName} gg.handler.kafkahandler.keyMappingTemplate${primaryKeys} gg.handler.kafkahandler.formatjson gg.handler.kafkahandler.modeop这里几个参数值得展开说。topicMappingTemplate决定了消息往哪个 Topic 发我一般按库名_表名的规则生成这样消费端一眼就知道数据来源。keyMappingTemplate用主键做消息 key这样同一主键的变更会落到同一分区保证顺序。formatjson让消息体是 JSON消费端解析方便。modeop表示只输出变更操作不输出完整行。kafka-producer.properties里是标准的 Kafka 生产者配置重点调这几个bootstrap.serverskafka1:9092,kafka2:9092,kafka3:9092 acksall retries3 linger.ms50 batch.size16384 compression.typelz4acksall保证消息被所有副本确认才算成功牺牲一点延迟换可靠性增量同步场景我认为这个取舍是值得的。linger.ms和batch.size配合使用适当攒批能显著提升吞吐。4.4 参数选择背后的计算逻辑有人会问batch.size到底设多少合适。这个没有万能值但可以算。假设你的峰值变更量是每秒 5000 条每条消息平均 1KB那每秒就是 5MB。如果linger.ms50一个批次窗口内大约积累 250KB 数据。batch.size设成 256KB 左右就比较匹配设太小会频繁发送设太大则批次永远填不满反而增加延迟。retries和acks要一起看。acksall加上retries3意味着最多重试三次配合幂等生产者enable.idempotencetrue可以避免重复消息。但幂等生产者对 Kafka 版本有要求11g 环境对接的 Kafka 版本如果较老可能不支持那就得在消费端做去重。5. 实操全流程与关键环节记录5.1 从零到跑通的完整步骤我把整个流程按顺序列一遍你可以对照着做确认源库归档模式和补充日志已开启检查v$database的LOG_MODE字段创建 OGG 专用用户并授权测试用该用户能正常连接安装 OGG 源端和目标端软件规划好目录和磁盘在源端配置 Manager 进程并启动确认端口监听正常配置 Extract 参数文件注册并启动 Extract配置 Data Pump 参数文件注册并启动在目标端配置 Manager 进程配置 Replicat 和 Kafka Handler注册并启动在源表做几条增删改观察 Kafka Topic 是否收到消息验证消息内容、顺序、事务完整性每一步做完都要验证不要一口气全配完再测出了问题很难定位是哪一环。5.2 验证环节的具体做法验证不是随便插一条数据看看有没有到就完事。我一般会设计一组测试用例-- 单条插入 INSERT INTO test_table (id, name, amount) VALUES (1, test_a, 100); -- 批量插入 INSERT INTO test_table (id, name, amount) SELECT level, batch_ || level, level * 10 FROM dual CONNECT BY level 100; -- 更新 UPDATE test_table SET amount 999 WHERE id 1; -- 删除 DELETE FROM test_table WHERE id 1; -- 事务测试多条操作在一个事务里 BEGIN INSERT INTO test_table VALUES (200, tx_a, 1); INSERT INTO test_table VALUES (201, tx_b, 2); UPDATE test_table SET amount 3 WHERE id 200; COMMIT; END;然后在 Kafka 端用命令行消费kafka-console-consumer.sh --bootstrap-server kafka1:9092 \ --topic ogg_schema_test_table --from-beginning重点看三件事消息是不是都到了、顺序对不对、事务内的多条变更是不是连续出现。如果事务内的消息被其他消息插队了说明分区或者提交策略有问题。5.3 性能调优的实操记录默认配置跑起来之后如果吞吐上不去可以从这几个方向调。Extract 端可以开并行抽取把大表拆成多个 Extract 进程各自负责一部分表。Replicat 端可以开并行应用但要注意并行度不能超过 Topic 分区数否则多出来的并行度没意义。我实测过一个场景单 Extract 单 Replicat 的吞吐大概在每秒 3000 到 5000 条开了 4 个并行之后能到每秒 15000 条左右。但并行不是越多越好每个进程都有内存和 CPU 开销而且并行度高了之后事务顺序更难保证。我的经验是先从 2 到 4 个并行试起压测看瓶颈在哪再决定加不加。6. 常见问题与排查技巧实录6.1 典型问题速查表现象可能原因排查方向Extract 启动报错 OGG-00446补充日志没开或权限不足检查补充日志和用户权限消息中文乱码字符集不一致检查 SETENV 的 NLS_LANG部分变更丢失补充日志级别不够确认表级补充日志已开消息重复重试机制导致检查幂等配置或消费端去重延迟越来越大消费端慢或 Trail 堆积看 Trail 目录增长和消费 lagReplicat 报 Kafka 连接失败网络或认证问题检查 bootstrap.servers 和认证配置6.2 几个我踩过的坑第一个坑是字符集。源库是 ZHS16GBKOGG 参数里没设 NLS_LANG结果同步过去的中文全是问号。这个问题的隐蔽性在于英文和数字都正常只有中文出问题很容易被忽略。解决办法就是SETENV (NLS_LANG AMERICAN_AMERICA.AL32UTF8)注意这里要设成 AL32UTF8 而不是源库的字符集因为 OGG 内部处理用的是 UTF8。第二个坑是主键更新。有一张表的主键是业务字段业务上偶尔会更新主键值。OGG 默认情况下主键更新会被拆成一条删除加一条插入如果消费端按主键做幂等就会出问题。解决办法是在 Extract 参数里加GETUPDATEBEFORES让 OGG 记录更新前的值这样消费端能识别出这是主键变更而不是删除加插入。第三个坑是 Trail 文件堆积。有一次目标端 Kafka 集群做维护停了两个小时源端 Extract 还在跑Trail 文件迅速涨到几百 G把磁盘撑满了Extract 直接挂掉。后来我加了监控Trail 目录使用率超过阈值就告警同时给 Extract 配了TRANLOGOPTIONS的流控参数目标端不可用时能自动暂停而不是无限堆积。6.3 排查问题的通用思路OGG 的问题排查核心是看日志。每个进程都有自己的日志文件在dirrpt目录下。Extract 的日志叫ext_kafka.rptReplicat 的叫rep_kafka.rpt。日志里会记录进程启动、检查点更新、错误信息这些关键事件。排查顺序我一般是先看进程状态INFO EXTRACT、INFO REPLICAT再看日志报错然后看 Trail 文件有没有在增长最后看 Kafka 端有没有收到消息。这个顺序能快速定位问题出在哪一段。如果 Extract 正常但 Kafka 没消息问题就在 Replicat 或者 Kafka 连接如果 Extract 就报错那问题在源端。还有一个技巧是用STATS命令看统计信息STATS EXTRACT ext_kafka STATS REPLICAT rep_kafka它会告诉你抽了多少条、插了多少条、更新多少条、删除多少条和源端的实际变更量对一下就能知道有没有丢。7. 一些关于稳定性和运维的补充增量同步上线只是开始真正考验人的是长期运维。我总结了几条经验都是实际运行中攒下来的。监控要覆盖三层OGG 进程层、Trail 文件层、Kafka 消费层。进程层看进程是否存活、检查点是否在推进Trail 层看文件增长速度和磁盘使用率Kafka 层看消费 lag 和消息积压。任何一层出问题都要能及时告警。变更管理要规范。源表加列、改类型、加索引这些操作都可能影响 OGG。加列的话 Extract 参数要同步更新否则新列不同步改类型可能导致解析失败加索引一般没影响但也要观察。我的做法是源库任何 DDL 变更都走一个流程评估对 OGG 的影响后再执行。容灾要有预案。OGG 的检查点是它断点续传的基础检查点目录一定要备份。如果 Extract 挂了重启后它会从最后一个检查点继续不会丢数据也不会重复太多。但如果检查点文件损坏那就只能从更早的位置重抽可能造成大量重复。所以检查点目录我一般会做定期备份。最后说一个关于版本的问题。11g 虽然稳定但毕竟不是最新版本对接 Kafka 的一些新特性支持有限。如果条件允许升级到更高版本的 OGG 会省很多事Kafka Handler 的成熟度和配置灵活性都会好很多。但如果短期内不能升级11g 这套方案也完全能跑只是配置上要多花点心思把上面提到的这些细节都照顾到。我个人在实际操作中的体会是OGG 这套东西入门不难难的是把边界情况都想到。事务一致性、字符集、主键变更、Trail 堆积这些都不是配置文档里会重点强调的但恰恰是生产环境最容易出问题的地方。把这些问题提前想清楚、配好监控后面就能省下大量救火的时间。