oracle到kafka的同步

偏执的太偏执、 2022-09-23 08:41 237阅读 0赞

# GoldenGate 12.2 新版支持同步如下图 #

![Center][]

安装kafka

安装zookeeper

配置环境变量，/etc/profile添加以下内容：

\[ [root@T2][root_T2] kafkaogg\]\# export ZOOKEEPER\_HOME=/opt/app/kafka/zookeeper-3.4.6

\[ [root@T2][root_T2] kafkaogg\]\# export PATH=$PATH:$ZOOKEEPER\_HOME/bin

修改配置文件

\[ [root@T2][root_T2] kafkaogg\]\# cd zookeeper-3.4.6/conf/

\[ [root@T2][root_T2] conf\]\# ls

configuration.xsl  log4j.properties  zoo\_sample.cfg

\[ [root@T2][root_T2] conf\]\# cat zoo\_sample.cfg | grep -v '\#' > zoo.cfg

\[ [root@T2][root_T2] conf\]\# vi zoo.cfg

tickTime=2000

initLimit=10

syncLimit=5

dataDir=/tmp/zookeeper

clientPort=2181

接着启动zookeeper服务

\[ [root@T2][root_T2] conf\]\# ../bin/zkServer.sh start

JMX enabled by default

Using config: /opt/app/kafkaogg/zookeeper-3.4.6/bin/../conf/zoo.cfg

Starting zookeeper ... STARTED

查看是否启动

\[ [root@T2][root_T2] conf\]\# ../bin/zkServer.sh status

JMX enabled by default

Using config: /opt/app/kafkaogg/zookeeper-3.4.6/bin/../conf/zoo.cfg

Mode: standalone

Server启动之后, 就可以启动client连接server了, 执行脚本:

\[ [root@T2][root_T2] bin\]\#./zkCli.sh -server localhost:2181

安装kafka

安装kafka server之前需要单独安装zookeeper server，而且需要修改config/server.properties里面的IP信息

\[ [root@T2][root_T2] config\]\# pwd

/opt/app/kafkaogg/kafka\_2.11-0.9.0.0/config

\[ [root@T2][root_T2]  config\]\# vi server.properties

zookeeper.connect=localhost:2181

这里需要修改默认zookeeper.properties配置

\[ [root@T2][root_T2] config\]\# vi zookeeper.properties

dataDir=/tmp/zkdata

先启动zookeeper，启动前先kill掉之前的zkServer.sh启动的zookeeper服务

\[ [root@T2][root_T2]  bin\]\# ./zookeeper-server-start.sh ../config/zookeeper.properties

启动kafka服务

\[ [root@T2][root_T2]  bin\]\# ./kafka-server-start.sh ../config/server.properties

查看kafka进程是否启动：

\[ [root@T2][root_T2]  bin\]\# jps

16882 QuorumPeerMain

17094 Kafka

17287 Jps

创建topic

\[ [root@T2][root_T2] bin\]\#./kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test

Created topic "test".

查看topic列表

\[ [root@T2][root_T2] bin\]\#./kafka-topics.sh --list --zookeeper localhost:2181

test

创建broker集群

\[ [root@T2][root_T2] config\]\#cp server.properties ./server-1.properties

\[ [root@T2][root_T2] config\]\#cp server.properties ./server-2.properties

更改以下内容

\[ [root@T2][root_T2] config\]\#vi server-1.properties

broker.id=1

listeners=PLAINTEXT://:9093

log.dir=/tmp/kafka-logs-1

\[ [root@T2][root_T2] config\]\#vi server-2.properties

broker.id=2

listeners=PLAINTEXT://:9094

log.dir=/tmp/kafka-logs-2

其中broker.id是每一个broker的唯一标识号

启动另外两个Broker进程：

\[ [root@T2][root_T2] config\]\#../bin/kafka-server-start.sh ./server-1.properties &

\[ [root@T2][root_T2] config\]\#../bin/kafka-server-start.sh ./server-2.properties &

在同一台机器上创建三个broker的topic：

\[ [root@T2][root_T2] bin\]\#./kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 3 --partitions 1 --topic fafacluster

Created topic "fafacluster".

查看三个broker中，leader和replica角色：

\[ [root@T2][root_T2] bin\]\#./kafka-topics.sh --describe --zookeeper localhost:2181 --topic fafacluster

Topic:fafacluster    PartitionCount:1    ReplicationFactor:3    Configs:

Topic: fafacluster    Partition: 0    Leader: 2    Replicas: 2,0,1    Isr: 2,0,1

对于三个broker的伪cluster，可以尝试杀掉其中一个或两个broker进程，然后看看leader，replica，isr会不会发生变化，消费者还能不能正常消费消息。

生产者推送消息：

\[ [root@T2][root_T2] bin\]\#./kafka-console-producer.sh --broker-list localhost:9092 --topic fafacluster

fafa

fafa01

杀掉一个broker进程

\[ [root@T2][root_T2] bin\]\#kill -9 22497

\[ [root@T2][root_T2] bin\]\#./kafka-console-consumer.sh --zookeeper localhost:2181 --from-beginning --topic fafacluster

..................................

java.nio.channels.ClosedChannelException

at kafka.network.BlockingChannel.send(BlockingChannel.scala:110)

at kafka.consumer.SimpleConsumer.liftedTree1$1(SimpleConsumer.scala:98)

at kafka.consumer.SimpleConsumer.kafka$consumer$SimpleConsumer$$sendRequest(SimpleConsumer.scala:83)

at kafka.consumer.SimpleConsumer.getOffsetsBefore(SimpleConsumer.scala:149)

at kafka.consumer.SimpleConsumer.earliestOrLatestOffset(SimpleConsumer.scala:188)

at kafka.consumer.ConsumerFetcherThread.handleOffsetOutOfRange(ConsumerFetcherThread.scala:84)

