porter是如何集成OGG

[dingyunsheng]

请问前辈 porter是如何集成OGG的？具体可以在哪个模块里看到？谢谢！

[zhangkewei]

porter要同步Oracle需要两个步骤

部署Ogg同步数据到Kafka

部署文档参阅: https://docs.oracle.com/goldengate/bd1221/gg-bd/GADBD/GUID-F0FA2781-0802-4530-B1F0-5E102B982EC0.htm#GADBD500 https://docs.oracle.com/goldengate/bd1221/gg-bd/GADBD/GUID-A6C0DEC9-480F-4782-BD2A-54FEDDE2FDD9.htm#GADBD112

Ogg输出到kafka格式相关配置

gg.handlerlist = kafkahandler
gg.handler.kafkahandler.type=kafka
gg.handler.kafkahandler.KafkaProducerConfigFile=custom_kafka_producer.properties
gg.handler.kafkahandler.topicMappingTemplate=接收数据的kafka主题

gg.handler.kafkahandler.format =json
gg.handler.kafkahandler.format.insertOpKey=I
gg.handler.kafkahandler.format.updateOpKey=U
gg.handler.kafkahandler.format.deleteOpKey=D
gg.handler.kafkahandler.format.truncateOpKey=T
gg.handler.kafkahandler.format.prettyPrint=false
gg.handler.kafkahandler.format.jsonDelimiter=CDATA[]
gg.handler.kafkahandler.format.includePrimaryKeys=true
gg.handler.kafkahandler.SchemaTopicName= 接收表结构schema的kafka主题，可以根据这个实现表结构变更通知，目前没做实现
gg.handler.kafkahandler.BlockingSend =false
gg.handler.kafkahandler.includeTokens=false
gg.handler.kafkahandler.mode=op


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/:/kafka lib路径/*

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

配置Porter任务

以下配置文件格式适用配置管理后台"同步管理->高级任务配置(原菜单名:本地任务)->新增" 如果是本地任务配置文件需要增加前缀"porter.task[任务下标，从0开始]"

taskId=任务ID
consumer.consumerName=KafkaFetch
consumer.converter=oggJson
consumer.source.sourceType=KAFKA
consumer.source.servers=kafka集群地址
consumer.source.topics=接收数据的kafka主题
consumer.source.group=kafka消费者组

loader.loaderName=JdbcBatch
loader.source.sourceType=JDBC
loader.source.dbType=可选项:MYSQL、ORACLE
loader.source.url=jdbc:mysql://127.0.0.1:3306/数据库?useUnicode=true&characterEncoding=utf8
loader.source.userName=用户名
loader.source.password=密码
loader.source.maxWait=60000
loader.source.minPoolSize=10
loader.source.maxPoolSize=50
loader.source.initialPoolSize=20
loader.source.connectionErrorRetryAttempts=3
loader.insertOnUpdateError=false

mapper[0].schema=源端schema,目标端数据库schema(mysql就是数据库名)
mapper[0].table=源表名,目标端表名
mapper[0].column.源表字段名=目标端表字段名

mapper[0].forceMatched=false