使用OGG将oracle增量数据实时同步至kafka（无安全认证模式）

本文将简单讲下使用ogg将oracle增量数据实时同步至kafka（无安全认证模式）的配置操作，后续也将更新在kafka试用不同认证模式下对应的ogg的配置方式。 一、环境描述 1.1 环境 简介

注：下面操作展示不涉及kafka集群搭建，默认 kafka 集群已经搭建完 。

二 、 检查生产库

2 .1附加日志检查 SQL> select supplemental_log_data_min from v$database; SUPPLEME -------- YES                       -------------若不是YES，需要打开 打开语句： SQL> alter database add supplemental log data; SQL> ALTER SYSTEM SWITCH LOGFILE;

2 .2打开 force log SQL> select force_logging from v$database; FOR --- YES SQL> ALTER DATABASE FORCE LOGGING;

2 .3查询压缩表SQL> select owner,table_name from dba_tables where owner in ('CBSRUN2') and compression='ENABLED';SQL> select table_owner,table_name from dba_tab_partitions where table_owner in ('CBSRUN2') and compression='ENABLED';

2 .4查询不支持的列的表格

SQL> select owner,table_name,column_name,data_type from dba_tab_columns where owner in ('CBSRUN2') and data_type in ('ANYDATA','ANYDATASET','ANYTYPE','BFILE','BINARY_INTEGER','MLSLABEL','PLS_INTEGER','TIMEZONE_ABBR','TIMEZONE_REGION','URITYPE,UROWID');

2 .5 查询 物化 视 图 

  查看要复制的表的日志信息是完整的，确保是logging。把nologing变成logging。 SQL> select owner,table_name from dba_tables where owner in ('CBSRUN2') and logging='NO' and temporary='N'; SQL> alter table xue.tt1291 logging; SQL> select object_name,owner from dba_objects where object_type='MATERIALIZED VIEW';

2 .6生产库 创建管理用户 --创建管理用户 SQL> create tablespace odc_tps datafile '+DATA/trac/datafile/odc01.dbf' size 100M autoextend on; SQL> create user odc identified by odc default tablespace odc_tps; --授权，DBA权限在安装成功后可以收回 GRANT CONNECT TO odc; GRANT ALTER ANY TABLE TO odc; GRANT ALTER SESSION TO odc; GRANT CREATE SESSION TO odc; GRANT FLASHBACK ANY TABLE TO odc; GRANT SELECT ANY DICTIONARY TO odc; GRANT SELECT ANY TABLE TO odc; GRANT RESOURCE TO odc; GRANT DBA TO odc; --安装成功后可以收回dba权限，但是要授予UNLIMITED TABLESPACE权限。 GRANT UNLIMITED TABLESPACE TO odc; 或者alter user odc quota unlimited on users; 三 、 安装软件 生产和备份端安装步骤一样，只有软件及环境变量可能不一样 3 .1创建安装目录或文件系统

  # su - oracle $ mkdir /backup/ogg

3 .2配置 library环境变量

LINUX  :  LIBRARY_PATH例如：export LIBRARY_PATH=/backup/ogg:$LIBPATHexoprt LIBRARY_PATH=$ORACLE_HOME/lib:$LIBPATH

3 .3安装软件

1. 解压软件安装包$ cd /odc$ unzip ogg112101_fbo_ggs_Linux_x64_ora11g_64bit.zip$ tar xf fbo_ggs_Linux_x64_ora11g_64bit.tar

2. 创建 子工作目录解压并解包成功后在安装目录（如：/odc）执行$ ./ggsciggsci>create subdirs

以上所创建的目录的作用：

3. 创建管理 进程

创建 管理进程 所需 参数：

具体 设置如 下 ： ggsci>edit param mgr 输入： port 7809 DYNAMICPORTLIST 7800-7820 PURGEOLDEXTRACTS ./dirdat/cz*, USECHECKPOINTS, MINKEEPHOURS 24 autorestart extract * retries 10 waitminutes 10 ggsci>edit param ./GLOBALS 输入： GGSCHEMA odc CHECKPOINTTABLE odc.ggs_checkpoint ---保存退出  ggsci>start mgr

  四、 为表格添加 supplement log

  ggsci>dblogin userid odc password odc   ggsci>add trandata CBSRUN2.SAVJ_ATOMJOUR0ggsci>info trandata CBSRUN2.SAVJ_ATOMJOUR0

五 、 创建抽取进程

抽取进程要 配置的参数：

具体参数 设置如 下 ：

  ggsci>add extract cz_ext tranlog threads 2 begin now ggsci>add exttrail ./dirdat/cz extract cz_ext 注： tranlog：表示数据抓取的来源是数据库的redo数据。 threads 2：表示我们数据库有多少个redo threads，单实例基本上是1或者不设，rac就自己设置了，配置过RAC的基本都了解。 begin now：表示我们在启动这个抓取进程的就去抓取数据。 ./dirdat：表示trail文件的目录 sm：trail文件的前缀 extract sm_ext：值指定给那个进程用的（sm_ext）。 ggsci>edit param cz_ext 参数： extract cz_ext setenv (NLS_LANG="SIMPLIFIED CHINESE_CHINA.ZHS16GBK") userid odc,password odc exttrail ./dirdat/cz tranlogoptions dblogreader FETCHOPTIONS FETCHPKUPDATECOLS table oggtest. * ; ggsci>start cz_ext