at kafka.server.AbstractFetcherThread$$anonfun$addPartitions$2.apply(AbstractFetcherThread.scala:187)

at kafka.server.AbstractFetcherThread$$anonfun$addPartitions$2.apply(AbstractFetcherThread.scala:182)

at scala.collection.TraversableLike$WithFilter$$anonfun$foreach$1.apply(TraversableLike.scala:778)

at scala.collection.immutable.Map$Map1.foreach(Map.scala:116)

at scala.collection.TraversableLike$WithFilter.foreach(TraversableLike.scala:777)

at kafka.server.AbstractFetcherThread.addPartitions(AbstractFetcherThread.scala:182)

at kafka.server.AbstractFetcherManager$$anonfun$addFetcherForPartitions$2.apply(AbstractFetcherManager.scala:88)

at kafka.server.AbstractFetcherManager$$anonfun$addFetcherForPartitions$2.apply(AbstractFetcherManager.scala:78)

at scala.collection.TraversableLike$WithFilter$$anonfun$foreach$1.apply(TraversableLike.scala:778)

at scala.collection.immutable.Map$Map1.foreach(Map.scala:116)

at scala.collection.TraversableLike$WithFilter.foreach(TraversableLike.scala:777)

at kafka.server.AbstractFetcherManager.addFetcherForPartitions(AbstractFetcherManager.scala:78)

at kafka.consumer.ConsumerFetcherManager$LeaderFinderThread.doWork(ConsumerFetcherManager.scala:95)

at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:63)

fafa

fafa01

可以看到一大堆报错之后，还是完整的消费到生产的消息

使用kafka连接工具导入导出信息

\[ [root@T2][root_T2]  kafka\_2.11-0.9.0.0\]\# cat test.txt

foo

bar

dadadada

fgdgeg

sfewfwef

sfd

\[ [root@T2][root_T2]  kafka\_2.11-0.9.0.0\]\# cat config/connect-standalone.properties

bootstrap.servers=localhost:9092

key.converter=org.apache.kafka.connect.json.JsonConverter

value.converter=org.apache.kafka.connect.json.JsonConverter

key.converter.schemas.enable=true

value.converter.schemas.enable=true

internal.key.converter=org.apache.kafka.connect.json.JsonConverter

internal.value.converter=org.apache.kafka.connect.json.JsonConverter

internal.key.converter.schemas.enable=false

internal.value.converter.schemas.enable=false

offset.storage.file.filename=/tmp/connect.offsets

offset.flush.interval.ms=10000

\[ [root@T2][root_T2]  kafka\_2.11-0.9.0.0\]\# cat config/connect-file-source.properties

name=local-file-source

connector.class=org.apache.kafka.connect.file.FileStreamSourceConnector

tasks.max=1

file=test.txt

topic=fafacluster

\[ [root@T2][root_T2]  kafka\_2.11-0.9.0.0\]\# cat config/connect-file-sink.properties

name=local-file-sink

connector.class=org.apache.kafka.connect.file.FileStreamSinkConnector

tasks.max=1

file=test.sink.txt

topics=fafacluster

\[ [root@T2][root_T2]  kafka\_2.11-0.9.0.0\]\# bin/connect-standalone.sh  config/connect-standalone.properties  config/connect-file-source.properties config/connect-file-sink.properties

此时在接受端收到消息

\[ [root@T2][root_T2]  bin\]\# ./kafka-console-consumer.sh --zookeeper localhost:2181 --from-beginning --topic fafacluster

\{"schema":\{"type":"string","optional":false\},"payload":"foo"\}

\{"schema":\{"type":"string","optional":false\},"payload":"bar"\}

\{"schema":\{"type":"string","optional":false\},"payload":"dadadada"\}

\{"schema":\{"type":"string","optional":false\},"payload":"fgdgeg"\}

\{"schema":\{"type":"string","optional":false\},"payload":"sfewfwef"\}

\{"schema":\{"type":"string","optional":false\},"payload":"sfd"\}

原理解析：

1，kafka连接进程启动以后，源端连接开始读test.txt数据

2，把消息推送给topic fafacluster

3，sink connector连接开始读topic为fafacluster的消息，并把它写入文件名为test.sink.txt的文件

继续插入文件

\[ [root@T2][root_T2]  kafka\_2.11-0.9.0.0\]\# echo "Another line" >> test.txt

接收端查看发现，新增加的信息到达接收端

\{"schema":\{"type":"string","optional":false\},"payload":"Another line"\}

测试结束，现在我们创建一个同步数据topic，这里先确保zookeeper和kafka进程在运行

\[ [root@T2][root_T2] bin\]\#./kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic fafatable

\[ [root@T2][root_T2] bin\]\# ./kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic fafatable

SLF4J: Class path contains multiple SLF4J bindings.

SLF4J: Found binding in \[jar:file:/opt/app/Kafka/kafka\_2.11-0.9.0.0/libs/slf4j-log4j12-1.7.6.jar!/org/slf4j/impl/StaticLoggerBinder.class\]

SLF4J: Found binding in \[jar:file:/opt/app/kafkaogg/kafka\_2.11-0.9.0.0/libs/slf4j-log4j12-1.7.6.jar!/org/slf4j/impl/StaticLoggerBinder.class\]

