ogg同步数据到kafka案例简介

##view param EX_KAF

EXTRACT ex_kaf

SETENV (NLS_LANG=AMERICAN_AMERICA.ZHS16GBK)

setenv (ORACLE_HOME=/u01/app/oracle/product/12.2.0.1/db_1)

SETENV (ORACLE_SID=db2)

USERID c##ggs@db, PASSWORD XXXX

EXTTRAIL ./dirdat/ha

--DISCARDFILE ./dirrpt/ex_kaf.DSC, APPEND,MEGABYTES 100

-- report info

REPORTCOUNT EVERY 10 MINUTES, RATE

WARNLONGTRANS 2h,CHECKINTERVAL 30m

dboptions allowunusedcolumn

fetchoptions nousesnapshot

LOGALLSUPCOLS  //加入前镜像（12c新版本特有参数，遇到josn格式不固定，新老参数同时加）

getupdatebefores  // 加入前镜像（老版本参数）

nocompressdeletes / /加入前镜像（老版本参数）

nocompressupdates // 加入前镜像（老版本参数）

--crm3hj 208 tables

TABLE crm3hj.CUST.TAB;

##view param DP_KAF

EXTRACT dp_kaf

PASSTHRU //传输进程透传，不做任何处理

rmthost 192.168.100.100, mgrport 7809 , TCPBUFSIZE 300000, TCPFLUSHBYTES 300000//目标端地址

rmttrail ./dirdat/hk

EOFDELAYCSECS 10

--gettruncates

--crm3hj 8 tables

TABLE crm3hj.CUST.TAB;

[oracle@server003 dirprm]$ cat rp_kaf.prm

REPLICAT rp_kaf //定义进程名称

sourcedefs ./dirdef/o2kaf_jt.def  //指定使用源和目标的表映射文件，高版本可省略

TARGETDB LIBFILE libggjava.so SET property=dirprm/kafka_crm_ha.props //定义kafka一些适配性的库文件以及配置文件，配置文件指定属性文件

--reperror default, abend

reperror default, discard  //错误处理，这里将错误信息记录DSC文件

DISCARDFILE ./dirrpt/rp_kaf.DSC, APPEND, MEGABYTES 4096 //对DSC文件的属性定义

REPORTCOUNT EVERY 10 MINUTES, RATE   //报告指定10min统计一次报告

--grouptransops 10000  //组提交，以事务传输时，事务合并的单位，减少IO操作

MAP CRM3HJ.CUST.TAB, TARGET CUST. TAB; //具体表的映射配置

[oracle@server003 dirprm]$ cat kafka_crm_ha.props

gg.handlerlist=kafkahandler //handler类型

gg.handler.kafkahandler.type=kafka

gg.handler.kafkahandler.topicMappingTemplate=tp_share_crm //指定kafka的topic

gg.handler.kafkahandler.KafkaProducerConfigFile=custom_kafka_producer_JT.properties //指定kafka生产者配置文件

gg.handler.kafkahandler.ProducerRecordClass=oracle.goldengate.handler.kafka.MyCreateProducerRecordHa//生产者方法

gg.handler.kafkahandler.BlockingSend=false

gg.handler.kafkahandler.includeTokens=false

gg.handler.kafkahandler.mode=op //OGG for BigData中传输模式，即op为一次SQL传输一次，tx为一次事务传输一次

gg.handler.kafkahandler.format=json #传输的消息最终解析的格式，格式相关

gg.handler.kafkahandler.format.includePrimaryKeys=true

gg.handler.kafkahandler.format.insertOpKey=I

gg.handler.kafkahandler.format.updateOpKey=U

gg.handler.kafkahandler.format.deleteOpKey=D

gg.handler.kafkahandler.authType=kerberos      //kerberos安全认证相关

gg.handler.kafkahandler.kerberosKeytabFile=/home/oracle/KDC/kafka_XXXX.keytab

gg.handler.kafkahandler.kerberosPrincipal=kafka_XXXX@HADOOP.XXXX.CN

goldengate.userexit.timestamp=utc+8

goldengate.userexit.writers=javawriter

javawriter.stats.display=true

javawriter.stats.full=true

javawriter.bootoptions=-Xmx512m -Xms32m -Djava.class.path=ggjava/ggjava.jar -Djava.security.krb5.conf=/etc/krb5.conf -Djava.security.auth.login.config=/home/oracle/kafka_jass.conf //JVM设置

gg.log=log4j  //日志记录

gg.log.level=INFO

gg.report.time=30sec

gg.classpath=dirprm/:/oggdata/kafka/kafka_2.11-1.0.2/libs/*:/oggdata/ggv191adp/:/oggdata/ggv191adp/lib/* #Kafka的lib目录

[oracle@server003 dirprm]$ cat custom_kafka_producer_JT.properties

bootstrap.servers=XXXX.COM //kafka集群的broker的地址

acks=1 //参考KAFKA的acks参数

compression.type=gzip  //压缩类型

reconnect.backoff.ms=1000 //重连延时

max.request.size=5024000 //请求发送设置

send.buffer.bytes=5024000

value.serializer=org.apache.kafka.common.serialization.ByteArraySerializer

key.serializer=org.apache.kafka.common.serialization.ByteArraySerializer

batch.size=102400

linger.ms=10000

security.protocol=SASL_PLAINTEXT

sasl.kerberos.service.name=kafka // kerberos安全认证相关

saal.machanism=GSSAPI