DataHub:OGG for Big Data

前提条件

• Oracle データベース、バージョン 19c 以前。データベースのバージョンは、ソース OGG のバージョンより新しくすることはできません。

• ソース：Oracle GoldenGate 19.1.0.0

• ターゲット：Oracle GoldenGate for Big Data 19.1.0.0

• OGG 公式ダウンロードリンク

インストール

このセクションでは、Oracle と OGG のインストールおよび設定プロセスについて説明します。Oracle データベース自体のインストールについては説明しません。

ソース OGG の設定

このセクションでは、Oracle 11g を使用した実用的な例を示します。Oracle 12c 以降のマルチテナントバージョンについては、公式ドキュメントをご参照ください。

1. ソース Oracle データベースの設定

DBA 権限を持つユーザーとして sqlplus にログインします： sqlplus / as sysdba

# 専用の表領域を作成
create tablespace ATMV datafile '/home/oracle/u01/app/oracle/oradata/uprr/ATMV.dbf' size 100m autoextend on next 50m maxsize unlimited;
# パスワード ogg_test で ogg_test という名前のユーザーを作成
create user ogg_test identified by ogg_test default tablespace ATMV;
# ogg_test ユーザーにすべての権限を付与
grant connect,resource,dba to ogg_test;
# 補足ログのステータスを確認
Select SUPPLEMENTAL_LOG_DATA_MIN, SUPPLEMENTAL_LOG_DATA_PK, SUPPLEMENTAL_LOG_DATA_UI, SUPPLEMENTAL_LOG_DATA_FK, SUPPLEMENTAL_LOG_DATA_ALL from v$database;
# データベースの補足ログを追加
alter database add supplemental log data;
alter database add supplemental log data (primary key, unique,foreign key) columns;
# ロールバックコマンド
alter database drop supplemental log data (primary key, unique,foreign key) columns;
alter database drop supplemental log data;
# 全列ロギングモード。注意：このモードでは、DELETE 操作にはプライマリキーの値のみが含まれます。他の値を含めるには、ソースの Extract プロセスで NOCOMPRESSDELETES を設定します。
ALTER DATABASE ADD SUPPLEMENTAL LOG DATA (ALL) COLUMNS;
# データベースの強制ロギングモードを有効化
alter database force logging;
# シーケンスサポートをインストール
@sequence.sql
#
alter table sys.seq$ add supplemental log data (primary key) columns;

2. ソース OGG のインストール

OGG インストールパッケージを解凍すると、次のディレクトリ構造が表示されます：

drwxr-xr-x install
drwxrwxr-x response
-rwxr-xr-x runInstaller
drwxr-xr-x stage

Oracle のインストールでは通常、レスポンスファイルを使用します。response/oggcore.rsp でインストールの依存関係を次のように設定します：

oracle.install.responseFileVersion=/oracle/install/rspfmt_ogginstall_response_schema_v12_1_2
# Oracle のバージョンと一致させる必要があります
INSTALL_OPTION=ORA11g
# GoldenGate のホームディレクトリ
SOFTWARE_LOCATION=/home/oracle/u01/ggate
# 最初は Manager プロセスを開始しない
START_MANAGER=false
# Manager のポート
MANAGER_PORT=7839
# 対応する Oracle のホームディレクトリ
DATABASE_LOCATION=/home/oracle/u01/app/oracle/product/11.2.0/dbhome_1
# 現時点では未設定のままでよい
INVENTORY_LOCATION=
# グループ (この例では、Oracle と OGG の両方で同じユーザー ogg_test を使用します。本番環境では、OGG 用に別のユーザーを作成できます。)
UNIX_GROUP_NAME=oinstall

次のコマンドを実行します：

runInstaller -silent -responseFile {YOUR_OGG_INSTALL_FILE_PATH}/response/oggcore.rsp

この例では、OGG は /home/oracle/u01/ggate ディレクトリにインストールされます。インストールログは /home/oracle/u01/ggate/cfgtoollogs/oui ディレクトリにあります。silentInstall{timestamp}.log ファイル内の次のメッセージは、インストールが成功したことを示します。

/home/oracle/u01/ggate/ggsci コマンドを実行します。プロンプトで CREATE SUBDIRS コマンドを入力して、OGG に必要なディレクトリ (dirxxx) を生成します。これでソース OGG のインストールは完了です。

3. ソース Manager の設定

GGSCI ユーティリティを使用してソース Manager を設定します。edit params mgr コマンドを実行し、次の設定を追加します：

