Oracle GoldenGate for Distributed Applications and AnalyticsによるKafkaへのリアルタイム・データ取込み

7.6 Oracle GoldenGate for Distributed Applications and AnalyticsによるKafkaへのリアルタイム・データ取込み

このトピックでは、Oracle GoldenGate for Distributed Applications and Analytics (GG for DAA)を使用してリアルタイムでKafkaにメッセージを取得する方法を手順を追って説明します。

親トピック: クイックスタート

7.6.1 依存性ファイルのインストール

Oracle GoldenGate for Big Dataは、レプリケーション・プロセスでクライアント・ライブラリを使用します。レプリケーション・プロセスを設定する前に、Oracle GoldenGate for Big Dataで使用可能な依存性ダウンローダ・ユーティリティを使用してこれらのライブラリをダウンロードする必要があります。依存性ダウンローダは、Mavenおよび他のリポジトリから依存性jarファイルをダウンロードするシェル・スクリプトのセットです。

必要な依存性ファイルをインストールするには:

依存性ダウンローダのインストール場所に移動します: GG_HOME/opt/DependencyDownloader/。
必要なバージョンを指定してkafka.shを実行します。

図7-34 必要なバージョンを使用したkafka.shの実行

ディレクトリはGG_HOME/opt/DependencyDownloader/dependenciesに作成されます。たとえば、/u01/app/ogg/opt/DependencyDownloader/dependencies/kafka_2.7.0です。

親トピック: Oracle GoldenGate for Distributed Applications and AnalyticsによるKafkaへのリアルタイム・データ取込み

7.6.2 Kafkaプロデューサのプロパティ・ファイルの作成

Oracle GoldenGate for Big Dataは、Kafkaプロデューサ構成ファイルにアクセスして、メッセージをKafkaに公開する必要があります。Kafkaプロデューサ構成ファイルには、Kafka専有のプロパティがあります。

Kafkaプロデューサ構成ファイルを作成するには:

Oracle GoldenGate for Big Dataで、GGBD_Deployment_Home/etc/conf/oggに移動します。

OCIストリーミング用のKafkaプロデューサ構成ファイルを作成します。構成ファイルの例:

bootstrap.servers=localhost:9092
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

親トピック: Oracle GoldenGate for Distributed Applications and AnalyticsによるKafkaへのリアルタイム・データ取込み

7.6.3 Oracle GoldenGate for Big DataでのReplicatの作成

Oracle GoldenGate for Big DataでReplicatを作成するには:

Oracle GoldenGate for Big Data UIの「管理サービス」タブで、「+」記号をクリックしてReplicatを追加します。

図7-35 「管理サービス」タブの「+」をクリックします。
Replicatのタイプを選択して、「次へ」をクリックします。
ここには、クラシックと調整済の2つの異なるReplicatタイプがあります。クラシックReplicatは単一のスレッド・プロセスですが、調整済Replicatは、トランザクションをパラレルに適用するマルチスレッド・プロセスです。
KafKaの場合、複数の脅威でメッセージを送信するとデータの一貫性の問題が発生する可能性があるため、OracleではクラシックReplicatをお薦めします。

図7-36 Replicatのタイプを選択して、「次へ」をクリックします。
基本情報を入力し、「次へ」をクリックします:
1. プロセス名: Replicatの名前
2. 証跡名: 必要な証跡ファイルの名前
3. ターゲット: Kafka
  
  図7-37 プロセス名、証跡名およびターゲット名
「パラメータ・ファイル」の詳細を入力し、「次へ」をクリックします。「パラメータ・ファイル」では、ソースからターゲットへのマッピングを指定するか、ワイルドカード選択でそのままにしておくことができます。

図7-38 「パラメータ・ファイル」の詳細を指定し、「次へ」をクリックします。
Oracle GoldenGate for Big Dataによって、プロパティ・ファイルが自動的に移入されます。次のフィールドを更新します。
- gg.handler.kafkahandler.kafkaProducerConfigFile=「Kafkaプロデューサのプロパティ・ファイルの作成」で作成されたKafkaプロデューサ・ファイルの名前。
- gg.handler.kafkahandler.topicMappingTemplate=ターゲット・トピックの名前 gg.classpath=「依存性ファイルのインストール」でダウンロードされた依存性ファイルへのパス。
「Create and Run」をクリックします。

図7-39 プロパティ・ファイルが更新されます。
Replicatが正常に起動すると、Replicatは実行状態になります。「アクション」/「詳細」/「統計」に移動して、レプリケーション統計を確認できます:

図7-40 レプリケーション統計

ノート:

ターゲットKafkaトピックが存在しない場合は、Oracle GoldenGate for Big Dataによって自動的に作成されます。トピック名を動的に割り当てるには、「テンプレートのキーワード」を参照してください。
Kafkaレプリケーションのパフォーマンスを向上させるには、ブログOracle GoldenGate for Big DataのKafkaハンドラのパフォーマンスを改善する方法を参照してください。
Oracle GoldenGate for Big Dataは、KafkaへのSSLおよびkerberos認証をサポートしています。詳細は、「スキーマの伝播」を参照してください。
Kafka接続の問題の場合、Oracleサポートにお問い合せください。

親トピック: Oracle GoldenGate for Distributed Applications and AnalyticsによるKafkaへのリアルタイム・データ取込み