SLF4J: See  [http://www.slf4j.org/codes.html\#multiple\_bindings][http_www.slf4j.org_codes.html_multiple_bindings] for an explanation.

SLF4J: Actual binding is of type \[org.slf4j.impl.Log4jLoggerFactory\]

Created topic "fafatable".

注意这里是kafka扫描到其他路径有SLF4J而报错

\[ [root@T2][root_T2] app\]\# mv Kafka Kafka.bak

这样既可解决

现在正式开始ogg oracle到kafka的同步

一．源端Oracle数据库安装ogg client

（过程略）不过需要在源端生产DEF文件

二．目标端Kafka 安装ogg for bigdata client

只有ogg for bigdata 12.2版本才开始支持kafka。

ogg for bigdata client所需的jar包在libs文件夹下都有的

\[ [root@T2][root_T2] kafka\_2.11-0.9.0.0\]\# cd libs/

\[ [root@T2][root_T2] libs\]\# ls

aopalliance-repackaged-2.4.0-b31.jar       javax.annotation-api-1.2.jar              jetty-servlet-9.2.12.v20150709.jar   log4j-1.2.17.jar

argparse4j-0.5.0.jar                       javax.inject-1.jar                        jetty-util-9.2.12.v20150709.jar      lz4-1.2.0.jar

connect-api-0.9.0.0.jar                    javax.inject-2.4.0-b31.jar                jopt-simple-3.2.jar                  metrics-core-2.2.0.jar

connect-file-0.9.0.0.jar                   javax.servlet-api-3.1.0.jar               kafka\_2.11-0.9.0.0.jar               osgi-resource-locator-1.0.1.jar

connect-json-0.9.0.0.jar                   javax.ws.rs-api-2.0.1.jar                 kafka\_2.11-0.9.0.0.jar.asc           scala-library-2.11.7.jar

connect-runtime-0.9.0.0.jar                jersey-client-2.22.1.jar                  kafka\_2.11-0.9.0.0-javadoc.jar       scala-parser-combinators\_2.11-1.0.4.jar

hk2-api-2.4.0-b31.jar                      jersey-common-2.22.1.jar                  kafka\_2.11-0.9.0.0-javadoc.jar.asc   scala-xml\_2.11-1.0.4.jar

hk2-locator-2.4.0-b31.jar                  jersey-container-servlet-2.22.1.jar       kafka\_2.11-0.9.0.0-scaladoc.jar      slf4j-api-1.7.6.jar

hk2-utils-2.4.0-b31.jar                    jersey-container-servlet-core-2.22.1.jar  kafka\_2.11-0.9.0.0-scaladoc.jar.asc  slf4j-log4j12-1.7.6.jar

jackson-annotations-2.5.0.jar              jersey-guava-2.22.1.jar                   kafka\_2.11-0.9.0.0-sources.jar       snappy-java-1.1.1.7.jar

jackson-core-2.5.4.jar                     jersey-media-jaxb-2.22.1.jar              kafka\_2.11-0.9.0.0-sources.jar.asc   validation-api-1.1.0.Final.jar

jackson-databind-2.5.4.jar                 jersey-server-2.22.1.jar                  kafka\_2.11-0.9.0.0-test.jar          zkclient-0.7.jar

jackson-jaxrs-base-2.5.4.jar               jetty-http-9.2.12.v20150709.jar           kafka\_2.11-0.9.0.0-test.jar.asc      zookeeper-3.4.6.jar

jackson-jaxrs-json-provider-2.5.4.jar      jetty-io-9.2.12.v20150709.jar             kafka-clients-0.9.0.0.jar

jackson-module-jaxb-annotations-2.5.4.jar  jetty-security-9.2.12.v20150709.jar       kafka-log4j-appender-0.9.0.0.jar

javassist-3.18.1-GA.jar                    jetty-server-9.2.12.v20150709.jar         kafka-tools-0.9.0.0.jar

前期配置工作

在目标端kafka的ogg配置，这里安装OGG\_v12.2.0.1\_bigdata\_Linux\_x64.zip

\[ [root@T2][root_T2] kafkaogg\]\# ls

AdapterExamples  dircrd  dirtmp       ggMessage.dat               lib               libggperf.so        libxml2.txt                               prvtclkm.plb      usrdecs.h

bcpfmt.tpl       dirdat  dirwlt       ggparam.dat                 libantlr3c.so     libggrepo.so        licenses                                  replicat          zlib.txt

bcrypt.txt       dirdef  dirwww       ggs\_Adapters\_Linux\_x64.tar  libdb-6.1.so      libicudata.so.48    logdump                                   retrace

cachefiledump    dirdmp  emsclnt      ggsci                       libggjava.so      libicudata.so.48.1  mgr                                       reverse

checkprm         dirout  extract      ggserr.log                  libggjava\_ue.so   libicui18n.so.48    notices.txt                               server

convchk          dirpcs  freeBSD.txt  help.txt                    libggjava\_vam.so  libicui18n.so.48.1  OGG\_BigData\_12.2.0.1.0\_Release\_Notes.pdf  sqlldr.tpl

convprm          dirprm  gendef       kafka\_2.11-0.9.0.0          libgglog.so       libicuuc.so.48      OGG\_BigData\_12.2.0.1\_README.txt           tcperrs

db2cntl.tpl      dirrpt  ggcmd        kafka\_2.11-0.9.0.0.tgz      libggnnzitp.so    libicuuc.so.48.1    oggerr                                    ucharset.h

dirchk           dirsql  ggjava       keygen                      libggparam.so     libxerces-c.so.28   OGG\_v12.2.0.1\_bigdata\_Linux\_x64.zip       UserExitExamples

这里使用的是oracle用户

cd $OGG\_HOME(在.bash\_profile中添加$OGG\_HOME环境变量)即/opt/app/kafkaogg

\[ [root@T2][root_T2] kafkaogg\]\# ./ggsci

Oracle GoldenGate Command Interpreter

Version 12.2.0.1.0 OGGCORE\_12.2.0.1.0\_PLATFORMS\_151101.1925.2

Linux, x64, 64bit (optimized), Generic on Nov 10 2015 16:18:12

Operating system character set identified as UTF-8.

GGSCI (T2) 1> CREATE SUBDIRS

Creating subdirectories under current directory /opt/app/kafkaogg

Parameter files                /opt/app/kafkaogg/dirprm: created

Report files                   /opt/app/kafkaogg/dirrpt: created

Checkpoint files               /opt/app/kafkaogg/dirchk: created

Process status files           /opt/app/kafkaogg/dirpcs: created

SQL script files               /opt/app/kafkaogg/dirsql: created

Database definitions files     /opt/app/kafkaogg/dirdef: created

Extract data files             /opt/app/kafkaogg/dirdat: created

Temporary files                /opt/app/kafkaogg/dirtmp: created

Credential store files         /opt/app/kafkaogg/dircrd: created

Masterkey wallet files         /opt/app/kafkaogg/dirwlt: created

Dump files                     /opt/app/kafkaogg/dirdmp: created

GGSCI (T2) 7> edit params mgr

port 1357

GGSCI (T2) 9> start mgr

Manager started.

拷贝kafka adapter 配置文件：

\[ [root@T2][root_T2] kafka\]\# pwd

/opt/app/kafkaogg/AdapterExamples/big-data/kafka

\[ [root@T2][root_T2] kafka\]\# cp \* /opt/app/kafkaogg/dirprm/

\[ [root@T2][root_T2] kafka\]\# vi /opt/app/kafkaogg/dirprm/custom\_kafka\_producer.properties

\#bootstrap.servers=host:port

bootstrap.servers=172.16.57.55: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

解释：

这里测试发现compression.type=gzip，replicate识别不了报错。尚不清楚原因

bootstrap是 HOSTNAME

\[ [root@T2][root_T2] kafka\]\# vi /opt/app/kafkaogg/dirprm/kafka.props

gg.handlerlist = kafkahandler

gg.handler.kafkahandler.type = kafka

gg.handler.kafkahandler.KafkaProducerConfigFile=custom\_kafka\_producer.properties

gg.handler.kafkahandler.TopicName = fafatable

gg.handler.kafkahandler.format =avro\_op

gg.handler.kafkahandler.SchemaTopicName= fafaschema

gg.handler.kafkahandler.BlockingSend = true

gg.handler.kafkahandler.includeTokens=false

\#gg.handler.kafkahandler.topicPartitioning=table

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

gg.classpath=dirprm/: /opt/app/kafkaogg/kafka\_2.11-0.9.0.0/libs /\*:

javawriter.bootoptions=-Xmx512m -Xms32m -Djava.class.path=ggjava/ggjava.jar

解释：

gg.handler.kafkahandler.TopicName 对应kafka的topic，跟源端数据库的table相对应，但可以不同名，用map可以匹配。

gg.handler.kafkahandler.SchemaTopicName 当format为avro时，需要配置此参数，否则可以不配。

gg.handler.kafkahandler.BlockingSend 当值为true时，表示同步更新，下一个消息发送需要等到写入到目标topic中，且确认已经收到才发下一条消息。为false为异步更新，将一次性发给目标topic。

gg.handler.kafkahandler.topicPartitioning有两种参数值none | table，控制是否已发布到kafka的数据应按表分区。设置为表，不同表的数据被写入到不同的kafka主题。 设置为None，来自不同表的数据交织在同一话题。

gg.handler.kafkahandler.mode 当值为tx时，表示源端一次事务内的操作在kafka上作为一个record。

查看当前Java版本

\[ [oracle@n3][oracle_n3] kafka\]$ java -version

java version "1.8.0\_77"

Java(TM) SE Runtime Environment (build 1.8.0\_77-b03)

Java HotSpot(TM) 64-Bit Server VM (build 25.77-b03, mixed mode)

注意：需要是jdk-1.7版本，不然启动时会报如下超时错误，ogg不支持jdk-1.8版本

删除1.8版本 yum remove java-1.8.0\*

[root@n3][root_n3]:/opt/app/kafkaogg\# yum install java-1.7.0\*

[root@bd-qa-oracle-86][root_bd-qa-oracle-86]:~\#java -version

java version "1.7.0\_79"

OpenJDK Runtime Environment (rhel-2.5.5.4.el6-x86\_64 u79-b14)

OpenJDK 64-Bit Server VM (build 24.79-b02, mixed mode)

[root@n3][root_n3]:/opt/app/kafkaogg\# ls /usr/lib/jvm/java-1.7.0-openjdk-1.7.0.79.x86\_64/jre/

bin  lib

\[ [root@T2][root_T2]  dirprm\]\# vi ~/.bash\_profile

export JAVA\_HOME=/usr/lib/jvm/java-1.7.0-openjdk-1.7.0.79.x86\_64/jre/

export LD\_LIBRARY\_PATH=$LD\_LIBRARY\_PATH:/usr/lib/jvm/java-1.7.0-openjdk-1.7.0.101.x86\_64/jre/lib/amd64/libjsig.so:/usr/lib/jvm/java-1.7.0-openjdk-1.7.0.101.x86\_64/jre/lib/amd64/server/libjvm.so:/usr/lib/jvm/java-1.7.0-openjdk-1.7.0.101.x86\_64/jre/lib/amd64/server:/usr/lib/jvm/java-1.7.0-openjdk-1.7.0.101.x86\_64/jre/lib/amd64

生效

\[ [root@T2][root_T2] dirprm\]\# source  ~/.bash\_profile

[root@n3][root_n3]:/opt/app/kafkaogg\# export  LD\_LIBRARY\_PATH=/usr/lib/jvm/java-1.7.0-openjdk-1.7.0.79.x86\_64/jre/lib/amd64/libjsig.so:/usr/lib/jvm/java-1.7.0-openjdk-1.7.0.79.x86\_64/jre/lib/amd64/server/libjvm.so:/usr/lib/jvm/java-1.7.0-openjdk-1.7.0.79.x86\_64/jre/lib/amd64/server:/usr/lib/jvm/java-1.7.0-openjdk-1.7.0.79.x86\_64/jre/lib/amd64/

源端oracle上搭建ogg，这里安装fbo\_ggs\_Linux\_x64\_shiphome.zip

1、开启主库归档日志、补充日志及force logging

SQL> alter database archivelog;

alter database archivelog

ERROR at line 1:

ORA-01126: database must be mounted in this instance and not open in any

instance

SQL> archive log list

Database log mode           Archive Mode

Automatic archival           Enabled

Archive destination           /opt/app/oracle/oraarch

Oldest online log sequence     91

Next log sequence to archive   93

Current log sequence           93

SQL> alter database add supplemental log data;

Database altered.

SQL> alter database force logging;

Database altered.

SQL> alter system set enable\_goldengate\_replication=true scope=both;

System altered.

2、关闭回收站

SQL> alter system set recyclebin=off scope=spfile;

System altered.

3、创建OGG管理用户（主备库都要设置）

create user ogg identified by ogg account unlock;

grant connect,resource to ogg;

grant select any dictionary to ogg;

grant select any table to ogg;

grant execute on utl\_file to ogg;

grant restricted session to ogg;

GRANT CREATE TABLE,CREATE SEQUENCE TO OGG;  (必须有的操作，后续会介绍)

grant dba to ogg;(可选)

数据初始化（Oracle initial load）

此时我们需要把dataprod用户下所有表导入到kafka中

查看

SQL> SELECT table\_name FROM all\_tables WHERE owner = upper('dataprod');

TABLE\_NAME

\------------------------------------------------------------

CRM\_AGG\_USERBEHAVIOR\_CALC1

DP\_TRADE\_PVCOUNT

CRM\_AGG\_USERBEHAVIOR\_CALC1\_BK

DP\_AGG\_USERBEHAVIORSCORE

CRM\_AGG\_USERPUSH\_CALC1

CRM\_AGG

6 rows selected.

SQL>  select owner,trigger\_name from all\_triggers where owner in ('DATAPROD');

OWNER

\------------------------------------------------------------

TRIGGER\_NAME

\------------------------------------------------------------

DATAPROD

EID\_ID

了解完dataprod用户表以后，我们开始初始化

安装源端ogg软件

[oracle@bd-qa-oracle-86][oracle_bd-qa-oracle-86]:/opt/app/kafkaogg$ll

total 464752

drwxr-xr-x 3 oracle oinstall      4096 Dec 12  2015 fbo\_ggs\_Linux\_x64\_shiphome

\-rw-r--r-- 1 oracle oinstall 475611228 Jul 13 17:54 fbo\_ggs\_Linux\_x64\_shiphome.zip

\-rw-r--r-- 1 oracle oinstall    282294 Jan 19 07:13 OGG-12.2.0.1.1-ReleaseNotes.pdf

\-rw-r--r-- 1 oracle oinstall      1559 Jan 19 07:12 OGG-12.2.0.1-README.txt

[oracle@bd-qa-oracle-86][oracle_bd-qa-oracle-86]:/opt/app/kafkaogg/fbo\_ggs\_Linux\_x64\_shiphome/Disk1$ls

install  response  runInstaller  stage

这里采用静默安装

[oracle@bd-qa-oracle-86][oracle_bd-qa-oracle-86]:/opt/app/kafkaogg/fbo\_ggs\_Linux\_x64\_shiphome/Disk1/response$ls

oggcore.rsp

[oracle@bd-qa-oracle-86][oracle_bd-qa-oracle-86]:/opt/app/kafkaogg/fbo\_ggs\_Linux\_x64\_shiphome/Disk1/response$vi oggcore.rsp

oracle.install.responseFileVersion=/oracle/install/rspfmt\_ogginstall\_response\_schema\_v12\_1\_2

INSTALL\_OPTION=ORA11g

START\_MANAGER=true

MANAGER\_PORT=1357

DATABASE\_LOCATION=/opt/app/oracle/product/11g

INVENTORY\_LOCATION=/opt/app/oraInventory

UNIX\_GROUP\_NAME=oinstall

解释：

DATABASE\_LOCATION是指 ORACLE\_HOME路径

[oracle@bd-qa-oracle-86][oracle_bd-qa-oracle-86]:/opt/app/kafkaogg/fbo\_ggs\_Linux\_x64\_shiphome/Disk1$./runInstaller -silent -responseFile /opt/app/kafkaogg/fbo\_ggs\_Linux\_x64\_shiphome/Disk1/response/oggcore.rsp Starting Oracle Universal Installer...

Checking Temp space: must be greater than 120 MB.   Actual 6427 MB    Passed

Checking swap space: must be greater than 150 MB.   Actual 7793 MB    Passed

Preparing to launch Oracle Universal Installer from /tmp/OraInstall2016-07-14\_09-54-11AM. Please wait  [...oracle@bd-qa-oracle-86][...oracle_bd-qa-oracle-86]:/opt/app/kafkaogg/fbo\_ggs\_Linux\_x64\_shiphome/Disk1$\[WARNING\] \[INS-75003\] The specified directory /opt/app/kafkaogg is not empty.

CAUSE: The directory specified /opt/app/kafkaogg contains files.

ACTION: Clean up the specified directory or enter a new directory location.

You can find the log of this install session at:

/opt/app/oraInventory/logs/installActions2016-07-14\_09-54-11AM.log

WARNING:OUI-10030:You have specified a non-empty directory to install this product. It is recommended to specify either an empty or a non-existent directory. You may, however, choose to ignore this message if the directory contains Operating System generated files or subdirectories like lost+found.

Do you want to proceed with installation in this Oracle Home?

The installation of Oracle GoldenGate Core was successful.

Please check '/opt/app/oraInventory/logs/silentInstall2016-07-14\_09-54-11AM.log' for more details.

Successfully Setup Software.

ogg软件安装成功

[oracle@bd-qa-oracle-86][oracle_bd-qa-oracle-86]:/opt/app/kafkaogg$./ggsci

Oracle GoldenGate Command Interpreter for Oracle

Version 12.2.0.1.1 OGGCORE\_12.2.0.1.0\_PLATFORMS\_151211.1401\_FBO

Linux, x64, 64bit (optimized), Oracle 11g on Dec 12 2015 00:54:38

Operating system character set identified as UTF-8.

GGSCI (bd-qa-oracle-86) 1> info all

Program     Status      Group       Lag at Chkpt  Time Since Chkpt

MANAGER     RUNNING

源端配置mgr进程

GGSCI (bd-qa-oracle-86) 10> edit params mgr

PORT 1357

dynamicportlist 9901-9920,9930

autostart er \*

autorestart er \*,retries 4,waitminutes 4

startupvalidationdelay 5

purgeoldextracts /opt/app/kafkaogg/dirdat/ff,usecheckpoints,minkeephours 2

配置数据同步用户

GGSCI (bd-qa-oracle-86) 7> dblogin userid ogg,password ogg

Successfully logged into database.

GGSCI (bd-qa-oracle-86 as  [ogg@BDDEV][ogg_BDDEV]) 8> add trandata dataprod.\*

2016-07-14 10:43:59  WARNING OGG-06439  No unique key is defined for table CRM\_AGG. All viable columns will be used to represent the key, but may not guarantee uniqueness. KEYCOLS may be used to define the key.

Logging of supplemental redo data enabled for table DATAPROD.CRM\_AGG.

TRANDATA for scheduling columns has been added on table 'DATAPROD.CRM\_AGG'.

TRANDATA for instantiation CSN has been added on table 'DATAPROD.CRM\_AGG'.

Logging of supplemental redo data enabled for table DATAPROD.CRM\_AGG\_USERBEHAVIOR\_CALC1.

TRANDATA for scheduling columns has been added on table 'DATAPROD.CRM\_AGG\_USERBEHAVIOR\_CALC1'.

TRANDATA for instantiation CSN has been added on table 'DATAPROD.CRM\_AGG\_USERBEHAVIOR\_CALC1'.

Logging of supplemental redo data enabled for table DATAPROD.CRM\_AGG\_USERBEHAVIOR\_CALC1\_BK.

TRANDATA for scheduling columns has been added on table 'DATAPROD.CRM\_AGG\_USERBEHAVIOR\_CALC1\_BK'.

TRANDATA for instantiation CSN has been added on table 'DATAPROD.CRM\_AGG\_USERBEHAVIOR\_CALC1\_BK'.

Logging of supplemental redo data enabled for table DATAPROD.CRM\_AGG\_USERPUSH\_CALC1.

TRANDATA for scheduling columns has been added on table 'DATAPROD.CRM\_AGG\_USERPUSH\_CALC1'.

TRANDATA for instantiation CSN has been added on table 'DATAPROD.CRM\_AGG\_USERPUSH\_CALC1'.

Logging of supplemental redo data enabled for table DATAPROD.DP\_AGG\_USERBEHAVIORSCORE.

TRANDATA for scheduling columns has been added on table 'DATAPROD.DP\_AGG\_USERBEHAVIORSCORE'.

TRANDATA for instantiation CSN has been added on table 'DATAPROD.DP\_AGG\_USERBEHAVIORSCORE'.

2016-07-14 10:44:00  WARNING OGG-06439  No unique key is defined for table DP\_TRADE\_PVCOUNT. All viable columns will be used to represent the key, but may not guarantee uniqueness. KEYCOLS may be used to define the key.

Logging of supplemental redo data enabled for table DATAPROD.DP\_TRADE\_PVCOUNT.

TRANDATA for scheduling columns has been added on table 'DATAPROD.DP\_TRADE\_PVCOUNT'.

TRANDATA for instantiation CSN has been added on table 'DATAPROD.DP\_TRADE\_PVCOUNT'.

源端初始化配置

A：配置extract进程，注意这里使用direct file模式，而不是load模式

GGSCI (bd-qa-oracle-86) 2> add extract fafainie, sourceistable

EXTRACT added.

GGSCI (bd-qa-oracle-86) 3> info extract \*,tasks

EXTRACT    FAFAINIE  Initialized   2016-07-14 11:03   Status STOPPED

Checkpoint Lag       Not Available

Log Read Checkpoint  Not Available

First Record         Record 0

Task                 SOURCEISTABLE

GGSCI (bd-qa-oracle-86) 8> edit params fafainie

extract fafainie

userid ogg,password ogg

rmthost 172.16.57.55,mgrport 1357

\--rmttask replicat,group fafainir

RMTFILE ./dirdat/ff

table dataprod.\*;

目标端初始化配置

配置mgr

GGSCI (T2) 2> edit params mgr

port 1357

ACCESSRULE, PROG REPLICAT, IPADDR 172.16.57.\*, ALLOW

dynamicportlist 9901-9920,9930

autostart er \*

autorestart er \*,retries 4,waitminutes 4

startupvalidationdelay 5

purgeoldextracts /opt/app/kafkaogg/dirdat/\*,usecheckpoints,minkeephours 2

LAGREPORTHOURS 1

LAGINFOMINUTES 30

LAGCRITICALMINUTES 45

A：配置replicat进程

GGSCI (T2) 5> add replicat fafainir, specialrun

REPLICAT added.

GGSCI (T2) 11> info replicat \*, TASKS

REPLICAT   FAFAINIR  Initialized   2016-07-14 11:05   Status STOPPED

Checkpoint Lag       00:00:00 (updated 00:00:07 ago)

Log Read Checkpoint  Not Available

Task                 SPECIALRUN

B:编辑replicat进程参数

注意这里修改这个文件

\[ [root@T2][root_T2] dirprm\]\# vi rkafka.prm

SPECIALRUN

end runtime

\-- REPLICAT fafainir

\-- Trail file for this example is located in "AdapterExamples/trail" directory

\-- Command to add REPLICAT

\-- add replicat rkafka, exttrail AdapterExamples/trail/ff

setenv (NLS\_LANG=AMERICAN\_AMERICA.ZHS16GBK)

TARGETDB LIBFILE libggjava.so SET property=dirprm/kafka.props

EXTFILE ./dirdat/ff

DDL INCLUDE ALL

REPORTCOUNT EVERY 1 MINUTES, RATE

GROUPTRANSOPS 10000

MAP dataprod.\*, TARGET fafaschema.\*;

\[ [root@T2][root_T2] dirprm\]\# mv rkafka.prm fafainir.prm

这里target 可以随便取名，但是这里可源库的gg.handler.kafkahandler.SchemaTopicName一致，只是做一个标识。kafka发出的消息所有消费者都可以消费了。

启动初始化进程

源端：

GGSCI (bd-qa-oracle-86) 5> start fafainie

Sending START request to MANAGER ...

EXTRACT FAFAINI starting

GGSCI (bd-qa-oracle-86) 6> view report fafainie

目标端：

\[ [root@T2][root_T2] kafkaogg\]\# replicat paramfile ./dirprm/fafainir.prm reportfile ./dirrpt/fafainir.rpt -p INITIALDATALOAD

\[ [root@T2][root_T2] kafkaogg\]\# tail -100f ggserr.log

验证传输情况，查看kafka消费情况

JFAFASCHEMA.CRM\_AGG\_USERBEHAVIOR\_CALC1I42016-07-14 07:24:09.02265842016-07-15T18:03:05.741000(00000000000000003583

USERID:2016-05-03:00:00:00.00000000020250006userA001탰?탰?퀄SHSHSHSH:2016-05-03:00:00:00.000000000:2016-05-03:00:00:00.000000000:2016-05-03:00:00:00.000000000:2016-05-03:00:00:00.0000000000000:2016-05-03:00:00:00.000000000000

JFAFASCHEMA.CRM\_AGG\_USERBEHAVIOR\_CALC1I42016-07-14 07:24:09.02265842016-07-15T18:03:25.776000(00000000000000004064

USERID:2016-05-03:00:00:00.00000000020250006userA002탰?탰?퀄SHSHSHSH:2016-05-03:00:00:00.000000000:2016-05-03:00:00:00.000000000:2016-05-03:00:00:00.000000000:2016-05-03:00:00:00.0000000000000:2016-05-03:00:00:00.000000000000

JFAFASCHEMA.CRM\_AGG\_USERBEHAVIOR\_CALC1I42016-07-14 07:24:09.02265842016-07-15T18:03:35.783000(00000000000000004545

数据成功传输

数据同步

1、配置DDL同步

A:目标库 配置globals参数

GGSCI (T2) 3> view param ./globals

ggschema ogg

B:源库 执行DDL配置脚本

sqlplus / as sysdba

SQL> @/opt/app/kafkaogg/marker\_setup.sql

Marker setup script

You will be prompted for the name of a schema for the Oracle GoldenGate database objects.

NOTE: The schema must be created prior to running this script.

NOTE: Stop all DDL replication before starting this installation.

Enter Oracle GoldenGate schema name:ogg

Marker setup table script complete, running verification script...

Please enter the name of a schema for the GoldenGate database objects:

Setting schema name to OGG

MARKER TABLE

\--------------------------------------------------------------

MARKER SEQUENCE

\--------------------------------------------------------------

Script complete.

输入OGG管理用户名：ogg

首先建立ogg独有的表空间

SQL> CREATE TABLESPACE TBS\_OGG DATAFILE '/opt/app/oracle/oradata/orcl11g/tbs\_ogg\_01.dbf' SIZE 2G AUTOEXTEND ON NEXT 50M MAXSIZE UNLIMITED;

Tablespace created.

SQL> alter user ogg DEFAULT TABLESPACE TBS\_OGG;

User altered.

SQL> grant connect,resource,unlimited tablespace to ogg;

Grant succeeded.

SQL>  @/opt/app/kafkaogg/ddl\_setup.sql

Oracle GoldenGate DDL Replication setup script

Verifying that current user has privileges to install DDL Replication...

You will be prompted for the name of a schema for the Oracle GoldenGate database objects.

NOTE: For an Oracle 10g source, the system recycle bin must be disabled. For Oracle 11g and later, it can be enabled.

NOTE: The schema must be created prior to running this script.

NOTE: Stop all DDL replication before starting this installation.

Enter Oracle GoldenGate schema name:ogg

Working, please wait ...

Spooling to file ddl\_setup\_spool.txt

Checking for sessions that are holding locks on Oracle Golden Gate metadata tables ...

Check complete.

Using OGG as a Oracle GoldenGate schema name.

Working, please wait ...

DDL replication setup script complete, running verification script...

Please enter the name of a schema for the GoldenGate database objects:

Setting schema name to OGG

CLEAR\_TRACE STATUS:

Line/pos

\--------------------------------------------------------------------------------

Error

\-----------------------------------------------------------------

No errors

....................

输入OGG管理用户名：ogg

注意1：此处可能会报错：ORA-04098: trigger 'SYS.GGS\_DDL\_TRIGGER\_BEFORE' is invalid and failed，同时OGG中的很多表和视图无法创建，原因主要由于OGG缺少权限引起，即便有

DBA权限也是不足的（OGG BUG），可以通过如下方法修复：

1）先将触发器关闭，否则执行任何sql都会包ORA-04098的错误

@/opt/app/OGG/ddl\_disable.sql

2)赋予ogg对应权限

grant execute on utl\_file to ogg;

grant restricted session to ogg;

GRANT CREATE TABLE,CREATE SEQUENCE TO OGG;

3)重新执行ddl\_setup.sql

注意2：当主库上有很多应用连接时，执行该sql会出现如下报警：

IMPORTANT: Oracle sessions that used or may use DDL must be disconnected. If you

continue, some of these sessions may cause DDL to fail with ORA-6508.

To proceed, enter yes. To stop installation, enter no.

Enter yes or no:

为了不影响主库，选no，选择一个时间点，停止应用再创建ddl。

如果不创建ddl，需要在主备库的ogg进程参数中添加truncate选项：

gettruncates，参考后面同步进程配置。

SQL> @/opt/app/kafkaogg/role\_setup.sql

GGS Role setup script

This script will drop and recreate the role GGS\_GGSUSER\_ROLE

To use a different role name, quit this script and then edit the params.sql script to change the gg\_role parameter to the preferred name. (Do not run the script.)

You will be prompted for the name of a schema for the GoldenGate database objects.

NOTE: The schema must be created prior to running this script.

NOTE: Stop all DDL replication before starting this installation.

Enter GoldenGate schema name:ogg

Wrote file role\_setup\_set.txt

PL/SQL procedure successfully completed.

Role setup script complete

Grant this role to each user assigned to the Extract, GGSCI, and Manager processes, by using the following SQL command:

GRANT GGS\_GGSUSER\_ROLE TO <loggedUser>

where <loggedUser> is the user assigned to the GoldenGate processes.

输入OGG管理用户名：ogg

SQL> GRANT GGS\_GGSUSER\_ROLE TO OGG;

Grant succeeded.

SQL> @/opt/app/kafkaogg/ddl\_enable.sql

Trigger altered.

2、配置数据同步

A:源库 配置日志抓取进程

GGSCI (bd-qa-oracle-86) 1> add extract fafae, tranlog, begin now

EXTRACT added.

GGSCI (bd-qa-oracle-86) 2> add rmttrail /opt/app/kafkaogg/dirdat/fa,extract fafae

RMTTRAIL added.

注意这里不能和初始化trail文件名称一致，初始化时ff，这里使用fa

GGSCI (bd-qa-oracle-86) 3> edit params fafae

extract fafae

userid ogg,password ogg

rmthost 172.16.57.55, mgrport 1357

rmttrail /opt/app/kafkaogg/dirdat/fa

discardfile /opt/app/kafkaogg/dirrpt/trail.dsc,append,megabytes 100

ddl include mapped

table dataprod.\*;

B:目标库 配置日志解析进程

1）编辑globals参数

GGSCI (T2) 4> edit params ./GLOBALS

ggschema ogg

checkpointtable ogg.chkpnt\_fafa

2）创建checkpoint表

这里在目标端由于kafka自己可以记录所以无需创建，其实也无法创建

3)配置解析进程

GGSCI (T2) 6> add replicat fafar,exttrail /opt/app/kafkaogg/dirdat/fa

REPLICAT added.

GGSCI (T2) 7> info replicat \*, TASKS

REPLICAT   FAFAINIR  Initialized   2016-07-14 16:18   Status STOPPED

Checkpoint Lag       00:00:00 (updated 22:31:52 ago)

Log Read Checkpoint  Not Available

Task                 SPECIALRUN

REPLICAT   REP\_INIT  Initialized   2016-07-15 09:02   Status STOPPED

Checkpoint Lag       00:00:00 (updated 05:48:32 ago)

Log Read Checkpoint  Not Available

Task                 SPECIALRUN

\[ [root@T2][root_T2] dirprm\]\# vi fafar.prm

REPLICAT fafar

\-- Trail file for this example is located in "AdapterExamples/trail" directory

\-- Command to add REPLICAT

\-- add replicat rkafka, exttrail AdapterExamples/trail/ff

setenv (NLS\_LANG=AMERICAN\_AMERICA.ZHS16GBK)

TARGETDB LIBFILE libggjava.so SET property=dirprm/kafka.props

DDL INCLUDE ALL

REPORTCOUNT EVERY 1 MINUTES, RATE

GROUPTRANSOPS 10000

MAP dataprod.\*, TARGET fafaschema.\*;

C:启动同步进程

这里总结一下：

1，初始化的时候用的是rmtfile，然后他的replicat进程也必须是rmtfile

2，然后同步进程源端是rmttrail，他的目标端必须是rmttrail，他们必须都保持一致

3，这里强调，初始化进程和同步进程的trail文件名称不能一致，不然会因为初始化trail文件和同步trail文件格式不同而报 with format RELEASE 12.2, does not match current format specification of RELEASE 9.0/9.5错误。

源库：

GGSCI (bd-qa-oracle-86) 5> start fafae

Sending START request to MANAGER ...

EXTRACT FAFAE starting

目标库：

GGSCI (T2) 2> start fafar

Sending START request to MANAGER ...

REPLICAT FAFAR starting

GGSCI (T2) 3> view report fafar

测试

源端：

SQL> conn dataprod/dataprod

Connected.

SQL> create table fafa(a int);

Table created.

SQL> insert into fafa values(1);

1 row created.

SQL> commit;

Commit complete.

目标端：

\[ [root@T2][root_T2] bin\]\# ./kafka-console-consumer.sh --zookeeper localhost:2181 --from-beginning --topic fafatable

FAFASCHEMA.FAFAI42016-07-15 09:37:38.20047342016-07-15T17:37:43.662000(00000000000000003128A1

FAFASCHEMA.FAFAI42016-07-15 09:38:51.19952442016-07-15T17:38:55.152000(00000000000000003242A2

同步进程成功

ogg--------------

[Center]: /images/20220720/0df1d7d9de2d45f3b0d8af3e3f4ecfce.png
[root_T2]: mailto:root@T2
[http_www.slf4j.org_codes.html_multiple_bindings]: http://www.slf4j.org/codes.html#multiple_bindings
[oracle_n3]: mailto:oracle@n3
[root_n3]: mailto:root@n3
[root_bd-qa-oracle-86]: mailto:root@bd-qa-oracle-86
[oracle_bd-qa-oracle-86]: mailto:oracle@bd-qa-oracle-86
[...oracle_bd-qa-oracle-86]: mailto:...oracle@bd-qa-oracle-86
[ogg_BDDEV]: mailto:ogg@BDDEV