kafka handler

1.配置kafka 参数文件

在ogg主目录下有示例文件:

[root@WH0PRDBRP00AP0013 ogg]# cd AdapterExamples/big-data/kafka/
[root@WH0PRDBRP00AP0013 kafka]# ls
custom_kafka_producer.properties  kafka.props  rkafka.prm

custom_kafka_producer.properties :

bootstrap.servers=localhost:9092  //kafka服务器的地址
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

kafka.props:

gg.handlerlist = kafkahandler
gg.handler.kafkahandler.type = kafka
gg.handler.kafkahandler.KafkaProducerConfigFile=custom_kafka_producer.properties
gg.handler.kafkahandler.TopicName =oggtopic
gg.handler.kafkahandler.format =avro_op  #有多种模式可以选 xml, delimitedtext, json, avro_row, avro_op
gg.handler.kafkahandler.SchemaTopicName=mySchemaTopic
gg.handler.kafkahandler.BlockingSend =false  #阻塞模式
gg.handler.kafkahandler.includeTokens=false
gg.handler.kafkahandler.mode =tx  #可选OP/TX,OP每次ORACLE的操作(I D U)都会当作生产者消费者的记录刷新一次,TX按事务提交刷新
#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/cloudera/parcels/KAFKA/lib/kafka/libs/*:
javawriter.bootoptions=-Xmx512m -Xms32m -Djava.class.path=ggjava/ggjava.jar 

rkafka.prm:

REPLICAT rkafka
TARGETDB LIBFILE libggjava.so SET property=dirprm/kafka.props
REPORTCOUNT EVERY 1 MINUTES, RATE
GROUPTRANSOPS 10000
MAP XDGL.BUSINESS_CONTRACT , TARGET XDGL.BUSINESS_CONTRACT;

2.添加kafka复制进程

来自为知笔记(Wiz)