-- 观察，能够挖掘再继续下面的操作

  六、 增长 pump进程

传输进程要 配置的参数：

  ggsci>add extract cz_dmp EXTTRAILSOURCE ./dirdat/cz ggsci>ADD RMTTRAIL ./dirdat/cz, EXTRACT cz_dmp 注： EXTTRAILSOURCE：指定提取文件作为数据源 ADD RMTTRAIL：在目标数据库上创建一个trail ggsci>edit param cz_dmp 参数： extract cz_dmp userid odc,password odc rmthost 192.168.238.141, mgrport 7809 rmttrail ./dirdat/cz passthru table oggtest. *; ggsci>start sm_dmp

七、目标 kafka端创建 replicat进程

  创建测试topic ./kafka-topics.sh --create --zookeeper 192.168.238.141:2181,192.168.238.142:2181,192.168.238.143:2181 --replication-factor 3 --partitions 3 --topic oggtest 模拟测试任务发布订阅 生产者 ./kafka-console-producer.sh --broker-list node1:9092,node2:9092,node3:9092 --topic oggtest 消费者 ./kafka-console-consumer.sh --bootstrap-server node1:9092 --topic oggtest --from-beginning 确认kafka可用正常使用后，创建ogg 应用进程，将源端增量数据同步到kafka集群 具体参数如下： ggsci> Add replicat rep_ka exttrail ./dirdat/cz REPLICAT rep_ka sourcedefs ./dirdef/kafka.def TARGETDB LIBFILE libggjava.so SET property= dirprm/kafka.props REPORTCOUNT EVERY 1 MINUTES, RATE GROUPTRANSOPS 10000 MAP oggtest .t_ogg , TARGET oggtest.t_ogg; 从/ogg/AdapterExamples/big-data/kafka中将kafka.props，custom_kafka_producer.properties复制到/ogg/dirprm并根据实际情况修改 kafka.props配置如下 gg.handlerlist = kafkahandler  //handler 类型 gg.handler.kafkahandler.type=kafka gg.handler.kafkahandler.KafkaProducerConfigFile=custom_kafka_producer.properties  //kafka 相关配置 #gg.handler.kafkahandler.TopicName=oggtest  //kafka 的 topic 名称，无需手动创建 gg.handler.kafkahandler.topicMappingTemplate=oggtest gg.handler.kafkahandler.format=json  // 传输文件的格式，支持 json ， xml 等 gg.handler.kafkahandler.mode=op  //OGG for Big Data 中传输模式，即 op 为一次 SQL 传输一次， tx 为一次事务传输一次 #Sample gg.classpath for Apache Kafka gg.classpath=dirprm/:/kafka/kafka_2.12-1.1.1/libs/*:/ogg/:/ogg/lib/*  

// 相关库文件的引用   custom_kafka_producer.properties配置如下 bootstrap.servers=192.168.238.141:9092,192.168.238.142:9092,192.168.238.143:9092   //kafkabroker 的地址 acks=1 reconnect.backoff.ms=1000  // 重连延时 value.serializer=org.apache.kafka.common.serialization.ByteArraySerializer key.serializer=org.apache.kafka.common.serialization.ByteArraySerializer # 100KB per partition batch.size=102400 linger.ms=10000 配置完后，启动应用进程观察 ggsci> start rep_kaSending START request to MANAGER ...REPLICAT REP_KA starting 八、测试同步效果

测试方法比较简单，直接在源端的数据表中insert，update，delete操作即可。

SQL> conn oggtest/oggtest

Connected.

SQL> select * from t_ogg;

no rows selected

SQL> desc t_ogg;

 Name                                      Null?    Type

 ----------------------------------------- -------- ----------------------------

 ID                                        NOT NULL NUMBER(38)

 TEXT_NAME                                          VARCHAR2(20)

SQL> insert into t_ogg values(1,'test');

1 row created.

SQL> commit;

Commit complete.

查看源端trail文件状态

$ ls -l /backup/ogg/dirdat/tc*

-rw-rw-rw- 1 oracle oinstall 1180 Nov  9 17:05 /backup/ogg/dirdat/tc000000

查看目标端trail文件状态

# ls -l /data/gg/dirdat/tc*      

-rw-r----- 1 root root 1217 Nov  9 17:05 /data/gg/dirdat/tc000000

检查实时同步到kafka的效果，在Oracle源端更新表的同时，使用kafka客户端自带的脚本去查看这里配置的ggtopic这个kafkatopic下的消息：

SQL> insert into t_ogg values(2,'test2');

1 row created.

SQL> commit;

Commit complete.

目标端Kafka的同步情况：

# bin/kafka-console-consumer.sh --zookeeper  192.168.238.141:2181  --

from-beginning --topic ggtopic

{"table":"OGGTEST.T_OGG","op_type":"I","op_ts":"2025-11-09

09:05:25.067082","current_ts":"2025-11-

09T17:59:20.943000","pos":"00000000000000001080","after":

{"ID":"1","TEXT_NAME":"test"}}

{"table":" OGGTEST.T_OGG","op_type":"I","op_ts":"2025-11-09

10:02:06.827204","current_ts":"2025-11-

09T18:02:12.323000","pos":"00000000000000001217","after":

{"ID":"2","TEXT_NAME":"test2"}}

可以看到，Oracle的增量数据已实时同步到Kafka。