Oracle實時同步到HDFS、Kafka以及關係資料庫的方法

Oracle裡儲存的結構化資料匯出到Hadoop體系做離線計算是一種常見資料處置手段。近期有場景需要做Oracle到Hadoop體系的實時匯入，這裡以此案例做以介紹。
Oracle作為商業化的資料庫解決方案，自發性的獲取資料庫事務日誌等比較困難，故選擇官方提供的同步工具OGG（Oracle GoldenGate）來解決。

安裝與基本配置

環境說明

軟體配置

角色	資料儲存服務及版本	OGG版本	IP
源伺服器	OracleRelease11.2.0.1	Oracle GoldenGate 11.2.1.0 for Oracle on Linux x86-64	10.0.0.25
目標伺服器	Hadoop 2.7.2	Oracle GoldenGate for Big Data 12.2.0.1 on Linux x86-64	10.0.0.2

以上源伺服器上OGG安裝在Oracle使用者下，目標伺服器上OGG安裝在root使用者下。

注意

Oracle匯出到異構的儲存系統，如MySQL，DB2，PG等以及對應的不同平臺，如AIX，Windows，Linux等官方都有提供對應的Oracle GoldenGate版本，可在這裡或者在舊版本查詢下載安裝。

Oracle源端基礎配置

將下載到的對應OGG版本放在方便的位置並解壓，本示例Oracle源端最終的解壓目錄為/u01/gg。

配置環境變數
這裡的環境變數主要是對執行OGG的使用者新增OGG相關的環境變數，本示例為Oracle使用者新增的環境變數如下：（/home/oracle/.bash_profile

檔案）
```
export OGG_HOME=/u01/gg/
export LD_LIBRARY_PATH=$ORACLE_HOME/lib:$OGG_HOME:/lib:/usr/lib
export CLASSPATH=$ORACLE_HOME/jdk/jre:$ORACLE_HOME/jlib:$ORACLE_HOME/rdbms/jlib
```

Oracle開啟歸檔模式
使用如下命令檢視當前是否為歸檔模式（archive）

SQL> archive log list 
Database log mode              Archive Mode
Automatic archival             Enabled
Archive destination            /u01/arch_log
Oldest online log sequence     6
Next log sequence to archive   8
Current log sequence           8

如非以上狀態，手動調整即可

SQL> conn / as sysdba(以DBA身份連線資料庫) 
SQL> shutdown immediate(立即關閉資料庫)
SQL> startup mount(啟動例項並載入資料庫，但不開啟)
SQL> alter database archivelog(更改資料庫為歸檔模式)
SQL> alter database open(開啟資料庫)
SQL> alter system archive log start(啟用自動歸檔)

Oracle開啟日誌相關
OGG基於輔助日誌等進行實時傳輸，故需要開啟相關日誌確保可獲取事務內容。通過一下命令檢視當前狀態：
```
SQL> select force_logging, supplemental_log_data_min from v$database;
FOR SUPPLEME--- --------
YES YES
```
如果以上查詢結果非YES，可通過以下命令修改狀態：
```
SQL> alter database force logging;
SQL> alter database add supplemental log data;
```

Oracle建立複製使用者
為了使Oracle裡使用者的複製許可權更加單純，故專門建立複製使用者，並賦予dba許可權

SQL> create tablespaceoggtbsdatafile '/u01/app/oracle/oradata/orcl/oggtbs01.dbf' size 1000M autoextend on;
SQL> create user ggs identified by ggs default tablespaceoggtbs;
User created.
SQL> grant dba to ggs;
Grant succeeded.

最終這個ggs帳號的許可權如下所示：

SQL> select * from dba_sys_privs where GRANTEE='GGS';
GRANTEE                        PRIVILEGE                                ADM
GGS                            DROP ANY DIRECTORY                       NO
GGS                            ALTER ANY TABLE                          NO
GGS                            ALTER SESSION                            NO
GGS                            SELECT ANY DICTIONARY                    NO
GGS                            CREATE ANY DIRECTORY                     NO
GGS                            RESTRICTED SESSION                       NO
GGS                            FLASHBACK ANY TABLE                      NO
GGS                            UPDATE ANY TABLE                         NO
GGS                            DELETE ANY TABLE                         NO
GGS                            CREATE TABLE                             NO
GGS                            INSERT ANY TABLE                         NO
GRANTEE                        PRIVILEGE                                ADM
GGS                            UNLIMITED TABLESPACE                     NO
GGS                            CREATE SESSION                           NO
GGS                            SELECT ANY TABLE                         NO