PORT 7839
DYNAMICPORTLIST  7840-7849
USERID ogg_test, PASSWORD ogg_test
PURGEOLDEXTRACTS ./dirdat/*, USECHECKPOINTS, MINKEEPDAYS 7
LAGREPORTHOURS 1
LAGINFOMINUTES 30
LAGCRITICALMINUTES 45
PURGEDDLHISTORY MINKEEPDAYS 3, MAXKEEPDAYS 7
PURGEMARKERHISTORY MINKEEPDAYS 3, MAXKEEPDAYS 7

設定後、view params mgr を実行して設定を確認します。

Manager を開始します： start mgr

Manager のステータスを確認します： info mgr

4. ソース Extract プロセスの設定

GGSCI を使用して Extract プロセスを設定します。この例では、Extract プロセスに dhext という名前を付けます。edit params dhext を実行し、次の設定を追加します：

EXTRACT dhext
SETENV (NLS_LANG="AMERICAN_AMERICA.AL32UTF8")
DBOPTIONS   ALLOWUNUSEDCOLUMN
USERID ogg_test, PASSWORD ogg_test
REPORTCOUNT EVERY 1 MINUTES, RATE
NUMFILES 5000
DISCARDFILE ./dirrpt/ext_test.dsc, APPEND, MEGABYTES 100
DISCARDROLLOVER AT 2:00
WARNLONGTRANS 2h, CHECKINTERVAL 3m
EXTTRAIL ./dirdat/st, MEGABYTES 200
DDL &
INCLUDE MAPPED OBJTYPE 'table' &
INCLUDE MAPPED OBJTYPE 'index' &
INCLUDE MAPPED OBJTYPE 'SEQUENCE' &
EXCLUDE OPTYPE COMMENT
DDLOPTIONS  NOCROSSRENAME  REPORT
TABLE  OGG_TEST.*,tokens (TKN-ROWID=@GETENV('RECORD','rowid'));
SEQUENCE  OGG_TEST.*;
GETUPDATEBEFORES

Extract プロセスを追加して開始します：

# Extract プロセスを追加
add extract dhext,tranlog, begin now

# 各トレイルファイルのサイズを 200 MB に設定
add exttrail ./dirdat/st,extract dhext, megabytes 200

# プロセスを開始
start dhext

開始されると、Extract プロセスはすべてのデータベースの変更を ggate/dirdat ディレクトリ内のファイルにキャプチャします。

5. ソース Pump プロセスの設定

GGSCI を開始し、edit params pump を実行して Pump プロセスを設定します：

EXTRACT pump
RMTHOST xx.xx.xx.xx, MGRPORT 7839, COMPRESS
PASSTHRU
NUMFILES 5000
RMTTRAIL ./dirdat/st
DYNAMICRESOLUTION
TABLE      OGG_TEST.*;
SEQUENCE  OGG_TEST.*;

Pump プロセスを追加して開始します：

# Pump プロセスを追加
add extract pump,exttrailsource ./dirdat/st

# ターゲットのトレイルファイルを追加し、各ファイルのサイズを 200 MB に設定
add rmttrail ./dirdat/st,extract pump,megabytes 200

# Pump プロセスを開始する前に、ターゲットの Manager が実行されていることを確認してください
start pump

Pump プロセスが正常に開始されると、トレイルファイルをターゲットシステムの dirdat ディレクトリに配信します。

ターゲット OGG for Big Data の設定

1. ターゲット OGG for Big Data のインストール

ターゲットの OGG for Big Data は正式なインストールを必要としません。パッケージを解凍するだけです。解凍後、GGSCI を開始し、create subdirs コマンドを実行します。これにより、OGG が必要とするさまざまな dirxxx ディレクトリが作成されます。

2. DataHub プラグインのインストールと設定

ご利用の環境には JDK 1.8 以降がインストールされている必要があります。JAVA_HOME および LD_LIBRARY_PATH 環境変数を設定します。これらを ~/.bash_profile ファイルに追加できます。例：

export JAVA_HOME=/xxx/xxx
export JRE_HOME=/xxx/xxx/jrexx
export LD_LIBRARY_PATH=${LD_LIBRARY_PATH}:$JRE_HOME/lib/amd64:$JRE_HOME/lib/amd64/server

環境変数を変更した後、datahub-ogg-plugin.tar.gz をダウンロードして解凍します。conf ディレクトリで、javaue.properties ファイルを編集し、${GG_HOME} をパッケージの解凍パスに置き換えます。

gg.handlerlist=ggdatahub

gg.handler.ggdatahub.type=com.aliyun.odps.ogg.handler.datahub.DatahubHandler
gg.handler.ggdatahub.configureFileName=${GG_HOME}/aliyun-datahub-ogg-plugin/conf/configure.xml

goldengate.userexit.timestamp=utc+8
goldengate.userexit.writers=javawriter

javawriter.stats.display=TRUE
javawriter.stats.full=TRUE

gg.includeggtokens=true
gg.classpath=${GG_HOME}/aliyun-datahub-ogg-plugin/lib/*
gg.log=log4j
gg.log.level=info
gg.log.file.count=64
gg.log.file.size=128MB

javawriter.bootoptions=-Xms512m -Xmx512m -Xmn256m -Djava.class.path=ggjava/ggjava.jar -Dlog4j.configurationFile=${GG_HOME}/aliyun-datahub-ogg-plugin/conf/log4j.properties

conf ディレクトリで、configure.xml ファイルを編集します。設定のガイダンスについては、ファイル内のコメントをご参照ください。

<?xml version="1.0" encoding="UTF-8"?>
<configue>
    <defaultOracleConfigure>
        <!-- Oracle SID。必須。 -->
        <sid>100</sid>
        <!-- Oracle スキーマ。マッピング内の oracleSchema で上書き可能。いずれかを指定する必要があります。 -->
        <schema>ogg_test</schema>
    </defaultOracleConfigure>
    <defalutDatahubConfigure>
        <!-- DataHub エンドポイント。必須。 -->
        <endPoint>YOUR_DATAHUB_ENDPOINT</endPoint>
        <!-- DataHub プロジェクト。マッピング内の datahubProject で上書き可能。いずれかを指定する必要があります。 -->
        <project>YOUR_DATAHUB_PROJECT</project>
        <!-- DataHub AccessKey ID。マッピング内の datahubAccessId で上書き可能。いずれかを指定する必要があります。 -->
        <accessId>YOUR_DATAHUB_ACCESS_ID</accessId>
        <!-- DataHub AccessKey Secret。マッピング内の datahubAccessKey で上書き可能。いずれかを指定する必要があります。 -->
        <accessKey>YOUR_DATAHUB_ACCESS_KEY</accessKey>
        <!-- データ変更タイプ用の DataHub フィールド。columnMapping 内の ctypeColumn で上書き可能。 -->
        <ctypeColumn>optype</ctypeColumn>
        <!-- データ変更時刻用の DataHub フィールド。columnMapping 内の ctimeColumn で上書き可能。 -->
        <ctimeColumn>readtime</ctimeColumn>
        <!-- データ変更シーケンス番号用の DataHub フィールド。変更ごとに増加しますが、連続しているとは限りません。columnMapping 内の cidColumn で上書き可能。 -->
        <cidColumn>record_id</cidColumn>
    </defalutDatahubConfigure>
    <!-- デフォルトでは、最も厳格なポリシーが適用されます：ダーティデータファイルは書き込まれず、プロセスは即座に終了し、リトライは無限に行われます。 -->
    <!-- バッチあたりの最大レコード数。任意。デフォルト：1000。 -->
    <batchSize>1000</batchSize>
    <!-- 時間フィールドのデフォルトフォーマット。任意。デフォルト：yyyy-MM-dd HH:mm:ss。 -->
    <defaultDateFormat>yyyy-MM-dd HH:mm:ss</defaultDateFormat>
    <!-- ダーティデータで続行するかどうか。任意。デフォルト：false。 -->
    <dirtyDataContinue>true</dirtyDataContinue>
    <!-- ダーティデータファイル名。任意。デフォルト：datahub_ogg_plugin.dirty。 -->
    <dirtyDataFile>datahub_ogg_plugin.dirty</dirtyDataFile>
    <!-- ダーティデータファイルの最大サイズ (MB)。任意。デフォルト：500。 -->
    <dirtyDataFileMaxSize>200</dirtyDataFileMaxSize>
    <!-- リトライ回数。-1：無限、0：リトライなし、n：n回リトライ。任意。デフォルト：-1。 -->
    <retryTimes>0</retryTimes>
    <!-- 使用するシャード ID のカンマ区切りリスト。こちらが優先されます。任意。例：0,1。 -->
    <shardId>0,1</shardId>
    <!-- リトライ間隔 (ミリ秒)。任意。デフォルト：3000。 -->
    <retryInterval>4000</retryInterval>
    <!-- チェックポイントファイル名。任意。デフォルト：datahub_ogg_plugin.chk。 -->
    <checkPointFileName>datahub_ogg_plugin.chk</checkPointFileName>
    <mappings>
        <mapping>
            <!-- Oracle スキーマ。上記の説明をご参照ください。 -->
            <oracleSchema></oracleSchema>
            <!-- Oracle テーブル。必須。 -->
            <oracleTable>t_person</oracleTable>
            <!-- DataHub プロジェクト。上記の説明をご参照ください。 -->
            <datahubProject></datahubProject>
            <!-- DataHub Topic。必須。 -->
            <datahubTopic>t_person</datahubTopic>
            <!-- Oracle テーブルの ROWID 用の DataHub フィールド。任意。 -->
            <rowIdColumn></rowIdColumn>
            <ctypeColumn></ctypeColumn>
            <ctimeColumn></ctimeColumn>
            <cidColumn></cidColumn>
            <columnMapping>
                <!--
                src：ソースの Oracle カラム名。必須。
                dest：ターゲットの DataHub フィールド。必須。
                destOld：更新前データ用のターゲット DataHub フィールド。任意。
                isShardColumn：このカラムをシャーディングのハッシュキーとして使用するかどうか。任意。デフォルト：false。shardId で上書き可能。
                isDateFormat：タイムスタンプフィールドを dateFormat を使用してフォーマットするかどうか。デフォルト：true。false の場合、ソースデータは long 型である必要があります。
                dateFormat：タイムスタンプフィールドのフォーマット。指定しない場合、デフォルトのフォーマットが使用されます。
                -->
                <column src="id" dest="id" isShardColumn="true"  isDateFormat="false" dateFormat="yyyy-MM-dd HH:mm:ss"/>
                <column src="name" dest="name" isShardColumn="true"/>
                <column src="age" dest="age"/>
                <column src="address" dest="address"/>
                <column src="comments" dest="comments"/>
                <column src="sex" dest="sex"/>
                <column src="temp" dest="temp" destOld="temp1"/>
            </columnMapping>
        </mapping>
    </mappings>
</configue>

3. ターゲット Manager の設定

port 7919
dynamicportlist  7910-7919
lagreportminutes  10
laginfoseconds  1
purgeoldextracts ./dirdat/*, usecheckpoints, minkeepdays 7

Manager を開始します： start mgr

4. ターゲット Replicat プロセスの設定

GGSCI で、edit params dhwt を実行して Replicat プロセスを設定します：

REPLICAT dhwt
getEnv (JAVA_HOME)
getEnv (LD_LIBRARY_PATH)
getEnv (PATH)
TARGETDB LIBFILE libggjava.so SET property=${GG_HOME}/aliyun-datahub-ogg-plugin/conf/javaue.properties -- このパスを手動で更新してください。
MAP ogg_test.*, TARGET ogg_test.*;

dhwt プロセスを追加して開始します：

# Replicat プロセスを追加
add replicat dhwt, exttrailsource dirdat/st
# dhwt プロセスを開始
start dhwt

よくある質問

Q：DataHub プラグインの問題をトラブルシューティングするためのログファイルはどこにありますか？

A：DataHub プラグインのログは、OGG for Big Data のインストール先の dirrpt ディレクトリにあります。

各 Replicat プロセスは、[プロセス名]-datahub.log という名前の独自のログファイルを生成します。たとえば、Replicat プロセス名が DHWT の場合、対応するプラグインのログファイルは DHWT-datahub.log です。DataHub の接続性、データフォーマット、または書き込みエラーに関連する問題については、このファイルを確認してください。

Q：一般的な OGG Replicat プロセスのログはどこにありますか？

A：レポートファイルとしても知られる一般的な OGG Replicat プロセスのログは、OGG for Big Data のインストール先の dirrpt ディレクトリにあります。

これらのファイルには、プロセスのステータス、エラー、およびパフォーマンスメトリックに関する詳細情報が含まれています。ログファイル名は [プロセス名].log (または .rpt) です。たとえば、DHWT という名前の Replicat プロセスの場合、レポートファイルは DHWT.log になります。

Q：DataHub でレプリケートされたデータはどのように表示されますか？

A：データが正常に DataHub に書き込まれると、DataHub プラグインのログに確認メッセージが表示されます：

2020-12-01 11:29:30.000461 [record_writer-1-thread-1] INFO ShardWriter - Write DataHub success, table: orders, topic: orders_new, shard: 0, recordNum: 1, rt: 3

DataHub では、レプリケートされたレコードには、元のテーブルのカラムに加えて、変更イベントに関するコンテキストを提供するためにプラグインによって追加されたいくつかのメタデータカラムが含まれます。

以下は、DataHub コンソールでのレプリケートされたレコードの例です：