2. 程式人生 > 其它 >使用ogg同步資料給kafka

使用ogg同步資料給kafka

阿新 • • 發佈：2021-10-28

使用ogg同步資料給kafka

2017-06-20 14:55:3123011收藏9 分類專欄：大資料-Kafka 大資料-Kafka專欄收錄該內容 5 篇文章0 訂閱 訂閱專欄

為更好的配合公司自然人專案的實施，我利用虛機搭了一個測試環境，把生產庫中的部分表的資料通過ogg軟體同步給kafka

1 測試環境說明
1）目標
源資料庫hr使用者下的一張表t1,當發生dml操作時，操作資料寫到kafka叢集裡，並顯示出來。
2）環境構成
192.168.60.88 tdb1
192.168.60.89 reps
192.168.60.91 kafka01
192.168.60.92 kafka02
192.168.60.93 kafka03

tdb1是源,Oracle資料庫,版本是10.2.0.4,sid:tdb
reps是介面伺服器，用於安裝ogg for bigdata軟體，接收源端ogg pump程序發過來的資料，並把這些資料寫入kafka叢集
kafka01-03這三臺組成了一個kafka的叢集
所有這些伺服器的os都是rhel 6.8
tdb1,kafka01-03的jdk版本是1.7.0.79
reps的jdk版本是1.8.0.121,因ogg for bigdata 12.2.xxx的版本需要jdk的版本是1.8以上，jdk 1.7會報版本不足的錯
這五臺伺服器都需關掉防火牆，關掉selinux

2 kafka叢集的安裝

1)下載軟體
zookeeper,版本3.4.10,檔名：zookeeper-3.4.10.tar.gz,下載地址:http://zookeeper.apache.org/releases.html
kafka，版本:2.10-0.10.2.0,檔名：kafka_2.10-0.10.2.0.tgz,下載地址:http://kafka.apache.org/downloads
2)準備工作
kafka01-03這三臺機器調整/etc/hosts檔案

  1. [root@kafka01 ~]# cd /etc
  2. [root@kafka01 etc]# cat hosts
  3. 127.0.0.1 localhost localhost.localdomain localhost4 localhost4.localdomain4
  4. ::1 localhost localhost.localdomain localhost6 localhost6.localdomain6
  5. 192.168.60.91 kafka01
  6. 192.168.60.92 kafka02
  7. 192.168.60.93 kafka03

kafka01-03這三臺機器都建一個名字為kafka的使用者，這三臺機器的kafka使用者需要做ssh互信任，如何做可以baidu。
kafka01主機,kafka使用者示例：

  1. [kafka@kafka01 ~]$ pwd
  2. /home/kafka
  3. [kafka@kafka01 ~]$ id
  4. uid=500(kafka) gid=501(kafka) groups=501(kafka)
  5. [kafka@kafka01 ~]$
  6. [kafka@kafka01 ~]$ cat .bash_profile
  7. # .bash_profile
  9. # Get the aliases and functions
  10. if [ -f ~/.bashrc ]; then
  11. . ~/.bashrc
  12. fi
  14. # User specific environment and startup programs
  16. PATH=$PATH:$HOME/bin
  18. export PATH
  19. export JAVA_HOME=/usr/java/jdk1.7.0_79
  20. export JRE_HOME=/usr/java/jdk1.7.0_79/jre
  21. export CLASSPATH=.:$JAVA_HOME/lib:$JRE_HOME/lib:$CLASSPATH
  22. export ZOOKEEPER_HOME=/home/kafaka/zookeeper
  23. export PATH=$JAVA_HOME/bin:$ZOOKEEPER_HOME/bin:$PATH
  24. [kafka@kafka01 ~]$

3)安裝配置zookeeper
如下操作在kafka01-03都要做,如下kafka01上做示例
a.解壓
zookeeper-3.4.10.tar.gz解壓後，目錄重新命名為：zookeeper,並挪到/home/kafka/下，效果如下：

  1. [kafka@kafka01 zookeeper]$ pwd
  2. /home/kafka/zookeeper
  3. [kafka@kafka01 zookeeper]$ ls
  4. bin conf data docs ivy.xml LICENSE.txt README_packaging.txt recipes zookeeper-3.4.10.jar zookeeper-3.4.10.jar.md5 zookeeper.out
  5. build.xml contrib dist-maven ivysettings.xml lib NOTICE.txt README.txt src zookeeper-3.4.10.jar.asc zookeeper-3.4.10.jar.sha1
  6. [kafka@kafka01 zookeeper]$

