使用Oracle GoldenGate 同步Oracle数据到KAFKA

叁歲伎倆 2022-12-14 13:52 137阅读 0赞

**OGG**  
oracle golden gate（OGG）是可以实时捕获oracle数据库的数据变动日志并将其同步到目标端的中间件。包括安装及配置过程。

**Oracle配置**  
oracle端需要开启归档日志

alter database force logging;
    alter database add SUPPLEMENTAL log data;

查看是否开启成功

SQL> select NAME,OPEN_MODE,FORCE_LOGGING,SUPPLEMENTAL_LOG_DATA_MIN from v$database;
    
    NAME   OPEN_MODE        FOR SUPPLEME
    --------- -------------------- --- --------
    ORA235   READ WRITE        YES YES

注意一点，如果你想在update时获取全部字段，还需要以下操作

SQL> alter database add supplemental log data(all) columns;
    
    Database altered.
    
    SQL> select supplemental_log_data_all from v$database;
    
    SUPPLE
    ------
    YES

**OGG安装**  
oracle官网上可以找到ogg的安装包，这里我匹配oracle数据库版本是11g，Linux为64bit,所以我下载的ogg为‘ggs\_Linux\_x64\_ora11g\_64bit\_v11\_1\_1\_0\_0\_078.tar’; 该ogg为源端ogg,可以extract and pump data to target。目标端的话比较特殊，因为我需要将源端ogg获取的实时数据发送至kafka，因为目标端是kafka(亦或其他大数据组件)，所以目标端我选择的ogg-bigdata（OGG\_BigData\_Linux\_x64\_19.1.0.0.1.tar）。

*  Linux创建ogg用户(源端也可以用existing user such as oracle)
 *  修改ogg用户的环境变量(.bash\_profile文件)
 *  创建ogg安装目录/data/ogg并将tar解压至该目录tar -xvf ggs\_Linux\_x64\_ora11g\_64bit\_v11\_1\_1\_0\_0\_078.tar -C /data/ogg
 *  oracle中创建ogg用户并将dba权限赋予ogg用户

create user ogg identified by ogg;
    grant dba to ogg;
    GRANT CREATE TABLE,CREATE SEQUENCE TO OGG;

**OGG配置**

*  ogg目录下进入oracle命令行模式，运行DDL，如下：

@marker_setup.sql
    @ddl_setup.sql
    @role_setup.sql
    @ddl_enable.sql

*  创建相关目录，./ggsci登录ogg，执行`create subdirs`
 *  定义表结构定义文件

edit params hs_asset_kafka
    
    
    defsfile /data/ogg/dirdef/kafka.def
    userid ogg,password ogg
    table HS_ASSET.fundaccount;
    table HS_ASSET.FUNDJOUR;

保存后，在ogg目录下执行`./defgen paramfile dirprm/hs_asset_kafka.def`，则在dirdef目录下生成配置文件中相关表的表结构文件，该文件同时也要发送到目标端dirdef目录下。

*  MGR进程配置  
    mgr进程负责启动ogg进程，ogg进程监听端口号，以及一些通用配置，如tail文件过期时间

源端

