Oracle goldengate 实现mysql到kafka同步配置

迷南。 2022-06-10 11:22 296阅读 0赞

一.oracle goldengate技术架构

Oracle GoldenGate 实现原理是通过抽取源端的redo log 或者 archive log ,然后通过TCP/IP投递到目标端,最后解析还原应用到目标端,使目标端实现 同源端数据同步。图1-1 是Oracle GoldenGate 的技术架构。

![Image 1][]

其中target接受的容器包含:kafka,hbase,hdfs,关系数据库等.

Center

二. 准备

在实现oracle goldengate 同步配置前,小求认为你已安装了一下环境

1. jdk配置

源端:jdk1.7或以上

目标端:jdk1.8

2.目标端zookeeper配置

版本:kafka_2.11-0.10.1.

3.目标端kafka配置

版本:zookeeper-3.4.8

详情配置,请见小求相关博文

三.源端

1. 环境

1.1 系统账号

  1. 账号:192.168.3.207root/root
  2. 路径:/usr/local/soft/ggs_source

1.2 系统环境配置

Center 1

![Image 1][]

2. 配置

2.1 ogg启动

GGSCI是GoldenGate Software Command Interface的缩写,它提供了十分丰富的命令 来对GoldenGate 进行各种操作,如创建、修改、监控GoldenGate 进程等。

配置命令:

  1. # cd /usr/local/soft/ggs_source
  2. # ./ggsci

2.2 mysql用户登录

  1. # dblogin sourcedb test@localhost:3306,userid root,password root

2.3 配置manager

Manager进程是GoldenGate 的控制进程。如果把所有的 Oracle 进程比喻为军队,那么 Manager 就相当于司令。Manager 进程运行在源端和目标端上,它主要有以下几个方面的 作用:启动、监控、重启GoldenGate 的其他进程,报告错误及事件,分配数据存储空间, 发布阈值报告等。

每个源端或者目标端有且只能存在一个 Manager 进程。其运行状态有两种即 RUNNING(正在运行)和STOPPED(已经停止)。

在Windows 系统上,Manager进程是作为一个服务来启动的,而在类UNIX 系统中, Manager则是一个操作系统进程。

