Version 12.3.0.1.0 OGGCORE_12.3.0.1.0_PLATFORMS_170721.0154_FBO

Oracle Release 12.2.0.1.0 Production 

Version 12.3.0.1.0 OGGCORE_OGGADP.12.3.0.1.0GA_PLATFORMS_170828.1608

kafka_2.12-2.0.0

zookeeper-3.4.13

1.编辑profile

[[email protected] bin]$ vim /etc/profile

2.添加如下

export OGG_HOME=/oracle/app/ogg123

export LD_LIBRARY_PATH=$ORACLE_HOME/lib:/usr/lib

export PATH=$OGG_HOME:$PATH

3.使之生效

[[email protected] bin]$ source /etc/profile

4.测试

[[email protected] bin]$ ggsci

1.编辑profile

[[email protected]]# $ vim /etc/profile

2.添加如下

export OGG_HOME=/usr/local/ogg
export LD_LIBRARY_PATH=$JAVA_HOME/jre/lib/amd64:

$JAVA_HOME/jre/lib/amd64/server:$JAVA_HOME/jre/lib/amd64/libjsig.so:

$JAVA_HOME/jre/lib/amd64/server/libjvm.so:$OGG_HOME/lib
export PATH=$OGG_HOME:$PATH

3.使之生效

[[email protected]]$ source /etc/profile

4.测试

[[email protected]]$ ggsci

1.为Oracle创建PDB 公用用户C##CDCCONNECT，并授予指定容器数据库DBA权限
2.为Oracle数据库启用OGG支持
SQL> ALTER SYSTEM SET ENABLE_GOLDENGATE_REPLICATION = TRUE SCOPE=BOTH;

3.启用最小补充日志记录与强制日志记录模式（必须开启）

[OGG基于辅助日志等进行实时传输，故需要打开相关日志确保可获取事务内容]

3.1登录sql plus
   查询当前状态，如果皆为YES则已开启，可直接跳至3. Oracle启用架构级补充日志继续操作
   SELECT supplemental_log_data_min, force_logging FROM v$database;
3.2.开启命令：
  SQL> ALTER DATABASE ADD SUPPLEMENTAL LOG DATA;

  SQL> ALTER DATABASE FORCE LOGGING;
3.3.开启后验证状态
  SELECT supplemental_log_data_min, force_logging FROM v$database;
3.4.开启后切换日志文件
  SQL> ALTER SYSTEM SWITCH LOGFILE;

4.启用架构级补充日志

4.1.tnsnames.ora配置PDB连接地址
4.2.登录数据库
  DBLOGIN USERID C##[email protected],password CDCCONNECT
4.3.启用补充日志
  ADD SCHEMATRANDATA schema ALLCOLS

5.开启挖掘服务器（必须开启）
5.1.登录数据库【需要使用CDB$root进行登录】
  DBLOGIN USERID C##CDCCONNECT password CDCCONNECT
5.2.注册挖掘服务器
  REGISTER EXTRACT extkafka DATABASE CONTAINER (zoehdc)