b.配置zoo.cfg

  1. cd /home/kafka/ zookeeper
  2. cp zoo_sample.cfg zoo.cfg

編輯zoo.cfg,內容如下：

  1. [kafka@kafka01 conf]$ pwd
  2. /home/kafka/zookeeper/conf
  3. zoo.cfg設定如下引數：
  4. dataDir=/home/kafka/zookeeper/data
  5. clientPort=2181
  6. server.1=kafka01:2888:3888
  7. server.2=kafka02:2888:3888
  8. server.3=kafka03:2888:3888

c.設定節點標識

  1. cd /home/kafka/zookeeper
  2. mkdir data
  3. cd data
  4. vi myid
  5. 輸入1
  7. [kafka@kafka01 data]$ pwd
  8. /home/kafka/zookeeper/data
  9. [kafka@kafka01 data]$ cat myid
  10. 1

d.啟動Zookeeper

  1. [kafka@kafka01 bin]$ pwd
  2. /home/kafka/zookeeper/bin
  3. ./zkServer.sh start
  5. 看狀態：
  6. [kafka@kafka01 bin]$ ./zkServer.sh status
  7. ZooKeeper JMX enabled by default
  8. Using config: /home/kafka/zookeeper/bin/../conf/zoo.cfg
  9. Mode: follower
  10. 三臺中一臺Mode是leader,其餘兩臺Mode為follower
  12. 排錯：
  13. 如果沒啟動，可以使用./zkServer.sh start-foreground啟動，螢幕上會顯示日誌資訊，能看出哪塊出了問題。

4)安裝配置kafka
如下操作在kafka01-03都要做,kafka01上示例如下：
a.解壓
kafka_2.10-0.10.2.0.tgz解壓後，目錄重新命名為：kafka,並挪到/home/kafka/下，效果如下：

  1. [kafka@kafka02 kafka]$ pwd
  2. /home/kafka/kafka
  3. [kafka@kafka02 kafka]$ ls
  4. bin config libs LICENSE logs NOTICE site-docs

b.修改Kafka Servre配置檔案

  1. cd /home/kafka/kafka/config
  2. vi server.properties
  4. [kafka@kafka01 config]$ cat server.properties --注：不需改的條目去掉了
  5. broker.id=1 #kafka01為1，kafka02為2，kafka03為3
  6. host.name=kafka01 #按主機名相應調整
  7. listeners=PLAINTEXT://kafka01:9092 #按主機名相應調整
  8. advertised.listeners=PLAINTEXT://kafka01:9092 #按主機名相應調整
  9. log.dirs=/home/kafka/kafka/logs
  10. num.partitions=4

c.後臺啟動kakfa
在叢集中的這三個節點上分別後臺啟動Kafka，分別執行如下命令：

  1. cd /home/kafka/kafka/bin
  2. ./kafka-server-start.sh/home/kafka/kafka/config/server.properties &

d.測試
建立一個名稱為oggtest的Topic，4個分割槽，並且複製因子為3:
可以任意一節點

  1. ./kafka-topics.sh -create -zookeeper kafka01:2181,kafka02:2181,kafka03:2181 -replication-factor 3 -partitions 3 –topic oggtest

檢視建立的Topic，執行如下命令：
可以任意一節點

  1. ./kafka-topics.sh -describe -zookeeper kafka01:2181,kafka02:2181,kafka03:2181 -topic oggtest
  3. 檢視所有topic:
  4. ./kafka-topics.sh -describe -zookeeper kafka01:2181,kafka02:2181,kafka03:2181

可以通過Kafka自帶的bin/kafka-console-producer.sh和bin/kafka-console-consumer.sh指令碼，來驗證演示如何釋出訊息、消費訊息:
在一個終端，啟動Producer，並向我們上面建立的名稱為oggtest的Topic中生產訊息，執行如下指令碼：

  1. ./kafka-console-producer.sh -broker-list kafka01:9092,kafka02:9092,kafka03:9092 -topic oggtest

在另一個終端，啟動Consumer，並訂閱我們上面建立的Topic中生產的訊息：

  1. ./kafka-console-consumer.sh –zookeeperkafka01:2181,kafka02:2181,kafka03:2181 –from-beginning –topic oggtest

