Oracle Goldengate实时复制Oracle数据到CDH Kafka

Oracle Goldengate（以下简称OGG）是业界广泛使用的准实时数据复制软件，主要基于数据库日志进行变更数据的抽取，也就是我们常说的CDC（change data capture）的能力，Goldengate的优势主要体现支持各种异构环境下的数据复制，对生产库的影响最低（基于日志读取，非Sqoop这样直接查询数据的方式，同时可以支持remote capture）。

OGG的架构图如下：

Goldengate能够支持源端数据抽取和目标数据投递的种类和平台非常的多，几乎涵盖了目前主流的各种数据平台。

可以通过Oracle的官网链接来查询兼容性列表情况：

OGG 18C：https://www.oracle.com/technetwork/middleware/data-integration/goldengate-12-3-x-cert-matrix-3424388.xls

OGG 12.3: https://www.oracle.com/technetwork/middleware/ogg-18-1-0-0-0-cert-matrix-5172014.xls

因为OGG 18c版本最近刚发布，从兼容性列表来看，还没有发布支持Hadoop平台的For Bigdata版本，所以本文选择源端用OGG 18c版进行Oracle数据库数据的抽取，而目标用OGG 12.3 for Bigdata投递到CDH 5.14中。

OGG 18c兼容性列表中支持Oracle的版本:

OGG 12.3 For Bigdata支持CDH的版本：

可以支持向HDFS、Hive、HBase、Flume、Kafka进行复制，做为系列文章的第一篇，我们首先选择向Kafka进行复制。

OGG下载链接：

https://www.oracle.com/technetwork/middleware/goldengate/downloads/index.html

源：

目标：

下面就开始Oracle Goldengate实现Oracle11204向CDH5.14 Kafka的数据复制链路

1.数据库前提条件：

• 调整数据库参数

alter system set enable_goldengate_replication=true scope=both;

• 开启数据库附加日志和force logging

alter database add supplemental log data;
alter database force logging;

设置后，查询数据库最小附加日志已经打开：

• 创建用户

需要在Oracle数据中创建一个Goldengate，用于保存OGG的元数据信息，建议为该用户配置独立的数据表空间；同时该用户的权限请参考链接：

https://docs.oracle.com/en/middleware/goldengate/core/18.1/oracle-db/establishing-oracle-goldengate-credentials.html#GUID-8207BAB7-A2AB-46CE-B7B5-14E75B0FDDEB

这里我们简化，直接授权DBA角色给该用户：

• 执行dbms_goldengate_auth.grant_admin_privilege

BEGIN
dbms_goldengate_auth.grant_admin_privilege(grantee => 'oggadmin',
privilege_type => 'CAPTURE',grant_select_privileges => TRUE,do_grants => TRUE);
END;
/

• 配置数据库stream pool

OGG对oracle数据库的抽取模式有2种：classic模式和Integrated模式。从OGG 18c开始classic模式不再有新的发展，从支持层面Integrated模式比classic模式支持的对象类型更广，限制更少，同时可以支持部署远程抽取模式，所以我们后面用配置的是Integrated模式。

而Integrated模式是直接对接logmining server来获取数据的变化，是通过XStream接口接入的：

这里只有32M，在后续配置Integrated模式中在OGG抽取参数需要配置max_sga_size参数（如果不配置，默认是1GB）的值一个不注意就会有冲突，这里可以先调整streams_pool_size=512M，而可以设置max_sga_size为200MB，具体如何调整，这个我们后续的文章会解释。

2.源端OGG安装

可以支持GUI和静默安装，我们这里使用静默安装。

修改其中几个参数：

INSTALL_OPTION=ORA11g
SOFTWARE_LOCATION=/home/oracle/ogghome
INVENTORY_LOCATION=/home/oracle/ogghome/inventory
UNIX_GROUP_NAME=oinstall

• 开始静默安装

注意：ogg的安装目录必须为空，所以创建一个新的目录：/home/oracle/ogghome

运行安装：

./runInstaller -silent -responseFile /home/oracle/oggsoftware/fbo_ggs_Linux_x64_shiphome/Disk1/response/oggcore.rsp

• 配置环境变量

Linux下设置：

• 创建ogg相关目录

