Windows Oracle11g OGG to kafka

你的名字 2023-07-20 12:10 21阅读 0赞

本文主要参考文档和博客如下：

[https://dongkelun.com/2018/05/23/oggOracle2Kafka/][https_dongkelun.com_2018_05_23_oggOracle2Kafka]

[https://www.jianshu.com/p/b1e1fb623f15][https_www.jianshu.com_p_b1e1fb623f15]

原因是这样的：当我使用debezium插件对oracle进行cdc日志获取的时候发现我的oracle版本过低为11G，不支持

alter system set enable\_goldengate\_replication=true;

的属性（虽然不知有何影响），然后看了看debezium对于oracle连接器的说法为孵化阶段，所以就滚去OGG将oracle数据库ddl变更推到kafka了。

# 一、开发环境 #

源端：

Oracle Database 11g Enterprise Edition Release 11.2.0.1.0

Oracle GoldenGate Manager for Oracle

Version 12.2.0.2.3 OGGCORE\_12.2.0.2.0\_PLATFORMS\_170727.1430

Windows x64 (optimized),

目标端：

OGG\_BigData\_Linux\_x64\_19.1.0.0.1.tar

kafka2.12-2.4.0

# 二、 源端配置 #

## 2.1 Oracle配置，OGG解压安装 ##

1.  解压文件，安装OGG
2.  登录Oracle，启动oracle归档模式

sqlplus / as sysdba
    archive log list   --- 该命令为查看归档日志情况
    
    conn / as sysdba (以DBA身份连接数据库) 
    shutdown immediate (立即关闭数据库)
    startup mount (启动实例并加载数据库，但不打开)
    alter database archivelog; (更改数据库为归档模式)
    alter database open; (打开数据库)
    alter system archive log start; (启用自动归档)
    
    --查看辅助日志情况
    select force_logging, supplemental_log_data_min from v$database;
    
    alter database force logging;
    alter database add supplemental log data;
    
    --这里创建复制用户，便于管理
    SQL> create tablespace oggtbs datafile 'C:\app\Administrator\product\oggdata\oggtbs01.dbf' size 1000M autoextend on;
    
    Tablespace created.
    
    SQL>  create user ogg identified by ogg default tablespace oggtbs;
    
    User created.
    
    -- 根据实际给予相关权限，这里便于操作给予dba权限，实际需求中请给予正确的权限
    SQL> grant dba to ogg;
    
    Grant succeeded.

3. 启动OGG，进入到OGG安装目录，启动文件夹下的ggsci.exe程序

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MTQ2MTk5Mg_size_16_color_FFFFFF_t_70][]

4.在ggsci.exe界面中执行以下命令

dblogin userid ogg password ogg   --使用oracle复制用户进行登录
    
    登录成功显示
    
    Successfully logged into database.

下列该步骤可有可无。我试了下并没有影响：

GGSCI > edit param ./globals
    
    添加内容
    oggschema ogg

## 2.2 配置mgr管理器 ##

GGSCI > edit param mgr
    
    PORT 7809
    DYNAMICPORTLIST 7810-7909
    AUTORESTART EXTRACT *,RETRIES 5,WAITMINUTES 3
    PURGEOLDEXTRACTS ./dirdat/*,usecheckpoints, minkeepdays 3
    ACCESSRULE, PROG SERVER, ALLOW;             --记住以;结尾

上述说明：

PORT即mgr的默认监听端口；

DYNAMICPORTLIST动态端口列表，当指定的mgr端口不可用时，会在这个端口列表中选择一个，最大指定范围为256个；

AUTORESTART重启参数设置表示重启所有EXTRACT进程，最多5次，每次间隔3分钟；

PURGEOLDEXTRACTS即TRAIL文件的定期清理

下一步，添加复制表

GGSCI > add trandata test_ogg.test_ogg    --此处填写自己需要同步的表 为 库名.表名 的形式编写，注意不要加；否则会找不到该表
    
    Logging of supplemental redo data enabled for table TEST_OGG.TEST_OGG.
    
    GGSCI > info trandata test_ogg.test_ogg
    
    Logging of supplemental redo log data is enabled for table TEST_OGG.TEST_OGG.
    
    Columns supplementally logged for table TEST_OGG.TEST_OGG: ID

## 2.3 配置extract进程 ##

GGSCI > edit param extkafka   --具体名称自己取，便于管理
    extract extkafka
    dynamicresolution
    SETENV (ORACLE_SID = "orcl")
    SETENV (NLS_LANG = "american_america.AL32UTF8")    --此处的编码问题也是个坑，可能会GBK问题
    userid ogg,password ogg
    exttrail c:\ogg\dirdat\to       --此处为linux环境的话就填linux下的OGG位置，为windows即填windows位置
    table test_ogg.test_ogg;

说明：

第一行指定extract进程名称；

dynamicresolution动态解析；

SETENV设置环境变量，这里分别设置了Oracle数据库以及字符集；

userid ggs,password ggs即OGG连接Oracle数据库的帐号密码，这里使用创建的复制帐号；

exttrail定义trail文件的保存位置以及文件名，注意这里文件名只能是2个字母（即会出现to0000000000这样的文件），其余部分OGG会补齐；

table即复制表的表名，支持\*通配，必须以;结尾（这里为单表的形式进行测试）

** 添加extract进程：**

GGSCI > add extract extkafka,tranlog,begin now
    EXTRACT added.

**      添加trail文件的定义与extract进程绑定：**

GGSCI (ambari.master.com) 17> add exttrail C:\app\Administrator\ogg122023\dirdat\to,extract extkafka
    EXTTRAIL added.
    --注意此处的位置依旧是linux环境和windows环境的位置不同，/和\也不同

## 2.4 配置pump进程 ##

pump进程本质上来说也是一个extract，只不过他的作用仅仅是把trail文件传递到目标端，配置过程和extract进程类似，只是逻辑上称之为pump进程。

GGSCI (ambari.master.com) 18> edit param pukafka
    extract pukafka
    passthru
    dynamicresolution
    userid ogg,password ogg
    rmthost 192.168.44.129 mgrport 7809  --目标端的地址和端口
    rmttrail /opt/ogg/dirdat/to   --目标端的trail文件存储位置名称
    table test_ogg.test_ogg;

说明：第一行指定extract进程名称；

passthru即禁止OGG与Oracle交互，我们这里使用pump逻辑传输，故禁止即可；

dynamicresolution动态解析；

userid ogg,password ogg即OGG连接Oracle数据库的帐号密码；

rmthost和mgrhost即目标端(kafka)OGG的mgr服务的地址以及监听端口；

rmttrail即目标端trail文件存储位置以及名称。

**将本地trail文件和目标端的trail文件绑定到extract进程：**

--注意第一条是源端路径，第二条命令是目标端路径，注意linux和windows地址写法
    GGSCI > add extract pukafka,exttrailsource C:\app\Administrator\ogg122023\dirdat\to
    EXTRACT added.
    GGSCI > add rmttrail /opt/ogg/dirdat/to,extract pukafka
    RMTTRAIL added.

## 2.5 配置define文件 ##

Oracle与MySQL，Hadoop集群（HDFS，Hive，kafka等）等之间数据传输可以定义为异构数据类型的传输，故需要定义表之间的关系映射，在OGG命令行执行：

GGSCI > edit param test_ogg
    defsfile C:\app\Administrator\ogg122023\dirdef\test_ogg.test_ogg
    userid ogg,password ogg
    table test_ogg.test_ogg;

在OGG主目录下执行命令，生成映射文件：

注意：linux和windows下的地址问题。
    如果是linux下，则在ogg目录下执行该语句
    
    ./defgen paramfile dirprm/test_ogg.prm
    
    如果是windows环境下，则进入OGG目录下，按住shift加右键打开powershell，执行下面命令
    
    .\defgen.exe paramfile C:\app\Administrator\ogg122023\dirprm\test_oggg.prm

如下

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MTQ2MTk5Mg_size_16_color_FFFFFF_t_70 1][]

将生成的test\_ogg.test\_ogg（自己命名）的文件移动到目标端（好像目标端还没配置。。）的dirdef文件夹下。

此时，源端的配置基本完成。

# 三、目标端部署 #

我们假设已经装好了kafka环境。

与上述源端配置相同，此处是linux下的目标端，所以将压缩包解压完毕。

此处解压到/usr/local/tools/app/ogg目录下

tar -xvf ggs_Adapters_Linux_x64.tar

同理，进行ggsci命令行：

ggsci

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MTQ2MTk5Mg_size_16_color_FFFFFF_t_70 2][]

此处关于mgr配置基本与源端相似了

## 3.1 配置mgr管理器 ##

mgr文件配置

GGSCI >  edit param mgr
    PORT 7809
    DYNAMICPORTLIST 7810-7909
    AUTORESTART EXTRACT *,RETRIES 5,WAITMINUTES 3
    PURGEOLDEXTRACTS ./dirdat/*,usecheckpoints, minkeepdays 3
    ACCESSRULE, PROG SERVER, ALLOW;

## 3.2 配置checkpoint ##

edit  param  ./GLOBALS
    CHECKPOINTTABLE test_ogg.checkpoint

## 3.3 配置replicate进程 ##

GGSCI > edit param rekafka
    REPLICAT rekafka
    sourcedefs /opt/ogg/dirdef/test_ogg.test_ogg
    TARGETDB LIBFILE libggjava.so SET property=dirprm/kafka.props
    REPORTCOUNT EVERY 1 MINUTES, RATE 
    GROUPTRANSOPS 10000
    MAP test_ogg.test_ogg, TARGET test_ogg.test_ogg;

说明：REPLICATE rekafka定义rep进程名称；

sourcedefs即在4.6中在源服务器上做的表映射文件；

TARGETDB LIBFILE即定义kafka一些适配性的库文件以及配置文件，配置文件位于OGG主目录下的dirprm/kafka.props；

REPORTCOUNT即复制任务的报告生成频率；

GROUPTRANSOPS为以事务传输时，事务合并的单位，减少IO操作；

MAP即源端与目标端的映射关系。

注意：该版本下的dirprm目录下只有刚刚生成的mgr.mgr文件，并没有kafka.props等配置文件，所以需要去

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MTQ2MTk5Mg_size_16_color_FFFFFF_t_70 3][]

下面的

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MTQ2MTk5Mg_size_16_color_FFFFFF_t_70 4][]

拿到配置文件并复制到到dirprm文件夹下。

此处拉取kafka.props和custom\_kafka\_producer.properties文件。

![20200418094457406.png][]

## 3.4 配置kafka.props和custom\_kafka\_producer.properties ##

1. kafka.props

gg.handlerlist=kafkahandler //handler类型
    gg.handler.kafkahandler.type=kafka
    gg.handler.kafkahandler.KafkaProducerConfigFile=custom_kafka_producer.properties //kafka相关配置
    gg.handler.kafkahandler.topicMappingTemplate=test_ogg //kafka的topic名称，无需手动创建
    gg.handler.kafkahandler.format=json //传输文件的格式，支持json，xml等
    gg.handler.kafkahandler.mode=op  //OGG for Big Data中传输模式，即op为一次SQL传输一次，tx为一次事务传输一次
    gg.classpath=dirprm/:/opt/kafka_2.11-2.4.0/libs/*:/opt/ogg/:/opt/ogg/lib/*
    
    注意：gg.classpath配置kafka和ogg的libs位置，不然启动的时候读取不到相关jar包，会报错

2. custom\_kafka\_producer.properties

bootstrap.servers=192.168.44.129:9092  //kafkabroker的地址
    acks=1
    compression.type=gzip  //压缩类型
    reconnect.backoff.ms=1000  //重连延时
    value.serializer=org.apache.kafka.common.serialization.ByteArraySerializer
    key.serializer=org.apache.kafka.common.serialization.ByteArraySerializer
    batch.size=102400
    linger.ms=10000

## 3.5 添加trail文件到replicate进程 ##

GGSCI > add replicat rekafka exttrail /usr/local/tools/app/ogg/dirdat/to,checkpointtable test_ogg.checkpoint
    REPLICAT added.

# 四、进程启动 #

如果上述操作都完成并没有报错ERROR情况，则接下来进行最终测试。

源端目标端都在ggsci命令行下输入命令启动进程。

## 4.1 源端启动 ##

他们说按照顺序启动

start mgr
    start extkafka
    start pukafka

启动成功，则会看到三个窗口弹出，并无明显错误信息、无窗口闪退情况：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MTQ2MTk5Mg_size_16_color_FFFFFF_t_70 5][]

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MTQ2MTk5Mg_size_16_color_FFFFFF_t_70 6][]

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MTQ2MTk5Mg_size_16_color_FFFFFF_t_70 7][]

## 4.2 目标端启动 ##

start mgr
    start rekafka

在ggsci命令行中使用

GGSCI > info all

可以查看启动信息

例如：

![20200418095006163.png][]

看到自己启动的进程为RUNNING状态则为启动成功。

若有问题（窗口闪退，报错启动失败），可以查看系统日志：

1. windows下

ogg目录下的ggserr.log日志文件

也可以在ggsci下使用下列命令查看

view report extkafka --这个取决于你的名称是什么，比如是extkafka

2. linux下

与windows相同，但是对于kafka报错可以去

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MTQ2MTk5Mg_size_16_color_FFFFFF_t_70 8][]

# 五、 数据测试 #

连接oracle，使用刚刚创建的测试表进行插入等语句测试：

conn test_ogg/test_ogg
    insert into test_ogg values(1,'test');
    commit;
    update test_ogg set name='zhangsan' where id=1;
    commit;
    delete test_ogg where id=1;
    commit;

注意：这里一定要commit，才可以提交并监测到数据变化。

此时，发现源端和目标端的dirdat文件夹下生成了to0000的文件。

查看kafka的主题情况：

bin/kafka-topics.sh --list --zookeeper localhost:2181

主题创建成功。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MTQ2MTk5Mg_size_16_color_FFFFFF_t_70 9][]

消费该test\_ogg主题，即监控的表：

![20200418095248265.png][]

能读取到其变化，此时全部步骤完成。

# 六、 注意问题 #

## 6.1 源端和目标端的编码问题 ##

![20200418095325344.png][]

参考：[https://blog.csdn.net/u010457406/article/details/75646547][https_blog.csdn.net_u010457406_article_details_75646547]

SHUTDOWN IMMEDIATE; --同样需要关闭数据库
    STARTUP MOUNT EXCLUSIVE; 
    ALTER SYSTEM ENABLE RESTRICTED SESSION;
    ALTER SYSTEM SET JOB_QUEUE_PROCESSES=0;
    ALTER SYSTEM SET AQ_TM_PROCESSES=0;
    ALTER DATABASE OPEN;
    ALTER DATABASE NATIONAL CHARACTER SET INTERNAL_USEAL32UTF8; --此处加上了INTENAL_USER
    SHUTDOWN immediate;
    startup;

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MTQ2MTk5Mg_size_16_color_FFFFFF_t_70 10][]

## 6.2 kafka序列化配置问题 ##

custom\_kafka\_producer.properties配置文件中，序列化与反序列化中选择ByteArray。若是选择String则会报错，待解决。

ByteArraySerializer不影响消费（即乱码问题没有出现）

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MTQ2MTk5Mg_size_16_color_FFFFFF_t_70 11][]

## 6.3 多表配置 ##

如果想通配整个库的话，只需要把上面的配置所有表名的改为，如test\_ogg.test\_ogg改为 test\_ogg.,但是kafka的topic不能通配，所以需要把所有的表的数据放在一个topic即可，后面再用程序解析表名即可。

# 七、 总结 #

可以说查找了挺多的文章博客和官网文档说明，一开始根本找不到之前版本的OGG插件包，然后梯子出去官网下载的根本没有12版本的包，最后还是百度搜索进去才下载好的。

当我很开心看到了有大佬配置的文章，虽然是linux下的配置，但是还是挺舒服解决了这个配置问题。

所以这里是windows下源端的配置，总结到这里。

另外Oracle我想骂娘了，配置这么复杂真的有人用吗（哦，我用啊，那没事了），何况我一开始想好好写这个配置文章的，当我写到后面发现这么长，这么繁琐。当真不想写下去了。前面老哥完整记录下来真的辛苦了。

另外，关于这个配置，我没有进行多表、全库、正则匹配表的尝试。或许之后工作需要会进行填坑。

配置是真的繁琐！！！

文章或许有表名不同的情况，请自行修改。

[https_dongkelun.com_2018_05_23_oggOracle2Kafka]: https://dongkelun.com/2018/05/23/oggOracle2Kafka/
[https_www.jianshu.com_p_b1e1fb623f15]: https://www.jianshu.com/p/b1e1fb623f15
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MTQ2MTk5Mg_size_16_color_FFFFFF_t_70]: /images/20230528/04acefa6f5534c16b1675e54fc05aaf0.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MTQ2MTk5Mg_size_16_color_FFFFFF_t_70 1]: /images/20230528/3eb4bb48c44b4266bc139fa6b46cbc3c.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MTQ2MTk5Mg_size_16_color_FFFFFF_t_70 2]: /images/20230528/cd15e2a8537141b7aadcbcfa662c26b7.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MTQ2MTk5Mg_size_16_color_FFFFFF_t_70 3]: /images/20230528/34ba7dc2ec77456c8a376083ec64b548.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MTQ2MTk5Mg_size_16_color_FFFFFF_t_70 4]: /images/20230528/9cc5a417ae1747399af86e0d779c7ac2.png
[20200418094457406.png]: /images/20230528/e677b11ffac04d0fab0b8993d7d1948c.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MTQ2MTk5Mg_size_16_color_FFFFFF_t_70 5]: /images/20230528/92d5ca55c01b459783bc6ded0c1bacda.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MTQ2MTk5Mg_size_16_color_FFFFFF_t_70 6]: /images/20230528/f169b53fe3a6430f8717d20838ed742b.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MTQ2MTk5Mg_size_16_color_FFFFFF_t_70 7]: /images/20230528/41860e4691be47dc9f67353194a33e38.png
[20200418095006163.png]: /images/20230528/3ae9217d5ec64ea5baa24c7136748edb.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MTQ2MTk5Mg_size_16_color_FFFFFF_t_70 8]: /images/20230528/95a5fb4647b640c89cc948c2a599df11.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MTQ2MTk5Mg_size_16_color_FFFFFF_t_70 9]: /images/20230528/f1227b6d58c14ecc915531c21be79b6a.png
[20200418095248265.png]: /images/20230528/b993e8570775408092859c4026c1408b.png
[20200418095325344.png]: /images/20230528/e7bb4091e8a14065a77b4deac8f783b8.png
[https_blog.csdn.net_u010457406_article_details_75646547]: https://blog.csdn.net/u010457406/article/details/75646547
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MTQ2MTk5Mg_size_16_color_FFFFFF_t_70 10]: /images/20230528/39878fd81a2c4e45a8c3a02017b65a12.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MTQ2MTk5Mg_size_16_color_FFFFFF_t_70 11]: /images/20230528/17ee55b580c54605911cdcbe93164268.png