如果kafka叢集配置的沒有問題，隨便在producer視窗敲入一些資訊回車後，在consumer視窗便能看到相應的訊息

3 ogg源端(sdb1主機)的安裝配置
1)準備工作
a.ogg 軟體
軟體名：Oracle GoldenGate V11.2.1.0.3 for Oracle 11g on Linux x86-64.zip,在https://edelivery.oracle.com可以下載到
b.源資料庫要開歸檔，置成force logging,開追加日誌

  1. [oracle@tdb1 ~]$ sqlplus / as sysdba
  2. SQL> archive log list
  3. Database log mode Archive Mode
  4. Automatic archival Enabled
  5. Archive destination /oracle/arc
  6. Oldest online log sequence 9
  7. Next log sequence to archive 11
  8. Current log sequence 11
  9. SQL>
  10. SQL> Select
  11. 2 SUPPLEMENTAL_LOG_DATA_MIN
  12. 3 ,SUPPLEMENTAL_LOG_DATA_PK
  13. 4 ,SUPPLEMENTAL_LOG_DATA_UI
  14. 5 ,SUPPLEMENTAL_LOG_DATA_ALL
  15. 6 , FORCE_LOGGING from v$database;
  17. SUPPLEME SUP SUP SUP FOR
  18. -------- --- --- --- ---
  19. YES NO NO NO YES
  21. SQL>

c.推薦新增一ogg使用者，以避免對oracle使用者的影響，放在oracle使用者的主group,根據資料extract的量為ogg使用者建立一個合適大小的檔案系統
ogg使用者最終效果示例：

  1. [root@tdb1 ~]# su - ogg
  2. [ogg@tdb1 ~]$ id
  3. uid=501(ogg) gid=500(dba) groups=500(dba)
  4. [ogg@tdb1 ~]$
  5. [ogg@tdb1 ~]$ cat .bash_profile
  6. # .bash_profile
  8. # Get the aliases and functions
  9. if [ -f ~/.bashrc ]; then
  10. . ~/.bashrc
  11. fi
  13. # User specific environment and startup programs
  15. PATH=$PATH:$HOME/bin
  17. export PATH
  19. umask 022
  20. export ORACLE_BASE=/oracle/app/oracle
  21. export ORACLE_HOME=$ORACLE_BASE/product/10.2.0
  22. export ORACLE_SID=tdb
  23. export PATH=$ORACLE_HOME/bin:$PATH:.
  24. export NLS_LANG=AMERICAN_AMERICA.ZHS16GBK
  25. export NLS_DATE_FORMAT=YYYYMMDDHH24MISS
  26. export DISPLAY=192.168.60.1:0.0
  27. #ogg
  28. export GG_HOME=/ogg
  29. #export PATH=$PATH:$GG_HOME
  30. export LD_LIBRARY_PATH=/ogg:$ORACLE_HOME/lib
  31. [ogg@tdb1 ~]$
  32. [ogg@tdb1 ~]$ sqlplus / as sysdba
  34. SQL*Plus: Release 10.2.0.5.0 - Production on Thu Apr 13 11:10:59 2017
  36. Copyright (c) 1982, 2010, Oracle. All Rights Reserved.
  39. Connected to:
  40. Oracle Database 10g Enterprise Edition Release 10.2.0.5.0 - 64bit Production
  41. With the Partitioning, Data Mining and Real Application Testing options
  43. SQL>

d.資料庫內建立ogg使用者，並給其授權

  1. create tablespace ogg datafile '/oracle/oradata/tdb/ogg.dbf' size 1G;
  2. create user ogg identified by gg_888 default tablespace ogg;
  3. grant connect,resource to ogg;
  4. grant dba to ogg; --如果不做ddl trigger,dba許可權可以不給
  6. GRANT CREATE SESSION TO ogg;
  7. GRANT ALTER SESSION TO ogg;
  8. GRANT SELECT ANY DICTIONARY TO ogg;
  9. GRANT SELECT ANY TABLE TO ogg;
  10. GRANT ALTER ANY TABLE TO ogg; --使用者配置表級追加日誌
  12. GRANT FLASHBACK ANY TABLE TO ogg;
  13. GRANT EXECUTE on DBMS_FLASHBACK package TO ogg;
  14. GRANT EXECUTE on DBMS_FLASHBACK TO ogg;
  15. GRANT EXECUTE ON utl_file TO ogg;
  16. execute DBMS_STREAMS_AUTH.GRANT_ADMIN_PRIVILEGE('OGG');
  17. grant execute on sys.dbms_lob to ogg;
  19. --如下pl/sql塊是在oracle 11g之上版本用的，10g版本不需要執行
  20. BEGIN
  21. DBMS_GOLDENGATE_AUTH.GRANT_ADMIN_PRIVILEGE(
  22. Grantee => 'OGG',
  23. privilege_type => 'CAPTURE',
  24. grant_select_privileges => TRUE,
  25. do_grants => TRUE);
  26. END;
  27. /