3.配置源端MGR进程

PORT 7809
DYNAMICPORTLIST 7810-7849
PURGEOLDEXTRACTS ./dirdat/*, USECHECKPOINTS, MINKEEPDAYS 3
LAGREPORTHOURS 1
LAGINFOMINUTES 30
LAGCRITICALMINUTES 45

启动MGR进程：

4.配置credentialstore，使得用户的密码不是明文

add credentialstore
alter credentialstore add user oggadmin password oracle alias oggadmin
info credentialstore

5.配置源端抽取和传输进程（Integrated模式）

• 查询源端数据库字符集（该步骤可选）

查询一下oracle字符集：

• 增加要复制对象的附件日志

检查一下表的trandata是否enable：

• 编辑源端抽取进程参数文件

extract exta
useridalias oggadmin 
setenv (ORACLE_HOME = /u01/app/oracle/product/11.2.0/dbhome_1)
setenv (ORACLE_SID = testdb1)
setenv (NLS_LANG = AMERICAN_AMERICA.AL32UTF8)
exttrail ./dirdat/st
discardfile ./dirrpt/exta.dsc,APPEND,MEGABYTES 100
tranlogoptions integratedparams (max_sga_size 200, parallelism 1)
reportcount every 1 minutes, rate
logallsupcols
ddl include mapped #这里启用ddl抽取，范围是mapped
TABLE scott.*;

• 编辑源端传输进程参数文件

extract dpa
rmthost allinone.thomasgu.com mgrport 7809
rmttrail ./dirdat/rt
passthru
reportcount every 1 minutes, rate
table scott.*

Note:提前在/etc/hosts中配置好目标主机名的解析，或者直接用目标主机的ip也可以。

• 创建抽取进程和传输进程

dblogin useridalias oggadmin
register extract exta database

add extract exta, integrated tranlog, begin now
add exttrail ./dirdat/st, extract exta, megabytes 100

add extract dpa, exttrailsource ./dirdat/st
add rmttrial ./dirdat/rt, extract dpa, megabytes 100

• 启动源端抽取进程exta

检查一下为什么会stop：

执行一下

GGSCI>view report exta

因为源库是11204版，没有打过任何的patch，所以这里需要apply一下p17030189_112040_Generic的patch

unzip p17030189_112040_Generic.zip
/u01/app/oracle/product/11.2.0/dbhome_1/OPatch/opatch lsinventory
cd 17030189
/u01/app/oracle/product/11.2.0/dbhome_1/OPatch/opatch apply
/u01/app/oracle/product/11.2.0/dbhome_1/OPatch/opatch lsinventory

数据库中还需要运行：

$ sqlplus /nolog
SQL> CONNECT / AS SYSDBA
SQL> @?/sqlpatch/17030189/postinstall.sql

此时再启动抽取进程：

• 启动源端传输进程

直接在GGSCI中执行:
GGSCI>start DPA

Note:要先保证目标OGG的MGR进程启动好之后，源端的传输进程才能运行正常，否则状态会stop，看report记录会提示连接不上目标的collector进程。

运行OGG For Bigdata需要JDK1.8的环境。

因为CDH 5.14安装默认使用了JDK1.7的包，所以我们部署采用了remote delivery的方式向Kafka进行投递，将OGG for Bigdata也安装在了Oracle所在的这台机器上，这样对CDH集群也是无侵入的，在生产环境这样的方式应该是更加合适的。

也就说现在OGG For Bigdata和OGG for Oracle安装在了同一台机器上。

所以这里需要注意：

（1）前面配置传输进程dpa就不用启动了，因为用不到了，目标OGG for bigdata进程直接读取OGG for oracle的抽取进程落地下来的队列文件（st作为前缀的文件）；

（2）OGG for bigdata的MGR进程就不能使用和OGG for Oracle相同的端口号了，因为一台主机上是不能启动两个同样port的MGR的。

（3）正常情况，如果OGG for oracle和OGG for Bigdata不在同一台主机上，就按照前面配置的方式，目标MGR进程和源端MGR进程配置相同的port，dpa进程rmthost参数连接目标同样的MGR端口即可。

1.目标端安装JDK 1.8

vi /etc/profile
export JAVA_HOME=/usr/java/jdk1.8.0_162
export JRE_HOME=/usr/java/jdk1.8.0_162/jre
export PATH=$JAVA_HOME/bin:$JRE_HOME/bin:$PATH
export CLASSPATH=$JAVA_HOME/lib:$JRE_HOME/lib:$CLASSPATH

2.配置OGG For Bigdata

• 解压OGG for bigdata到一个新的目录oggbigdatahome

unzip OGG_BigData_Linux_x64_12.3.2.1.1.zip -d oggbigdatahome 
cd oggbigdatahome
tar xvf OGG_BigData_Linux_x64_12.3.2.1.1.tar

执行for bigdata的ggsci的时候，会提示libjvm.so加载错误。

所以需要把libjvm.so文件也加入到环境变量LD_LIBRARY_PATH中，libjvm.so是在：

/usr/java/jdk1.8.0_162/jre/lib/amd64/server/目录下

• 创建OGG for bigdata相关目录

• 创建OGG for bigdata的MGR进程（注意前面的注意说明，因为把OGG for Bigdata和OGG for Oracle安装在同一台主机上，这里用了和源端MGR不一样的port）

PORT 8809
DYNAMICPORTLIST 8810-8849
PURGEOLDEXTRACTS ./dirdat/*, USECHECKPOINTS, MINKEEPDAYS 3
LAGREPORTHOURS 1
LAGINFOMINUTES 30
LAGCRITICALMINUTES 45

• 准备目标投递到Kafka的进程的相关文件

从OGG for Bigdata目录可以看出有两种Kafka的投递方法：

在/home/oracle/oggbidatahome/dirprm下编辑上面的3个文件：

具体参数的含义可以查询OGG for Bigdata官方手册，这里只修改了几个必要的参数。

（1）  rkafka.prm

（2）  kafka.props

（3）custome_kafka_producer.properties

• 复制cdh中的kafka jar文件到OGG for bigdata中定义的目录下

只需要一个jar包即可：kafka-clients-0.10.2-kafka-2.2.0.jar

• 创建CDH kafka中的topic

/opt/cloudera/parcels/KAFKA-2.2.0-1.2.2.0.p0.68/lib/kafka/bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 2 --topic oggtopic1

这里设置2个partition是为了后面一些参数校验。

• 创建目标端投递进程

add replicat rkafka, exttrail /home/oracle/ogghome/dirdat/st
注意：这里是读取的ogg for oracle的抽取进程落地的队列文件st，原因已经解释多次了。

到这里就完成源和目标的OGG配置，下面我们简单测试一下复制链路是否正常。

源端提交一些sql语句，看是否能够复制到kafka topic中：

conn scott/oracle
insert into dept values(1,'A','SH');
insert into dept values(2,'B','SH');
insert into dept values(3,'C','SH');
insert into dept values(4,'D','SH');
insert into dept values(5,'E','SH');
insert into dept values(6,'F','SH');
insert into dept values(7,'G','SH');
insert into dept values(8,'H','SH');
insert into dept values(9,'I','SH');
insert into dept values(11,'J','SH');
commit;

上面是一个小的JAVA文件，用于监控消费端的信息，可以看到，已经消费到了10条消息，且按照partition基本是均匀分配的。在OGG for bigdata中有很多参数可以控制kafka分发消息的规则，后续文章我们也会简单介绍一下。

• 从cloudera manager中检查

测试集群只启动了zookeeper和kafka：

• 从OGG for Bigdata检查复制情况

明显有10条insert操作提交到kafka

通过配置OGG for Oracle和OGG for bigdata，完成基于日志抽取方式将数据变更从Oracle复制到CDH Kafka，可以解决直接通过sqoop类似直接读取数据的手段对源库性能开销的问题，同时是准实时的抽取，延迟低，复制效率高。

后续我们会介绍如何使用OGG完成全量数据抽取到Kafka，以及相关参数配置，实现不同的partition分发策略，kerberos环境下的复制；以及到Hbase、Hive等的复制。

喜欢 (0) 打赏 分享

感谢您的支持，我会继续努力的!

扫码打赏，你说多少就多少

打开支付宝扫一扫，即可进行扫码打赏哦