配置命令:

  1. # edit param mgr
  2. port 17809
  3. dynamicportlist 17800-18000
  4. purgeoldextracts ./dirdat/*,usecheckpoints, minkeepdays 7

备注:源端口与目标端口对应

2.4 启动mgr

  1. # start mgr
  2. 查看启动进程
  3. # info all

2.5 配置extract

Extract 运行在数据库源端,负责从源端数据表或者日志中捕获数据。在早期的 GoldenGate 版本中,它通常被称为Collect 进程。按照其所处的阶段不同,Extract 的作用 可以按照时间来划分。

初始数据装载阶段:在初始数据装载阶段,Extract 进程直接从源端的数据表中抽取数据。 同步变化捕获阶段:初始数据同步完成以后,Extract 进程负责捕获源端数据的变化 (DML和DDL)。

Extract 进程利用其内在的checkpoint 机制,周期性地检查并记录其读写的位置,通常 是写入到一个本地的trail 文件。这种机制是为了保证如果Extract 进程终止或者操作系统 宕机,重新启动 Extract 进程后,GoldenGate 能够恢复到以前的状态,从上一个断点处继续 往下运行,而不会有任何数据损失。

Extract 进程的状态包括STOPPED(正常停止)、STARTING (正在启动)、RUNNING (正在运行)、ABENDED (Abnomal End 的缩写,表示异常结束)。

配置命令:

  1. # edit param ext_wpkg
  2. extract ext_wpkg
  3. setenv (MYSQL_HOME="/data/mysql/data")
  4. tranlogoptions altlogdest /data/mysql/data/mysql-bin.index
  5. dboptions host localhost,connectionport 3306
  6. sourcedb test, userid root,password root
  7. exttrail /usr/local/soft/ggs_source/dirdat/W3
  8. dynamicresolution
  9. gettruncates
  10. GETUPDATEBEFORES
  11. NOCOMPRESSDELETES
  12. NOCOMPRESSUPDATES
  13. table test.wms_test,TOKENS(TK-MY=@GETENV ('GGENVIRONMENT' , 'OSUSERNAME'));

备注:TOKENS,自定义的,可以去除

  1. table test.wms_entry_warehouse_package;
  2. # ADD EXTRACT ext_wpkg, tranlog,begin now
  3. # ADDEXTTRAIL /usr/local/soft/ggs_source/dirdat/W3, EXTRACT ext_wpkg

2.6 配置pump

Pump 进程运行在数据库源端,其作用非常简单。如果源端使用了本地的trail 文件, 那么Pump 进程就会把trail 以数据块的形式通过TCP/IP协议发送到目标端,这通常也是推 荐的方式。Pump 进程本质是Extract 进程的一种特殊形式,如果不使用trail 文件,那么就 是Extract 进程在抽取完数据以后,直接投递到目标端。

与Pump 进程相对应的叫做Server Collector 进程,这个进程不需要引起人们的关注, 因为在实际操作过程中无需对其进行任何配置,所以对人们来说它是透明的。它运行在目 标端,其任务就是把 Extract/Pump 投递过来的数据块重新组装成trail 文件,人们称之为远 程trail 文件。

配置命令:

  1. # edit param pum_wpkg
  2. extract pum_wpkg
  3. rmthost 192.168.3.65,mgrport 17809
  4. rmttrail /usr/local/ogg/ogg/dirdat/WC
  5. passthru
  6. gettruncates
  7. table test.wms_test;
  8. # ADD EXTRACT pum_wcmd,EXTTRAILSOURCE /usr/local/soft/ggs_source/dirdat/W3;
  9. # ADDRMTTRAIL /usr/local/ogg/ogg/dirdat/WC, EXTRACT pum_wpkg

备注:/usr/local/ogg/ogg/dirdat目标端 ogg路径

2.7 配置defgen

  1. # edit param defgen_wpkg
  2. defsfile /usr/local/soft/ggs_source/dirdef/defgen_wpkg.prm
  3. sourcedb test@localhost:3306,userid root,password root
  4. table test.wms_entry_warehouse_wpkg;

备注:用于生成表字段映射

2.8 生成defgen表字段映射

进入ogg根目录

  1. # ./defgen paramfile/usr/local/soft/ggs_source/dirprm/defgen_wpkg.prm

备注:拷贝dirdef/defgen_wpkg.prm文件到目标端dirdef/目录下

2.9 启动extract和pump

  1. # start ext_wpkg
  2. # start pum_wpkg
  3. # info all

Center 2

![Image 1][]

四.目标端

1. 环境

1.1 系统账号

  1. 账号:182.168.3.65:ogg/ogg
  2. 路径:/usr/local/ogg/ogg

1.2 环境变量

  1. # vi ~/.bashrc

Center 3

![Image 1][]

备注:需要jdk1.8

2. 配置

2.1 配置kafka

  1. 定位../dirprm
  2. # vi kafka.props
  3. gg.handlerlist = kafkahandler
  4. gg.handler.kafkahandler.type = kafka
  5. gg.handler.kafkahandler.KafkaProducerConfigFile=custom_kafka_producer.properties
  6. gg.handler.kafkahandler.TopicName =wms_spark
  7. #gg.handler.kafkahandler.format =avro_op
  8. #gg.handler.kafkahandler.format=delimitedtext
  9. #json
  10. gg.handler.kafkahandler.format = json
  11. gg.handler.kafkahandler.format.insertOpKey = I
  12. gg.handler.kafkahandler.format.updateOpKey = U
  13. gg.handler.kafkahandler.format.deleteOpKey = D
  14. gg.handler.kafkahandler.format.prettyPrint = true
  15. gg.handler.kafkahandler.format.jsonDelimiter = CDATA[]
  16. #gg.handler.kafkahandler.format.generateSchema = true
  17. gg.handler.kafkahandler.format.schemaDirectory = dirdef
  18. #gg.handler.kafkahandler.format.treatAllColumnsAsString = true
  19. gg.handler.kafkahandler.format.includePrimaryKeys = true
  20. #gg.handler.kafkahandler.format.includeColumnNames=true
  21. #gg.handler.kafkahandler.format.fieldDelimiter=♫
  22. gg.handler.kafkahandler.SchemaTopicName=wms_spark_test
  23. gg.handler.kafkahandler.BlockingSend =false
  24. gg.handler.kafkahandler.includeTokens=true
  25. gg.handler.kafka.topicPartitioning=table
  26. gg.handler.kafkahandler.mode =op
  27. #gg.handler.kafkahandler.mode =tx
  28. #gg.handler.kafkahandler.maxGroupSize =100, 1Mb
  29. #gg.handler.kafkahandler.minGroupSize =50, 500Kb
  30. goldengate.userexit.timestamp=utc
  31. goldengate.userexit.writers=javawriter
  32. javawriter.stats.display=TRUE
  33. javawriter.stats.full=TRUE
  34. gg.log=log4j
  35. gg.log.level=INFO
  36. gg.report.time=30sec
  37. #Sample gg.classpath for Apache Kafka
  38. gg.classpath=dirprm/:/usr/local/hadoop/kafka_2.11-0.10.1.0/libs/*:/usr/local/ogg/ogg:/usr/local/ogg/ogg/lib/*
  39. #Sample gg.classpath for HDP
  40. #gg.classpath=/etc/kafka/conf:/usr/hdp/current/kafka-broker/libs/*
  41. javawriter.bootoptions=-Xmx512m -Xms32m -Djava.class.path=ggjava/ggjava.jar
  42. # vi custom_kafka_producer.properties
  43. bootstrap.servers=hadoop-test-01:9092,hadoop-test-02:9092,hadoop-test-03:9092
  44. acks=1
  45. compression.type=gzip
  46. reconnect.backoff.ms=1000
  47. value.serializer=org.apache.kafka.common.serialization.ByteArraySerializer
  48. key.serializer=org.apache.kafka.common.serialization.ByteArraySerializer
  49. # 100KB per partition
  50. batch.size=102400
  51. linger.ms=10000

2.2 启动ogg

  1. # ./ggsci

2.3 配置manager

  1. # edit param mgr
  2. PORT 17809
  3. DYNAMICPORTLIST 17810-17909
  4. AUTORESTART EXTRACT *,RETRIES 5,WAITMINUTES 3
  5. PURGEOLDEXTRACTS /usr/local/ogg/ogg/dirdat/*,usecheckpoints, minkeepdays 3

备注:端口与源端端口对应

2.4 启动mgr

  1. # start mgr

2.5 配置replicat

Replicat 进程,通常也把它叫做应用进程。运行在目标端,是数据传递的最后一站, 负责读取目标端trail 文件中的内容,并将其解析为 DML或DDL语句,然后应用到目标数 据库中。

和Extract 进程一样,Replicat也有其内部的checkpoint 机制,保证进程重新启动后可 以从上次记录的位置开始恢复,而无数据损失的风险。 它的运行状态和 Extract 进程一致,包括 STOPPED、STARTING 、RUNNING 、 ABENDED 。

配置命令:

  1. # edit param rep_wpkg
  2. REPLICAT rep_wpkg
  3. sourcedefs /usr/local/ogg/ogg/dirdef/defgen_wpkg.prm
  4. TARGETDB LIBFILE libggjava.so SET property=dirprm/kafka_wms.props
  5. REPORTCOUNT EVERY 1 MINUTES, RATE
  6. GROUPTRANSOPS 10000
  7. #getUpdateBefores
  8. MAP test.wms_test , TARGET test.wms_test,keycols(ID), colMap(USEDEFAULTS,CREATEDBY=UPDATEDBY);

备注:keyclos设置主键,colMap字段对应

如果默认可以直接:

  1. MAP test.wms_test , TARGET test.wms_test;

增加replicat

  1. # add replicat rep_wpkg, exttrail ./dirdat/WC

备注:WC与源端对应

2.6 启动replicat

  1. # start rep_wpkg
  2. # info all

Center 4

![Image 1][]

五.kafka接受消息格式

  1. {
  2. "table":"test.wms_test",
  3. "op_type":"U",
  4. "op_ts":"2017-08-17 11:04:52.602743",
  5. "current_ts":"2017-08-17T19:04:57.208000",
  6. "pos":"00000000070000013868",
  7. "primary_keys":[
  8. "ID"
  9. ],
  10. "tokens":{
  11. "TK-MY":"root"
  12. },
  13. "before":{
  14. "ID":1,
  15. ...
  16. },
  17. "after":{
  18. "ID":1,
  19. ...
  20. }
  21. }

本章主要利用oracle godengate 实现mysql到kafka同步配置,数据格式为json格式。下章,我们将实现mysql到hbase同步配置。

[Image 1]:

发表评论

表情:
评论列表 (有 0 条评论,296人围观)

还没有评论,来说两句吧...

相关阅读