OGG初始化
進入OGG的主目錄執行./ggsci，進入OGG命令列

[[email protected]_0_25_centos gg]$ ./ggsci 
Oracle GoldenGate Command Interpreter for Oracle
Version 11.2.1.0.3 14400833 OGGCORE_11.2.1.0.3_PLATFORMS_120823.1258_FBO
Linux, x64, 64bit (optimized), Oracle 11g on Aug 23 2012 20:20:21
Copyright (C) 1995, 2012, Oracle and/or its affiliates. All rights reserved.
GGSCI (VM_0_25_centos) 1>
執行create subdirs進行目錄建立
GGSCI (VM_0_25_centos) 4> create subdirs
Creating subdirectories under current directory /u01/gg
Parameter files                /u01/gg/dirprm: already exists
Report files                   /u01/gg/dirrpt: already exists
Checkpoint files               /u01/gg/dirchk: already exists
Process status files           /u01/gg/dirpcs: already exists
SQL script files               /u01/gg/dirsql: already exists
Database definitions files     /u01/gg/dirdef: already exists
Extract data files             /u01/gg/dirdat: already exists
Temporary files                /u01/gg/dirtmp: already exists
Stdout files                   /u01/gg/dirout: already exists

Oracle建立模擬複製庫表
模擬建一個使用者叫tcloud，密碼tcloud，同時基於這個使用者建一張表，叫t_ogg。

SQL> create user tcloud  identified by tcloud default tablespace users;
User created.
SQL> grant dba to tcloud;
Grant succeeded.
SQL> conn tcloud/tcloud;
Connected.
SQL> create table t_ogg(id int ,text_name varchar(20),primary key(id));
Table created.

目標端基礎配置

將下載到的對應OGG版本放在方便的位置並解壓，本示例Oracle目標端最終的解壓目錄為/data/gg。

配置環境變數
這裡需要用到HDFS相關的庫，故需要配置java環境變數以及OGG相關，並引入HDFS的相關庫檔案，參考配置如下：

export JAVA_HOME=/usr/java/jdk1.7.0_75/
export LD_LIBRARY_PATH=/usr/java/jdk1.7.0_75/jre/lib/amd64:/usr/java/jdk1.7.0_75/jre/lib/amd64/server:/usr/java/jdk1.7.0_75/jre/lib/amd64/libjsig.so:/usr/java/jdk1.7.0_75/jre/lib/amd64/server/libjvm.so:$OGG_HOME:/lib
export OGG_HOME=/data/gg

OGG初始化
目標端的OGG初始化和源端類似進入OGG的主目錄執行./ggsci，進入OGG命令列

GGSCI (10.0.0.2) 2> create subdirs
Creating subdirectories under current directory /data/gg
Parameter files                /data/gg/dirprm: already exists
Report files                   /data/gg/dirrpt: already exists
Checkpoint files               /data/gg/dirchk: already exists
Process status files           /data/gg/dirpcs: already exists
SQL script files               /data/gg/dirsql: already exists
Database definitions files     /data/gg/dirdef: already exists
Extract data files             /data/gg/dirdat: already exists
Temporary files                /data/gg/dirtmp: already exists
Credential store files         /data/gg/dircrd: already exists
Masterkey wallet files         /data/gg/dirwlt: already exists
Dump files                     /data/gg/dirdmp: already exists

Oracle源配置

Oracle實時傳輸到Hadoop叢集（HDFS，Hive，Kafka等）的基本原理如圖：

根據如上原理，配置大概分為如下步驟：源端目標端配置ogg管理器（mgr）；源端配置extract程序進行Oracle日誌抓取；源端配置pump程序傳輸抓取內容到目標端；目標端配置replicate程序複製日誌到Hadoop叢集或者複製到使用者自定義的解析器將最終結果落入到Hadoop叢集。

配置全域性變數