PORT 7810
    AUTORESTART EXTRACT *, WAITMINUTES 2, RESETMINUTES 5
    PURGEOLDEXTRACTS /data/ogg/dirdat/*, USECHECKPOINTS, MINKEEPDAYS 2

目标端  
这里注意要配置动态端口，应为源端到目标端的pump过程可能有多个，也可能选择不同的端口进行传输，避免端口不可用情况发生。

PORT 7809
    DYNAMICPORTLIST 8000-8030
    AUTORESTART EXTRACT *, WAITMINUTES 2, RESETMINUTES 5
    PURGEOLDEXTRACTS ./dirdat/*, USECHECKPOINTS, MINKEEPDAYS 2

*  源端EXTRACT进程配置  
    抽取进程，负责抽取指定table中的数据变动到指定的目录tail文件下

EXTRACT EORA_1
    SETENV (NLS_LANG=AMERICAN_AMERICA.ZHS16GBK)
    GETUPDATEBEFORES
    NOCOMPRESSDELETES
    NOCOMPRESSUPDATES
    USERID ogg, PASSWORD ogg
    EXTTRAIL /data/ogg/dirdat/HS_ASSET_FUNDACCOUNT/aa
    TABLE HS_ASSET.fundaccount;

*  源端PUMP进程配置  
    负责将trail文件传输到目标端，这里源端服务器和目标端服务器不是同一台服务器

EXTRACT PORA_1
    SETENV (NLS_LANG=AMERICAN_AMERICA.ZHS16GBK)
    PASSTHRU
    RMTHOST 10.10.48.85, MGRPORT 7809
    RMTTRAIL /data/ogg/dirdat/HS_ASSET_FUNDACCOUNT/pa
    TABLE HS_ASSET.fundaccount;

*  目标端REPLICAT进程配置  
    源端数据传输到目标端后（/data/ogg/dirdat/HS\_ASSET\_FUNDACCOUNT/pa），需要将数据进一步传输到其他贮存组件中，这里我是进一步上传到KAFKA

REPLICAT RORA_1
    sourcedefs /data/ogg/dirdef/kafka.def
    TARGETDB LIBFILE libggjava.so SET property=dirprm/fundaccount_kafka.props
    REPORTCOUNT EVERY 1 MINUTES, RATE
    GROUPTRANSOPS 100
    SOURCECHARSET ZHS16GBK
    REPLACEBADCHAR ESCAPE
    MAP HS_ASSET.FUNDACCOUNT, TARGET HS_ASSET.FUNDACCOUNT;

第二行是源端同步过来的表结构定义文件；  
第三行指定了一个和kafka相关的properties文件，下面根据这个文件展开，看是如何在目标端实现replicat进程将tail文件里的数据传输至kafka的。  
看一下fundaccount\_kafka.props文件里写了什么

gg.handlerlist = kafkahandler
    gg.handler.kafkahandler.type=kafka
    gg.handler.kafkahandler.KafkaProducerConfigFile=custom_kafka_producer.properties
    #The following resolves the topic name using the short table name
    gg.handler.kafkahandler.topicMappingTemplate=HS_ASSET.FUNDACCOUNT
    #The following selects the message key using the concatenated primary keys
    #gg.handler.kafkahandler.keyMappingTemplate=${primaryKeys}
    gg.handler.kafkahandler.format=json
    gg.handler.kafkahandler.SchemaTopicName=HS_ASSET.FUNDACCOUNT
    gg.handler.kafkahandler.BlockingSend =false
    gg.handler.kafkahandler.includeTokens=false
    gg.handler.kafkahandler.mode=op
    gg.handler.kafkahandler.format.includePrimaryKeys=true
    
    gg.handler.kafkahandler.format.insertOpKey = I
    gg.handler.kafkahandler.format.updateOpKey = U
    gg.handler.kafkahandler.format.deleteOpKey = D
    
    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/ogg/ggjava/resources/lib/*
    #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

假如replicat进程启动失败，看是否有缺失的jar包，并将其copy至相应的位置，需要将kafkaclient.jar放入如下目录下

gg.classpath=dirprm/:/data/ogg/ggjava/resources/lib/*

如下配置了kafka的topicName用来接受该replicat进程的数据

gg.handler.kafkahandler.SchemaTopicName=HS_ASSET.FUNDACCOUNT

如下指定的配置文件主要指定了kafka服务器地址以及相关配置

gg.handler.kafkahandler.KafkaProducerConfigFile=custom_kafka_producer.properties

来看下custom\_kafka\_producer.properties

bootstrap.servers=10.10.48.83:9092,10.10.48.84:9092,10.10.48.85:9092
    acks=1
    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

目标端的replicat进程配置完毕后，使用add replicat 添加replicat进程

GGSCI (oracle221) 8> add replicat RORA_1, exttrail /data/ogg/dirdat/HS_ASSET_FUNDACCOUNT/pa
    REPLICAT added.