e.為了測試，我建立了一個hr使用者，並在其下面建了一個t1的表

  1. -- Create table
  2. create table T1
  3. (
  4. id NUMBER not null,
  5. name VARCHAR2(100)
  6. )
  7. tablespace USERS;
  8. -- Create/Recreate primary, unique and foreign key constraints
  9. alter table T1
  10. add constraint PK_T1_ID primary key (ID)
  11. using index
  12. tablespace USERS;

2)配置ogg
a.解壓縮ogg軟體，放在$GG_HOME下
效果類似如下：

  1. [ogg@tdb1 ogg]$ ls -l gg*
  2. -rwxr-x--- 1 ogg dba 6577392 Aug 24 2012 ggcmd
  3. -rw-r----- 1 ogg dba 1841 Apr 12 15:58 gglog-defgen.dmp
  4. -rw-r----- 1 ogg dba 1239 Apr 12 16:40 gglog-DPE_TEST-43680.dmp
  5. -rw-r----- 1 ogg dba 962 Apr 12 16:49 gglog-DPE_TEST-43782.dmp
  6. -rw-r----- 1 ogg dba 0 Apr 12 16:40 gglog-DPE_TEST.dmp
  7. -rw-r----- 1 ogg dba 1280336 Aug 24 2012 ggMessage.dat
  8. -rwxr-x--- 1 ogg dba 13899588 Aug 24 2012 ggsci
  9. -rw-rw-rw- 1 ogg dba 21819 Apr 13 08:47 ggserr.log
  10. [ogg@tdb1 ogg]$

b.建立ogg相關子目錄

  1. [ogg@tdb1 ogg]$ pwd
  2. /ogg
  3. [ogg@tdb1 ogg]$ ./ggsci
  4. GGSCI>create subdirs

ggsci報錯處理：

  1. [ogg@tdb1 ogg]$ ggsci
  2. ggsci: error while loading shared libraries: libnnz11.so: cannot open shared object file: No such file or directory
  3. [ogg@tdb1 ogg]$ ldd ggsci
  4. linux-vdso.so.1 => (0x00007ffd3db73000)
  5. libdl.so.2 => /lib64/libdl.so.2 (0x00000035bbc00000)
  6. libgglog.so => /ogg/./libgglog.so (0x00007ff824130000)
  7. libggrepo.so => /ogg/./libggrepo.so (0x00007ff823fdc000)
  8. libdb-5.2.so => /ogg/./libdb-5.2.so (0x00007ff823d3b000)
  9. libicui18n.so.38 => /ogg/./libicui18n.so.38 (0x00007ff8239da000)
  10. libicuuc.so.38 => /ogg/./libicuuc.so.38 (0x00007ff8236a1000)
  11. libicudata.so.38 => /ogg/./libicudata.so.38 (0x00007ff8226c5000)
  12. libpthread.so.0 => /lib64/libpthread.so.0 (0x00000035bc400000)
  13. libxerces-c.so.28 => /ogg/./libxerces-c.so.28 (0x00007ff8221ad000)
  14. libantlr3c.so => /ogg/./libantlr3c.so (0x00007ff822097000)
  15. libnnz11.so => not found
  16. libclntsh.so.11.1 => not found
  17. libstdc++.so.6 => /usr/lib64/libstdc++.so.6 (0x00000035c7400000)
  18. libm.so.6 => /lib64/libm.so.6 (0x00000035bcc00000)
  19. libgcc_s.so.1 => /lib64/libgcc_s.so.1 (0x00000035c7000000)
  20. libc.so.6 => /lib64/libc.so.6 (0x00000035bc000000)
  21. /lib64/ld-linux-x86-64.so.2 (0x00000035bb800000)
  23. [oracle@tdb1 ~]$ cd $ORACLE_HOME/lib
  24. [oracle@tdb1 lib]$ ln -s libnnz10.so libnnz11.so
  25. [oracle@tdb1 lib]$ ln -s libclntsh.so libclntsh.so.11.1
  26. [oracle@tdb1 lib]$ ls -l libclntsh.so.11.1
  27. lrwxrwxrwx 1 oracle dba 12 Apr 11 22:33 libclntsh.so.11.1 -> libclntsh.so
  28. [oracle@tdb1 lib]$ ls -l libnnz11.so
  29. lrwxrwxrwx 1 oracle dba 11 Apr 11 22:31 libnnz11.so -> libnnz10.so