在源端伺服器OGG主目錄下，執行./ggsci到OGG命令列下，執行如下命令：

GGSCI (VM_0_25_centos) 1> dblogin userid ggs password ggs
Successfully logged into database.
GGSCI (VM_0_25_centos) 3> view params ./globals
ggschema ggs

其中./globals變數沒有的話可以用edit params ./globals來編輯新增即可（編輯器預設使用的vim）

配置管理器mgr

在OGG命令列下執行如下命令：

GGSCI (VM_0_25_centos) 4> edit param mgr
PORT 7809
DYNAMICPORTLIST 7810-7909
AUTORESTART EXTRACT *,RETRIES 5,WAITMINUTES 3
PURGEOLDEXTRACTS ./dirdat/*,usecheckpoints, minkeepdays 3

說明：PORT即mgr的預設監聽埠；DYNAMICPORTLIST動態埠列表，當指定的mgr埠不可用時，會在這個埠列表中選擇一個，最大指定範圍為256個；AUTORESTART重啟引數設定表示重啟所有EXTRACT程序，最多5次，每次間隔3分鐘；PURGEOLDEXTRACTS即TRAIL檔案的定期清理
在命令列下執行start mgr即可啟動管理程序，通過info mgr可檢視mgr狀態

GGSCI (VM_0_25_centos) 5> info mgr
Manager is running (IP port VM_0_25_centos.7809).

新增複製表

在OGG命令列下執行新增需要複製的表的操作，如下：

GGSCI (VM_0_25_centos) 7> add trandata tcloud.t_ogg
Logging of supplemental redo data enabled for table TCLOUD.T_OGG.
GGSCI (VM_0_25_centos) 8> info trandata tcloud.t_ogg
Logging of supplemental redo log data is enabled for table TCLOUD.T_OGG.
Columns supplementally logged for table TCLOUD.T_OGG: ID.

配置extract程序

配置extract程序OGG命令列下執行如下命令：

GGSCI (VM_0_25_centos) 10> edit params ext2hd
extract ext2hd
dynamicresolution
SETENV (ORACLE_SID = "orcl")
SETENV (NLS_LANG = "american_america.AL32UTF8")
userid ggs,password ggs
exttrail /u01/gg/dirdat/tc
table tcloud.t_ogg;

說明：第一行指定extract程序名稱；dynamicresolution動態解析；SETENV設定環境變數，這裡分別設定了Oracle資料庫以及字符集；userid ggs,password ggs即OGG連線Oracle資料庫的帳號密碼，這裡使用2.3.4中特意建立的複製帳號；exttrail定義trail檔案的儲存位置以及檔名，注意這裡檔名只能是2個字母，其餘部分OGG會補齊；table即複製表的表明，支援*通配，必須以;結尾
接下來在OGG命令列執行如下命令新增extract程序：

GGSCI (VM_0_25_centos) 11> add extract ext2hd,tranlog,begin now
EXTRACT added.

最後新增trail檔案的定義與extract程序繫結：

GGSCI (VM_0_25_centos) 12> add exttrail /u01/gg/dirdat/tc,extract ext2hd
EXTTRAIL added

可在OGG命令列下通過info命令檢視狀態：

GGSCI (VM_0_25_centos) 14> info ext2hd
EXTRACT    EXT2HD    Initialized   2016-11-09 15:37   Status STOPPED
Checkpoint Lag       00:00:00 (updated 00:02:32 ago)
Log Read Checkpoint  Oracle Redo Logs
                     2016-11-09 15:37:14  Seqno 0, RBA 0
                     SCN 0.0 (0)

配置pump程序

pump程序本質上來說也是一個extract，只不過他的作用僅僅是把trail檔案傳遞到目標端，配置過程和extract程序類似，只是邏輯上稱之為pump程序
在OGG命令列下執行：

GGSCI (VM_0_25_centos) 16> edit params push2hd
extract push2hd
passthru
dynamicresolution
userid ggs,password ggs
rmthost 10.0.0.2 mgrport 7809
rmttrail /data/gg/dirdat/tc
table tcloud.t_ogg;

說明：第一行指定extract程序名稱；passthru即禁止OGG與Oracle互動，我們這裡使用pump邏輯傳輸，故禁止即可；dynamicresolution動態解析；userid ggs,password ggs即OGG連線Oracle資料庫的帳號密碼，這裡使用2.3.4中特意建立的複製帳號；rmthost和mgrhost即目標端OGG的mgr服務的地址以及監聽埠；rmttrail即目標端trail檔案儲存位置以及名稱
分別將本地trail檔案和目標端的trail檔案繫結到extract程序：

GGSCI (VM_0_25_centos) 17> add extract push2hd,exttrailsource /u01/gg/dirdat/tc
EXTRACT added.
GGSCI (VM_0_25_centos) 18> add rmttrail /data/gg/dirdat/tc,extract push2hd
RMTTRAIL added.

同樣可以在OGG命令列下使用info檢視程序狀態：

GGSCI (VM_0_25_centos) 19> info push2hd

EXTRACT    PUSH2HD   Initialized   2016-11-09 15:52   Status STOPPED
Checkpoint Lag       00:00:00 (updated 00:01:04 ago)
Log Read Checkpoint  File /u01/gg/dirdat/tc000000
                     First Record  RBA 0

配置define檔案

Oracle與MySQL，Hadoop叢集（HDFS，Hive，kafka等）等之間資料傳輸可以定義為異構資料型別的傳輸，故需要定義表之間的關係對映，在OGG命令列執行：

GGSCI (VM_0_25_centos) 20> edit params tcloud
defsfile /u01/gg/dirdef/tcloud.t_ogg
userid ggs,password ggs
table tcloud.t_ogg;

在OGG主目錄下執行：
./defgen paramfile dirprm/tcloud.prm
完成之後會生成這樣的檔案/u01/gg/dirdef/tcloud.t_ogg，將這個檔案拷貝到目標端的OGG主目錄下的dirdef目錄即可。

目標端的配置

建立目標表（目錄）

這裡主要是當目標端為HDFS目錄或者Hive表或者MySQL資料庫時需要手動先在目標端建立好目錄或者表，建立方法都類似，這裡我們模擬實時傳入到HDFS目錄，故手動建立一個接收目錄即可
hadoop –fs mkdir /gg/replication/hive/

配置管理器mgr

目標端的OGG管理器（mgr）和源端的配置類似，在OGG命令列下執行：

GGSCI (10.0.0.2) 2> edit params mgr
PORT 7809
DYNAMICPORTLIST 7810-7909
AUTORESTART EXTRACT *,RETRIES 5,WAITMINUTES 3
PURGEOLDEXTRACTS ./dirdat/*,usecheckpoints, minkeepdays 3

配置checkpoint

checkpoint即複製可追溯的一個偏移量記錄，在全域性配置裡新增checkpoint表即可

GGSCI (10.0.0.2) 5> edit  params  ./GLOBALS
CHECKPOINTTABLE tcloud.checkpoint

儲存即可

配置replicate程序

在OGG的命令列下執行：

GGSCI (10.0.0.2) 8> edit params r2hdfs
REPLICAT r2hdfs
sourcedefs /data/gg/dirdef/tcloud.t_ogg
TARGETDB LIBFILE libggjava.so SET property=dirprm/hdfs.props
REPORTCOUNT EVERY 1 MINUTES, RATE 
GROUPTRANSOPS 10000
MAP tcloud.t_ogg, TARGET tcloud.t_ogg;

說明：REPLICATE r2hdfs定義rep程序名稱；sourcedefs即在3.6中在源伺服器上做的表對映檔案；TARGETDB LIBFILE即定義HDFS一些適配性的庫檔案以及配置檔案，配置檔案位於OGG主目錄下的dirprm/hdfs.props；REPORTCOUNT即複製任務的報告生成頻率；GROUPTRANSOPS為以事務傳輸時，事務合併的單位，減少IO操作；MAP即源端與目標端的對映關係
其中property=dirprm/hdfs.props的配置中，最主要的幾項配置及註釋如下：

gg.handlerlist=hdfs //OGG for Big Data中handle型別
gg.handler.hdfs.type=hdfs //OGG for Big Data中HDFS目標
gg.handler.hdfs.rootFilePath=/gg/replication/hive/ //OGG for Big Data中HDFS儲存主目錄
gg.handler.hdfs.mode=op //OGG for Big Data中傳輸模式，即op為一次SQL傳輸一次，tx為一次事務傳輸一次
gg.handler.hdfs.format=delimitedtext //OGG for Big Data中檔案傳輸格式
gg.classpath=/usr/hdp/2.2.0.0-2041/hadoop/share/hadoop/common/*:/usr/hdp/2.2.0.0-2041/hadoop/share/hadoop/common/lib/*:/usr/hdp/2.2.0.0-2041/hadoop/share/hadoop/hdfs/*:/usr/hdp/2.2.0.0-2041/hadoop/etc/hadoop/:/data/gg/:/data/gg/lib/*:/usr/hdp/2.2.0.0-2041/hadoop/client/* //OGG for Big Data中使用到的HDFS庫的定義

具體的OGG for Big Data支援引數以及定義可參考地址

最後在OGG的命令列下執行：

GGSCI (10.0.0.2) 9> add replicat r2hdfs exttrail /data/gg/dirdat/tc,checkpointtable tcloud.checkpointtab
REPLICAT added.

將檔案與複製程序繫結即可

測試

啟動程序

在源端和目標端的OGG命令列下使用start [程序名]的形式啟動所有程序。
啟動順序按照源mgr——目標mgr——源extract——源pump——目標replicate來完成。

檢查程序狀態

以上啟動完成之後，可在源端與目標端的OGG命令列下使用info [程序名]來檢視所有程序狀態，如下：
源端：

GGSCI (VM_0_25_centos) 7> info mgr
Manager is running (IP port VM_0_25_centos.7809).
GGSCI (VM_0_25_centos) 9> info ext2hd
EXTRACT    EXT2HD    Last Started 2016-11-09 16:05   Status RUNNING
Checkpoint Lag       00:00:00 (updated 00:00:09 ago)
Log Read Checkpoint  Oracle Redo Logs
                     2016-11-09 16:45:51  Seqno 8, RBA 132864000
                     SCN 0.1452333 (1452333)
GGSCI (VM_0_25_centos) 10> info push2hd
EXTRACT    PUSH2HD   Last Started 2016-11-09 16:05   Status RUNNING
Checkpoint Lag       00:00:00 (updated 00:00:01 ago)
Log Read Checkpoint  File /u01/gg/dirdat/tc000000
                     First Record  RBA 1043

目標端：

GGSCI (10.0.0.2) 13> info mgr 
Manager is running (IP port 10.0.0.2.7809, Process ID 8242).

GGSCI (10.0.0.2) 14> info r2hdfs
REPLICAT   R2HDFS    Last Started 2016-11-09 16:45   Status RUNNING
Checkpoint Lag       00:00:00 (updated 00:00:02 ago)
Process ID           4733
Log Read Checkpoint  File /data/gg/dirdat/tc000000
                     First Record  RBA 0

所有的狀態均是RUNNING即可。（當然也可以使用info all來檢視所有程序狀態）

測試同步更新效果

測試方法比較簡單，直接在源端的資料表中insert，update，delete操作即可。由於Oracle到Hadoop叢集的同步是異構形式，目前尚不支援truncate操作。
源端進行insert操作

SQL> conn tcloud/tcloud
Connected.
SQL> select * from t_ogg;
no rows selected
SQL> desc t_ogg;
 Name                                      Null?    Type
 ----------------------------------------- -------- ----------------------------
 ID                                        NOT NULL NUMBER(38)
 TEXT_NAME                                          VARCHAR2(20)
SQL> insert into t_ogg values(1,'test');
1 row created.
SQL> commit;
Commit complete.

檢視源端trail檔案狀態

[[email protected]_0_25_centos dirdat]$ ls -l /u01/gg/dirdat/tc*
-rw-rw-rw- 1 oracle oinstall 1180 Nov  9 17:05 /u01/gg/dirdat/tc000000

檢視目標端trail檔案狀態

[[email protected] dirdat]# ls -l /data/gg/dirdat/tc*      
-rw-r----- 1 root root 1217 Nov  9 17:05 /data/gg/dirdat/tc000000

檢視HDFS中是否有寫入

hadoop fs -ls /gg/replication/hive/tcloud.t_ogg
-rw-rw-r--   3 root hdfs        110 2016-11-09 17:05
/gg/replication/hive/tcloud.t_ogg/tcloud.t_ogg_2016-11-09_17-05-30.514.txt

注意：從寫入到HDFS的檔案內容看，檔案的格式如下：

ITCLOUD.T_OGG2016-11-09 09:05:25.0670822016-11-09T17:05:30.51200000000000000000001080ID1TEXT_NAMEtest

很明顯Oracle的資料已準實時匯入到HDFS了。匯入的內容實際是一條條的類似流水日誌（具體日誌格式不同的傳輸格式，內容略有差異，本例使用的delimitedtext。格式為操作符資料庫.表名操作時間戳(GMT+0) 當前時間戳(GMT+8) 偏移量欄位1名稱欄位1內容欄位2名稱欄位2內容），如果要和Oracle的表內容完全一致，需要客戶手動實現解析日誌並寫入到Hive的功能，這裡官方並沒有提供介面卡。目前騰訊側已實現該功能的開發。
當然你可以直接把這個HDFS的路徑通過LOCATION的方式在Hive上建外表（external table）達到實時匯入Hive的目的。

總結

OGG for Big Data實現了Oracle實時同步到Hadoop體系的介面，但得到的日誌目前仍需應用層來解析（關係型資料庫如MySQL時OGG對應版本已實現應用層的解析，無需人工解析）。
OGG的幾個主要程序mgr，extract，pump，replicate配置方便，可快速配置OGG與異構關係儲存結構的實時同步。後續如果有新增表，修改對應的extract，pump和replicate程序即可，當然如果是一整個庫，在配置上述2個程序時，使用通配的方式即可。

附錄

OGG到Hadoop體系的實時同步時，可在源端extract和pump程序配置不變的情況下，直接在目標端增加replicate程序的方式，增加同步目標，以下簡單介紹本示例中增加同步到Kafka的配置方法。
本示例中extract，pump程序都是現成的，無需再新增。只需要在目標端增加同步到Kafka的replicate程序即可。

在OGG的命令列下執行：

GGSCI (10.0.0.2) 4> edit params r2kafka
REPLICAT r2kafka
sourcedefs /data/gg/dirdef/tcloud.t_ogg
TARGETDB LIBFILE libggjava.so SET property=dirprm/r2kafka.props
REPORTCOUNT EVERY 1 MINUTES, RATE
GROUPTRANSOPS 10000
MAP tcloud.t_ogg, TARGET tcloud.t_ogg;

replicate程序和匯入到HDFS的配置類似，差異是呼叫不同的配置dirprm/r2kafka.props。這個配置的主要配置如下：

gg.handlerlist = kafkahandler //handler型別
gg.handler.kafkahandler.type = kafka
gg.handler.kafkahandler.KafkaProducerConfigFile=custom_kafka_producer.properties //kafka相關配置
gg.handler.kafkahandler.TopicName =ggtopic //kafka的topic名稱，無需手動建立
gg.handler.kafkahandler.format =json //傳輸檔案的格式，支援json，xml等
gg.handler.kafkahandler.mode =op  //OGG for Big Data中傳輸模式，即op為一次SQL傳輸一次，tx為一次事務傳輸一次
gg.classpath=dirprm/:/usr/hdp/2.2.0.0-2041/kafka/libs/*:/data/gg/:/data/gg/lib/* //相關庫檔案的引用

r2kafka.props引用的custom_kafka_producer.properties定義了Kafka的相關配置如下：

bootstrap.servers=10.0.0.62:6667 //kafkabroker的地址
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
batch.size=102400
linger.ms=10000

以上配置以及其他可配置項可參考

Oracle實時同步到HDFS、Kafka以及關係資料庫的方法

安裝與基本配置

環境說明

注意

Oracle源端基礎配置

目標端基礎配置

Oracle源配置

配置全域性變數

配置管理器mgr

新增複製表

配置extract程序

配置pump程序

配置define檔案

目標端的配置

建立目標表（目錄）

配置管理器mgr

配置checkpoint

配置replicate程序

測試

啟動程序

檢查程序狀態

測試同步更新效果

總結

附錄

相關推薦