一、源端(oracle数据库端)---同这篇文章源端配置http://www.jianshu.com/p/53882229b70e
#Extract进程
#一定要记得同步之前要开启表的全补充日志
#alter table tb_name add supplemental log data (all) columns;
GGSCI (zwjfdb3) 7> view param EZWJFBOR
EXTRACT EZWJFBOR
SETENV (NLS_LANG="AMERICAN_AMERICA.AL32UTF8")
SETENV (ORACLE_HOME = "/u01/app/oracle/product/11.2.0/db_1")
SETENV (ORACLE_SID = "zwjfdb3")
--捕获 truncate 操作
gettruncates
--定义discardfile文件位置，如果处理中有记录出错会写入到此文件中
DISCARDFILE ./dirrpt/ezwjfbor.dsc, APPEND, MEGABYTES 1024
--动态解析表名
DYNAMICRESOLUTION
--获取更新之前数据
GETUPDATEBEFORES
--当抽取进程遇到一个没有使用的字段时只生成一个警告，进程会继续执行而不会被异常终止（abend）
DBOPTIONS ALLOWUNUSEDCOLUMN
--每隔30分钟报告一次从程序开始到现在的抽取进程或者复制进程的事物记录数，并汇报进程的统计信息
REPORTCOUNT EVERY 30 MINUTES, RATE
--每隔3分钟检查一下大事务，超过2小时还没结束的进行报告
WARNLONGTRANS 2h,CHECKINTERVAL 3m
--不会从闪回日志中获取数据
FETCHOPTIONS NOUSESNAPSHOT
USERID xxxxxx,PASSWORD xxxxxx
EXTTRAIL ./dirdat/zb
TABLE xx.xx;
TABLE xx.xx;
#添加抽取进程
GGSCI (zwjfdb3) 11> add extract EZWJFBOR,TRANLOG, BEGIN NOW
EXTRACT added.
#定义trail文件
GGSCI (zwjfdb3) 12> ADD EXTTRAIL ./dirdat/zb,EXTRACT EZWJFBOR, MEGABYTES 200
EXTTRAIL added.
#pump extract进程
GGSCI (zwjfdb3) 8> view param PZWJFBOR
EXTRACT PZWJFBOR
SETENV (NLS_LANG = "AMERICAN_AMERICA.AL32UTF8")
PASSTHRU
DYNAMICRESOLUTION
RMTHOST xx.xx.xx.xx,MGRPOT 7809
RMTTRAIL ./dirdat/zb
TABLE xx.xx;
TABLE xx.xx;
#添加pump捕获组
GGSCI (zwjfdb3) 23> ADD EXTRACT PZWJFBOR,EXTTRAILSOURCE ./dirdat/zb
EXTRACT added.
#定义pump trail文件
GGSCI (zwjfdb3) 25> ADD RMTTRAIL ./dirdat/zb,EXTRACT PZWJFBOR, MEGABYTES 200
RMTTRAIL added.
#启动进程
GGSCI (zwjfdb3) 8> start EXTRACT EZWJFBOR
GGSCI (zwjfdb3) 8> start EXTRACT PZWJFBOR
GGSCI (zwjfdb3) 9> info all
Program Status Group Lag at Chkpt Time Since Chkpt
MANAGER RUNNING
EXTRACT RUNNING EZWJFBOR 00:00:00 00:00:08
EXTRACT RUNNING PZWJFBOR 00:00:00 00:00:16
#传递表结构
GGSCI (zwjfdb3) 4> view param defgen
DEFSFILE dirdef/source.def, PURGE
USERID xxxxxx, PASSWORD xxxx
TABLE xx.xx;
TABLE xx.xx;
[oracle@zwjfdb3 12.2]$ ./defgen paramfile dirprm/defgen.prm
***********************************************************************
Oracle GoldenGate Table Definition Generator for Oracle
Version 11.2.1.0.3 14400833 OGGCORE_11.2.1.0.3_PLATFORMS_120823.1258
Linux, x64, 64bit (optimized), Oracle 11g on Aug 23 2012 16:58:29
Copyright (C) 1995, 2012, Oracle and/or its affiliates. All rights reserved.
Starting at 2017-12-05 16:21:03
***********************************************************************
Operating System Version:
Linux
Version #1 SMP Thu Feb 23 03:04:39 UTC 2017, Release 3.10.0-514.6.2.el7.x86_64
Node: zwjfdb3
Machine: x86_64
soft limit hard limit
Address Space Size : unlimited unlimited
Heap Size : unlimited unlimited
File Size : unlimited unlimited
CPU Time : unlimited unlimited
Process id: 375
***********************************************************************
** Running with the following parameters **
***********************************************************************
把./dirdef/source.def文件拷贝到目标端的./dirdef目录下
二、目标端(kafka)
#java必须为1.8
[root@bigdata01 ~]$ java -version
java version "1.8.0_45"
Java(TM) SE Runtime Environment (build 1.8.0_45-b14)
Java HotSpot(TM) 64-Bit Server VM (build 25.45-b02, mixed mode)
[root@bigdata01 ~]$
#创建ogg管理用户，并更改ogg安装路径权限
[root@bigdata01 ~]$ groupadd ogg
[root@bigdata01 ~]$ useradd -g ogg ogg
[root@bigdata01 ~]$ chown -R ogg:ogg /opt/ogg
#上传并解压ggs_Adapters_Linux_x64.tar到相应目录
[ogg@bigdata01 ogg]$ pwd
/opt/ogg
[ogg@bigdata01 ogg]$ ll
total 582548
drwxr-xr-x 6 ogg ogg 4096 Jun 7 2016 AdapterExamples
-rw-r----- 1 ogg ogg 426 Oct 15 2010 bcpfmt.tpl
-rw-r----- 1 ogg ogg 1725 Oct 15 2010 bcrypt.txt
-rwxrwxr-x 1 ogg ogg 8557335 May 1 2016 cachefiledump
-rwxrwxr-x 1 ogg ogg 8730645 May 1 2016 checkprm
-rwxr-x--- 1 ogg ogg 9567306 May 1 2016 convchk
-rwxrwxr-x 1 ogg ogg 15019428 May 1 2016 convprm
-rw-r----- 1 ogg ogg 159 Oct 15 2010 db2cntl.tpl
drwxr-x--- 2 ogg ogg 4096 Dec 5 14:05 dirchk
drwxr-x--- 2 ogg ogg 4096 Dec 5 14:05 dircrd
#配置环境变量
[ogg@bigdata01 ~]$ cat ~/.bash_profile
# .bash_profile
# Get the aliases and functions
if [ -f ~/.bashrc ]; then
. ~/.bashrc
fi
# User specific environment and startup programs
export JAVA_HOME=/opt/java8
export OGG_HOME=/opt/ogg
export LD_LIBRARY_PATH=/opt/java8/jre/lib/amd64/libjsig.so:/opt/java8/jre/lib/amd64/server/libjvm.so:/opt/java8/jre/lib/amd64/server:/opt/java8/jre/lib/amd64
PATH=$PATH:$HOME/bin:$JAVA_HOME:$JAVA_HOME/bin:$OGG_HOME
export PATH
[ogg@bigdata01 ~]$
#登录并用命令创建ogg所需文件夹
[ogg@bigdata01 ogg]$ ./ggsci
Oracle GoldenGate Command Interpreter
Version 12.2.0.1.160419 OGGCORE_12.2.0.1.0OGGBP_PLATFORMS_160430.1401
Linux, x64, 64bit (optimized), Generic on Apr 30 2016 16:21:34
Operating system character set identified as UTF-8.
Copyright (C) 1995, 2016, Oracle and/or its affiliates. All rights reserved.
GGSCI (bigdata01) 1> info all
Program Status Group Lag at Chkpt Time Since Chkpt
MANAGER STOPPED
GGSCI (bigdata01) 2> create subdirs
Creating subdirectories under current directory /opt/ogg
Parameter files /opt/ogg/dirprm: created
Report files /opt/ogg/dirrpt: created
Checkpoint files /opt/ogg/dirchk: created
Process status files /opt/ogg/dirpcs: created
SQL script files /opt/ogg/dirsql: created
Database definitions files /opt/ogg/dirdef: created
Extract data files /opt/ogg/dirdat: created
Temporary files /opt/ogg/dirtmp: created
Credential store files /opt/ogg/dircrd: created
Masterkey wallet files /opt/ogg/dirwlt: created
Dump files /opt/ogg/dirdmp: created
#配置mgr
GGSCI (bigdata01) 3> view param MGR
PORT 7809
DYNAMICPORTLIST 7810-7909
--定期清理dirdat路径下的本地队列,保留期限10天，过期后自动删除。
PURGEOLDEXTRACTS ./dirdat/*, USECHECKPOINTS, MINKEEPDAYS 7
LAGREPORTHOURS 1
LAGINFOMINUTES 30
LAGCRITICALMINUTES 45
#开启mgr
GGSCI (bigdata01) 4> start mgr
Manager started.
GGSCI (bigdata01) 5> info all
Program Status Group Lag at Chkpt Time Since Chkpt
MANAGER RUNNING
#配置kafka
[ogg@bigdata01 big-data]$ cp $OGG_HOME/AdapterExamples/big-data/kafka/* $OGG_HOME/dirprm/
[ogg@bigdata01 big-data]$ cd $OGG_HOME/dirprm/
[ogg@bigdata01 big-data]$ vi kafka.props
[ogg@bigdata01 big-data]$ cat kafka.props
gg.handlerlist = kafkahandler
gg.handler.kafkahandler.type = kafka
gg.handler.kafkahandler.KafkaProducerConfigFile=custom_kafka_producer.properties
gg.handler.kafkahandler.TopicName =ogg_zwjfborrower
gg.handler.kafkahandler.format =json
gg.handler.kafkahandler.SchemaTopicName=mySchemaTopic
gg.handler.kafkahandler.BlockingSend =false
gg.handler.kafkahandler.includeTokens=false
gg.handler.kafkahandler.mode =op
#gg.handler.kafkahandler.maxGroupSize =100, 1Mb
#gg.handler.kafkahandler.minGroupSize =50, 500Kb
goldengate.userexit.timestamp=utc
goldengate.userexit.writers=javawriter
javawriter.stats.display=TRUE
javawriter.stats.full=TRUE
gg.log=log4j
gg.log.level=INFO
gg.report.time=30sec
#Sample gg.classpath for Apache Kafka
gg.classpath=dirprm/:/data/cloudera/parcels/KAFKA-2.2.0-1.2.2.0.p0.68/lib/kafka/libs/*
#Sample gg.classpath for HDP
#gg.classpath=/etc/kafka/conf:/usr/hdp/current/kafka-broker/libs/*
javawriter.bootoptions=-Xmx512m -Xms32m -Djava.class.path=ggjava/ggjava.jar
#配置custom_kafka_producer.properties
[ogg@bigdata01 dirprm]$ cat custom_kafka_producer.properties
bootstrap.servers=bigdata01:9092,bigdata02:9092,bigdata03: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
# 100KB per partition
batch.size=102400
linger.ms=10000
#定义replication
GGSCI (bigdata01) 2> view param RZWJFBOR
REPLICAT RZWJFBOR
Setenv (NLS_LANG="AMERICAN_AMERICA.AL32UTF8")
TARGETDB LIBFILE libggjava.so SET property=dirprm/kafka.props
SOURCEDEFS dirdef/source.def
REPORTCOUNT EVERY 1 MINUTES, RATE
GROUPTRANSOPS 10000
MAP ZWJFBORROWER.*, TARGET ZWJFBORROWER.*;
#指定Trail文件
GGSCI (bigdata01) 2> add replicat RZWJFBOR, exttrail ./dirdat/zb
#启动replicat进程
GGSCI (bigdata01) 2>start replicat RZWJFBOR
GGSCI (bigdata01) 3> info all
Program Status Group Lag at Chkpt Time Since Chkpt
MANAGER RUNNING
REPLICAT RUNNING RZWJFBOR 00:00:00 00:00:08
三、测试数据是否到kafka
[root@bigdata01 ~]# /data/cloudera/parcels/KAFKA-2.2.0-1.2.2.0.p0.68/bin/kafka-console-consumer --zookeeper bigdata01:2181 --topic ogg_zwjfborrower --from-beginning
网友评论