GGSCI (zoedb41 as C##[email protected]/ZOEHDC) 42> edit params mgr

--默认监听端口

PORT 7809

--动态端口列表，当默认监听端口不可用时，会在端口列表中选择一个，范围为256个
DYNAMICPORTLIST 7810-7909
AUTOSTART ER *

--重启参数设置表示重启所有EXTRACT进程，最多3次，每次间隔5分钟
AUTORESTART ER *,RETRIES 3,WAITMINUTES 5,RESETMINUTES 60

--TRAIL文件的定期清理
PURGEOLDEXTRACTS ./dirdat/*,USECHECKPOINTS,MINKEEPDAYS 7

GGSCI (localhost.localdomain) 54> edit params mgr

PORT 7809
DYNAMICPORTLIST 7810-7909
AUTORESTART EXTRACT *,RETRIES 5,WAITMINUTES 3
PURGEOLDEXTRACTS ./dirdat/*,usecheckpoints, minkeepdays 3

GGSCI (zoedb41 as C##[email protected]/ZOEHDC) 49> edit param extkafka

--进程名称

extract extkafka

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

--SETENV (ORACLE_SID = "zoehdc")
SETENV (NLS_LANG = "AMERICAN_AMERICA.ZHS16GBK")

--OGG连接Oracle数据库的帐号密码
userid  C##CDCCONNECT,password CDCCONNECT
DBOPTIONS ALLOWUNUSEDCOLUMN
discardfile ./dirrpt/exkafka.dsc,purge

--定义trail文件的保存位置以及文件名，文件名只能是2个字母，其余部分OGG会补齐
exttrail ./dirdat/to
FETCHOPTIONS FETCHPKUPDATECOLS

SOURCECATALOG ZOEHDC

--复制表的表名，支持*通配，必须以;结尾
table ODS.PAT_BASIC_INFO;

使用PDB用户登录

GGSCI (zoedb41) 1>DBLOGIN USERID C##CDCCONNECT password CDCCONNECT

添加进程

GGSCI (zoedb41 as C##[email protected]/ZOEHDC) 2>add extract extkafka,tranlog,begin now

EXTRACT added.

添加trail文件与extract进程绑定

GGSCI (zoedb41 as C##[email protected]/ZOEHDC)3>add exttrail ./dirdat/to,extract extkafka

EXTTRAIL added.

GGSCI (zoedb41 as C##[email protected]/ZOEHDC) 60> edit params pumkafka

--指定extract进程名称
extract pumkafka

SETENV(NLS_LANG ="AMERICAN_AMERICA.ZHS16GBK")

--OGG连接Oracle数据库的帐号密码
userid C##CDCCONNECT,password CDCCONNECT

--目标端(kafka)OGG的mgr服务的地址以及监听端口
rmthost 192.168.3.63 mgrport 7809

--目标端trail文件存储位置以及名称
rmttrail /usr/local/ogg/dirdat/to

--复制表的表名，支持*通配，必须以;结尾
SOURCECATALOG ZOEHDC
table ODS.PAT_BASIC_INFO;

同extract进程一样：

GGSCI (zoedb41 as C##[email protected]/ZOEHDC) 1> add extract pumkafka,exttrailsource ./dirdat/to

EXTRACT added.

GGSCI (zoedb41 as C##[email protected]/ZOEHDC) 2> add rmttrail ./dirdat/to,extract pumkafka

RMTTRAIL added.

GGSCI (zoedb41 as C##[email protected]/ZOEHDC) 79> edit params ormkafka

--Oracle与MySQL，Hadoop集群（HDFS，Hive，kafka等）等之间数据传输可以定义为异构数据类型的传输，

--故需要定义表之间的关系映射
defsfile /oracle/app/ogg123/dirdef/ods.PAT_BASIC_INFO

--同步的表信息
userid [email protected],password ods

table ods.PAT_BASIC_INFO;

在OGG主目录下执行(oracle用户)：

[[email protected] ogg123]$ ./defgen paramfile ./dirprm/ormkafka.prm

检查生成的define文件是否存在

[[email protected] dirdef]$ ll -l /oracle/app/ogg123/dirdef

将生成的/oracle/app/ogg123/dirdef/ods.PAT_BASIC_INFO发送的目标端ogg目录下的dirdef里：

[email protected] dirdef]$ scp -r /oracle/app/ogg123/dirdef/ods.PAT_BASIC_INFO

[email protected]:/usr/local/ogg/dirdef/

检查从源端生成的define文件是否存在

[[email protected]]# ll -l /usr/local/ogg/dirdef/

GGSCI (localhost.localdomain) 1> edit params rekafka

--rep进程名称
REPLICAT rekafka

--在源服务器上定义的表映射文件ormkafka
sourcedefs /usr/local/ogg/dirdef/ods.PAT_BASIC_INFO

--定义kafka一些适配性的库文件以及配置文件，配置文件位于OGG主目录下的dirprm/kafka.props
TARGETDB LIBFILE libggjava.so SET property=dirprm/kafka.props

--复制任务的报告生成频率
--REPORTCOUNT EVERY 1 MINUTES, RATE 

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

--源端与目标端的映射关系
SOURCECATALOG ZOEHDC
MAP ODS.PAT_BASIC_INFO, TARGET ODS.PAT_BASIC_INFO;

添加trail文件与replicat进程绑定

GGSCI (localhost.localdomain) 2> add replicat rekafka exttrail ./dirdat/to,checkpointtable ormkafka.checkpoint

REPLICAT added.

进入config/server.properties

[[email protected] kafka_2.12-2.0.0]# vi config/server.properties 

编辑一下属性

#设置监听地址

listeners=PLAINTEXT://192.168.3.63:9092

配置kafka.props

[[email protected] ogg]# cd /usr/local/ogg/dirprm/

[[email protected] ogg]# vim ./dirprm/kafka.props 
gg.handlerlist=kafkahandler
gg.handler.kafkahandler.type=kafka
gg.handler.kafkahandler.KafkaProducerConfigFile=custom_kafka_producer.properties
gg.handler.kafkahandler.topicMappingTemplate=test_ogg
gg.handler.kafkahandler.format=json
gg.handler.kafkahandler.mode=op
gg.classpath=dirprm/:/usr/local/kafka_2.12-2.0.0/libs/*:/usr/local/ogg/:/usr/local/ogg/lib/*

配置custom_kafka_producer.properties

[[email protected] ogg]# vim ./dirprm/custom_kafka_producer.properties
bootstrap.servers=192.168.3.63:9092
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

注意：ogg不识别注释，所以properties文件中不能有注释

start mgr 

start extkafka

start pumkafka

GGSCI (zoedb41 as C##[email protected]/ZOEHDC) 1> info all

Program     Status      Group       Lag at Chkpt  Time Since Chkpt

MANAGER     RUNNING                                           
EXTRACT     RUNNING     EXTKAFKA    00:00:00      00:00:08    
EXTRACT     RUNNING     PUMKAFKA    00:00:00      00:00:00    

start mgr

start rekafka

GGSCI (localhost.localdomain) 1> info all 

Program     Status      Group       Lag at Chkpt  Time Since Chkpt

MANAGER     RUNNING                                           
REPLICAT    RUNNING     REKAFKA     00:00:00      00:00:06