c.開啟hr.t1表級附加日誌

  1. [ogg@tdb1 ogg]$ ./ggsci
  2. GGSCI>Dblogin userid ogg, password gg_888
  3. Add trandata hr.t1

d.配置ogg manager

  1. [ogg@tdb1 ogg]$ ./ggsci
  2. GGSCI>edit params mgr
  3. 內容如下，儲存
  4. PORT 7809
  5. DYNAMICPORTLIST 7810-7860
  6. AUTORESTART EXTRACT *, RETRIES 5, WAITMINUTES 3, RESETMINUTES 60
  7. PURGEOLDEXTRACTS ./dirdat/*, usecheckpoints, minkeepdays 1
  8. LAGREPORTHOURS 1
  9. LAGINFOMINUTES 30
  10. LAGCRITICALMINUTES 45
  12. 啟動OGG manager
  13. GGSCI>start mgr
  15. 檢視manager程序狀態，正確的形態如下：
  16. GGSCI (tdb1) 1> info mgr
  17. Manager is running (IP port tdb1.7809).

e.建立Extract

  1. GGSCI>edit params ext_test
  2. 內容如下，儲存
  3. EXTRACT ext_test
  4. Setenv (NLS_LANG="AMERICAN_AMERICA.ZHS16GBK")
  5. USERID ogg, PASSWORD gg_888
  6. gettruncates
  7. DISCARDFILE ./dirrpt/ext_test.dsc, APPEND, MEGABYTES 1024
  8. DBOPTIONS ALLOWUNUSEDCOLUMN
  9. REPORTCOUNT EVERY 1 MINUTES, RATE
  10. WARNLONGTRANS 2h,CHECKINTERVAL 3m
  11. FETCHOPTIONS NOUSESNAPSHOT
  12. TRANLOGOPTIONS CONVERTUCS2CLOBS
  13. EXTTRAIL ./dirdat/te
  14. WILDCARDRESOLVE DYNAMIC
  15. GETUPDATEBEFORES
  16. NOCOMPRESSUPDATES
  17. NOCOMPRESSDELETES
  18. dynamicresolution
  19. table hr.t1;
  21. 新增抽取程序組：
  22. GGSCI>add extract ext_test, TRANLOG, BEGIN NOW
  24. 定義trail檔案:
  25. GGSCI> ADD EXTTRAIL ./dirdat/te, EXTRACT ext_test, MEGABYTES 200

f.pump extract程序

  1. GGSCI>edit param dpe_test
  2. 內容如下，儲存
  3. EXTRACT dpe_test
  4. PASSTHRU
  5. RMTHOST 192.168.60.89, MGRPORT 7809
  6. RMTTRAIL ./dirdat/te
  7. DYNAMICRESOLUTION
  8. TABLE hr.t1;
  10. 新增pump捕獲組：
  11. GGSCI> ADD EXTRACT dpe_test, EXTTRAILSOURCE ./dirdat/te
  13. 定義pump trail檔案:
  14. GGSCI> ADD RMTTRAIL ./dirdat/te, EXTRACT dpe_test, MEGABYTES 200

g.啟動捕獲程序

  1. GGSCI> start extract ext_test;
  2. GGSCI> start extract dpe_test;
  4. 看狀態，如果如正就是對的：
  5. GGSCI> info all
  7. Program Status Group Lag at Chkpt Time Since Chkpt
  9. MANAGER RUNNING
  10. EXTRACT RUNNING DPE_TEST 00:00:00 00:00:03
  11. EXTRACT RUNNING EXT_TEST 00:00:00 00:00:01

4 介面機reps安裝配置
1)安裝OGG for Big Data
a.如源端類以,解壓縮ogg for big data軟體，放在$GG_HOME下
b./etc/hosts檔案

  1. [root@reps etc]# cd /etc
  2. [root@reps etc]# cat hosts
  3. 127.0.0.1 localhost localhost.localdomain localhost4 localhost4.localdomain4
  4. ::1 localhost localhost.localdomain localhost6 localhost6.localdomain6
  5. 192.168.60.89 reps
  6. 192.168.60.91 kafka01
  7. 192.168.60.92 kafka02
  8. 192.168.60.93 kafka03
  9. [root@reps etc]#

c.安裝jdk 1.8及之以的版本
ogg for big data 12.2.xx需要jdk 1.8以上的版本，我這裡用的是1.8.0_121

  1. [ogg@reps ogg]$ java -version
  2. java version "1.8.0_121"
  3. Java(TM) SE Runtime Environment (build 1.8.0_121-b13)
  4. Java HotSpot(TM) 64-Bit Server VM (build 25.121-b13, mixed mode)

d.建立ogg使用者，配置環境變數，安裝kafka軟體

  1. [root@reps etc]# su - ogg
  2. [ogg@reps ~]$ id
  3. uid=500(ogg) gid=501(ogg) groups=501(ogg)
  4. [ogg@reps ~]$ cat .bash_profile
  5. # .bash_profile
  7. # Get the aliases and functions
  8. if [ -f ~/.bashrc ]; then
  9. . ~/.bashrc
  10. fi
  12. # User specific environment and startup programs
  14. PATH=$PATH:$HOME/bin
  16. export PATH
  17. export OGG_HOME=/ogg
  18. export PATH=$PATH:$GG_HOME
  19. export LD_LIBRARY_PATH=$OGG_HOME:$JAVA_HOME/jre/lib/amd64/libjsig.so:$JAVA_HOME/jre/lib/amd64/server/libjvm.so:$JAVA_HOME/jre/lib/amd64/server:$JAVA_HOME/jre/lib/amd64
  20. [ogg@reps ~]$
  21. [ogg@reps ~]$ ls -l
  22. total 8
  23. drwxrwxr-x 2 ogg ogg 4096 Apr 11 22:56 install
  24. drwxr-xr-x 6 ogg ogg 4096 Feb 15 01:28 kafka --把kafka軟體包解壓到這，也可以從kafka主機拷貝這個目錄
  25. [ogg@reps ~]$
  26. [ogg@reps ~]$ cd /ogg
  27. [ogg@reps ogg]$ ls -l ggsci
  28. -rwxr-x--- 1 ogg ogg 39120528 Oct 20 07:05 ggsci
  29. [ogg@reps ogg]$

2)配置OGG for kafka
a.啟動ogg,並建立相關子目錄

  1. ./ggsci
  2. GGSCI>create subdirs

b.複製example

  1. cp $OGG_HOME/AdapterExamples/big-data/kafka/* $OGG_HOME/dirprm/

c.配置manager

  1. GGSCI>edit params mgr
  2. 內容如下：
  3. PORT 7809
  4. DYNAMICPORTLIST 7810-7860
  5. AUTORESTART REPLICAT *, RETRIES 5, WAITMINUTES 3, RESETMINUTES 60
  6. PURGEOLDEXTRACTS ./dirdat/*, usecheckpoints, minkeepdays 1
  7. LAGREPORTHOURS 1
  8. LAGINFOMINUTES 30
  9. LAGCRITICALMINUTES 45

d.配置kafka.props
內容如下：

  1. [ogg@reps dirprm]$ cat 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 =oggtest
  7. gg.handler.kafkahandler.format =avro_op
  8. gg.handler.kafkahandler.format=delimitedtext
  9. gg.handler.kafkahandler.format.fieldDelimiter=|
  10. gg.handler.kafkahandler.SchemaTopicName=myoggtest
  11. gg.handler.kafkahandler.BlockingSend =false
  12. gg.handler.kafkahandler.includeTokens=false
  14. gg.handler.kafkahandler.mode =op
  15. #gg.handler.kafkahandler.maxGroupSize =100, 1Mb
  16. #gg.handler.kafkahandler.minGroupSize =50, 500Kb
  19. goldengate.userexit.timestamp=utc
  20. goldengate.userexit.writers=javawriter
  21. javawriter.stats.display=TRUE
  22. javawriter.stats.full=TRUE
  24. gg.log=log4j
  25. gg.log.level=INFO
  27. gg.report.time=30sec
  29. #Sample gg.classpath for Apache Kafka
  30. gg.classpath=dirprm/:/home/ogg/kafka/libs/*
  31. #Sample gg.classpath for HDP
  32. #gg.classpath=/etc/kafka/conf:/usr/hdp/current/kafka-broker/libs/*
  34. javawriter.bootoptions=-Xmx512m -Xms32m -Djava.class.path=ggjava/ggjava.jar

說明：
gg.handler.kafkahandler.TopicName必須指定kafka端定義的topic
gg.handler.kafkahandler.format下面配置使用文字，並用”|”相隔，kafka最終接收到如下格式的訊息。
gg.classpath須指定相應的lib路徑
e.配置custom_kafka_producer.properties
內容如下：

  1. bootstrap.servers=kafka01:9092,kafka02:9092,kafka03:9092
  2. acks=1
  3. compression.type=gzip
  4. reconnect.backoff.ms=1000
  6. value.serializer=org.apache.kafka.common.serialization.ByteArraySerializer
  7. key.serializer=org.apache.kafka.common.serialization.ByteArraySerializer
  8. # 100KB per partition
  9. batch.size=102400
  10. linger.ms=10000

3)表結構傳遞
源端：

  1. GGSCI> ggsci
  2. GGSCI> edit param defgen
  3. 內容
  4. DEFSFILE dirdef/source.def, PURGE
  5. USERID ogg, PASSWORD gg_888
  6. TABLE hr.t1 ;
  8. [ogg@tdb1 ogg]$ defgen paramfile dirprm/defgen.prm --shell命令

把defgen.prm放到介面機(reps)的/ogg/dirdef/下
4)定義replication
a.定義引數

  1. ./ggsci
  2. GGSCI>edit params rep_test
  3. 輸入如下內容：
  4. REPLICAT rep_test
  5. TARGETDB LIBFILE libggjava.so SET property=dirprm/kafka.props
  6. SOURCEDEFS dirdef/source.def
  7. REPORTCOUNT EVERY 1 MINUTES, RATE
  8. GROUPTRANSOPS 10000
  9. MAP hr.*, TARGET hr.*;

b.指定Trail檔案

  1. GGSCI> add replicat rep_test, exttrail ./dirdat/te

c.啟動replicat程序，並檢查狀態

  1. GGSCI> start replicat rep_test
  3. 檢查狀態，類似如下輸出表示正常
  4. GGSCI (reps) 1> info all
  6. Program Status Group Lag at Chkpt Time Since Chkpt
  8. MANAGER RUNNING
  9. REPLICAT RUNNING REP_TEST 00:00:00 00:00:05

其它：如果replicat程序啟動不了，可以使用如下命令啟動，以方便診斷問題：

  1. cd $OGG_HOME
  2. ./replicat paramfile dirprm/rep_test.prm

5 測試驗證
1）啟動kafka consumerconsole
kafka任一結點：

  1. ./kafka-console-consumer.sh -zookeeper :2181 -topic oggtest -from-beginning

2)在源端測試表中插入資料

  1. sqlplus hr/hr
  2. SQL> insert into t1 values(5,'shihaifeng');
  3. 1 row created.
  5. SQL> commit;
  6. Commit complete.

3)檢視kafka消費控制檯是否接收到該行資料
我的有如下顯示：

  1. I|HR.T1|2017-04-13 03:31:03.835147|2017-04-13T11:31:08.973000|00000000000000001478|5|shihaifeng

相關推薦

使用ogg同步資料給kafka

使用ogg同步資料給kafka 兩榜進士2017-06-20 14:55:3123011收藏9 分類專欄：大資料-Kafka

利用ogg實現oracle到kafka的增量資料實時同步

前言 ogg即Oracle GoldenGate是Oracle的同步工具，本文講如何配置ogg以實現Oracle資料庫增量資料實時同步到kafka中，其中同步訊息格式為json。

使用ogg實現oracle到kafka的增量資料實時同步

使用ogg實現oracle到kafka的增量資料實時同步 ​彬彬 ​2022.04.07 一、OGG概述 OGG全稱為Oracle GoldenGate，是由Oracle官方提供的用於解決異構資料環境中資料複製的一個商業工具。相比於其它遷移工具OGG的優勢在於

ktl工具實現mysql向mysql同步資料方法

使用ktl工具實現mysql向mysql同步資料 1.新建作業步驟 2.完善作業步驟，主要是完成作業中的轉換工作

Oracle -- 增量同步資料_遊標和merge into 方式同步

  增量抽取 只需抽取新增的或修改的資料。此方法效能好，但容易遺漏。 目標表中有，但是源表中沒有的話，更新不了。

Web API返回自定義資料給客戶端

服務端出現異常時，返回給客戶端status仍然是ok的。因此在前端的catch或是error是得到不到服務端的throw異常資訊的。

【Mysql】Linux系統下的資料庫伺服器主從複製、讀寫分離、同步資料

Linux系統下的資料庫伺服器主從複製、讀寫分離、同步資料到新的從機 （1）主從複製

Oracle同步資料到kafka

目錄文章介紹環境準備軟體準備下載地址實施過程Oracle主機（A）配置Kafka主機（B）配置主機hosts檔案新增解析安裝JDK配置apache-maven工具配置Kafka 2.13-2.6.0配置kafka-connect-oracle-maste啟動kafka-connect-ora

vue中的父元件傳遞資料給子元件、子元件傳遞資料給父元件

vue中的父元件傳遞資料給子元件、子元件傳遞資料給父元件 1、父元件傳遞資料給子元件

Fluentd直接傳輸日誌給kafka

官方文件地址：https://docs.fluentd.org/output/kafka td-agent版本自帶包含out_kafka2外掛，不用再安裝了，可以直接使用。

Elasticsearch5 Python操作Elasticsearch，mysql和Elasticsearch同步資料(開源外掛go-mysql-elasticsearch)，haystack的使用

Python操作Elasticsearch的兩種方式 第一種方式 # 第一種使用方式(推薦使用，簡單)

什麼,kafka能夠從follower副本讀資料了 —kafka新功能介紹

最近看了kafka2.4新版本的一些功能特性，不得不說，在kafka2.0以後，kafka自身就比較少推出一些新的feature了，基本都是一些修修補補的東西。倒是kafka connect和kafka stream相關的開發工作做的比較多。可能kafka的

JAVA後臺重定向頁面並且跨域傳遞資料給前端

技術標籤：後臺重定向並傳輸局javavuehtml 註明： 本文為學習記錄筆記，不喜勿噴。有問題請留言。

tcpserver廣播發送資料給所有client_ThinkSwoole 教程（六） WebSocket 訊息、廣播以及 Swoole 原生方法呼叫...

技術標籤：tcpserver廣播發送資料給所有clientwebsocket 傳送多條訊息如何判斷全部接收websocket官方文件

使用Spark SQL JDBC同步資料至MySQL

Spark SQL JDBC 我們可以使用一個 JDBC 的連結來定義一個 Spark SQL 的表或者檢視，這裡用表來做示例：

java 定時同步資料的任務優化

前言 定時任務在系統中並不少見，主要目的是用於需要定時處理資料或者執行某個操作的情況下，如定時關閉訂單，或者定時備份。而常見的定時任務分為2種，第一種：固定時間執行，如：每分鐘執行一次，每天執行一次。第

父元件傳遞非同步資料給子元件遇到的問題

技術標籤：vue.jsprops/state 父元件可以通過props傳遞非同步資料給子元件，但是值得注意的是，子元件中需要watch去監聽父元件傳過來的值，並且要對對應的函式進行重新呼叫。 程式碼如下：

Vue通過props實現父元件與子元件之間的通訊(父元件傳資料給子元件)

技術標籤：Vue學習vue Vue父元件傳資料給子元件 1.通過props關鍵詞實現父元件傳給子元件1.1在子元件中通過props關鍵詞定義變數1.2通過自己定義的子元件標籤和v-bind進行資料繫結1.3完整例子原始碼1.4測試

soul-websocket同步資料到閘道器原始碼分析

通過官網資料瞭解到，soul支援websocket,http,zookeeper等同步方式。 如果是通過websocket 同步策略，則將變更後的資料主動推送給 soul-web，並且在閘道器層，會有對應的WebsocketCacheHandler 處理器處理來處

響應資料給前端

/** * 響應資料給前端 */ public static void sendJsonMessage(HttpServletResponse response, Object obj) throws IOException {