| Oracle® Fusion Middleware Oracle GoldenGate for Big Dataの統合 リリース12c (12.2.0.1) E72569-02 |
|
 前 |
 次 |
Apache Kafkaは、オープン・ソースで、パーティション化および複製された分散型のメッセージング・サービスです。Kafkaとその関連ドキュメントは、www.kafka.apache.orgを参照してください。
Kafkaは、単一インスタンスとしても、複数サーバー上のクラスタとしても実行できます。Kafkaの各サーバー・インスタンスは、ブローカと呼ばれます。Kafkaにおけるトピックは、メッセージがプロデューサによって公開され、コンシューマによって取得されるときのカテゴリ、またはフィード名です。
Kafkaハンドラは、シリアライズされた変更取得データを複数の表から1つのトピックに書き込むKafkaプロデューサを実装します。
この章は次の項で構成されています:
ここでは、Kafkaハンドラの各コンポーネントの設定とハンドラの実行について説明します。
KafkaおよびKafkaブローカ(単数または複数)の前提条件となるコンポーネントであるZooKeeperが稼働している必要があります。
データ・トピックとスキーマ・トピック(該当する場合)は、実行中のKafkaブローカで事前に構成されていることを、ベスト・プラクティスとしてお薦めします。Kafkaトピックを動的に作成することはできますが、そのためにはKafkaブローカが動的トピックを許可するように構成されている必要があります。
Kafkaブローカが、Oracle GoldenGate for Big Data Kafka Handlerプロセスと同じ場所に配置されていない場合は、Kafkaハンドラを実行しているマシンから、リモートのホスト:ポートにアクセスできる必要があります。
KafkaハンドラをKafkaに接続して実行するには、gg.classpath構成変数に2つのものを含める必要があります。必要なのは、Kafkaプロデューサ・プロパティ・ファイルと、Kafkaクライアントjarです。Kafkaクライアントjarは、Kafkaハンドラが接続するKafkaのバージョンと一致する必要があります。必要なクライアントJARファイルのバージョン別リストは、「Kafkaハンドラ・クライアント依存性」を参照してください。
Kafkaプロデューサ・プロパティ・ファイルの格納場所として推奨されるのは、Oracle GoldenGateのdirprmディレクトリです。
Kafkaクライアントjarのデフォルトの場所は、Kafka_Home/libs/*です。
gg.classpathは、指示に従って正確に構成する必要があります。Kafkaプロデューサ・プロパティ・ファイルのパス指定に含めるパスには、ワイルドカードを使用しないでください。Kafkaプロデューサ・プロパティ・ファイルのパスにワイルドカード(*)を含めると、選択されなくなります。逆に、依存関係jarのパス指定には、そのディレクトリにあるjarファイルがすべて関連するクラスパスに含まれるように、ワイルドカード(*)を含める必要があります。*.jarは使用しないでください。 正しく構成されたクラスパスの例を次に示します。
gg.classpath=dirprm:/ggwork/kafka/lib/*
Kafkaハンドラは、次のものを含めてすべてのビッグ・データ・フォーマッタをサポートします。
Avro行
Avro Operation
JSON
XML
区切りテキスト
Kafkaハンドラの設定可能な値は次のとおりです。これらのプロパティは、Java Adapterプロパティ・ファイルにあり、Replicatプロパティ・ファイルにはありません。
表5-1 12.2.0.1 Kafkaハンドラの構成プロパティ
| プロパティ名 | プロパティ値 | 必須 | 説明 |
|---|---|---|---|
|
|
はい |
使用するハンドラのリスト。 |
|
|
はい |
使用するハンドラのタイプ。Kafka、Flume、HDFSなど。 |
|
任意のカスタム・ファイル名 |
いいえ。デフォルトは |
クラスパスにあるファイル名は、Apache Kafkaプロデューサを構成するApache Kafkaプロパティを保持します。 |
|
TopicName |
はい |
ペイロード・レコードが送信されるKafkaトピックの名前。 |
|
フォーマッタ・クラスまたはショート・コード |
いいえ。デフォルトは |
ペイロードのフォーマットに使用するフォーマッタ。 |
|
スキーマ・トピックの名前 |
はい。スキーマ配信が必要な場合 |
スキーマ・データが配信されるトピック名。このプロパティを設定する場合、スキーマは伝播されません。スキーマが伝播されるのは、Avroフォーマッタの場合のみです。 |
|
Oracle GoldenGate for Big Data Kafka Handlerの |
いいえ。デフォルトの実装クラス |
スキーマは |
|
|
いいえ。デフォルトは |
このプロパティをtrueに設定すると、Kafkaへの配信は完全に非同期のモデルで動作します。次のペイロードは、現在のペイロードが目的のトピックに出力され、確認が受信されてから送信されます。そのための、トランザクション・モードでは厳密に1回のセマンティックが提供されます。このプロパティをfalseに設定すると、Kafkaへの配信は非同期モデルで動作します。ペイロードは、確認を待たずに順次送信されます。Kafkaの内部キューでコンテンツをバッファすると、スループットを増やすことができます。チェックポイントは、Javaコールバックを使用してKafkaブローカから確認が受信されたときにのみ作成されます。 |
|
Oracle GoldenGate for Big Data Kafka Handlerの |
いいえ。そのまま使用できるデフォルトの実装クラス |
Kafkaでのデータの単位 - |
|
|
いいえ。デフォルトは |
Kafkaハンドラが操作モードの場合、変更取得データレコード(Insert、Update、Deleteなど)の各ペイロードはKafkaプロデューサ・レコードとして表され、一度に1つずつフラッシュされます。Kafkaハンドラがトランザクション・モードの場合、ソース・トランザクション内の操作はすべて、1つのKafkaプロデューサ・レコードとして表されます。組み合されたこのバイト・ペイロードが、トランザクション・コミット・イベント時にフラッシュされます。 |
|
|
いいえ |
Kafkaに公開されるデータを表ごとにパーティション化するかどうかを決定します。 tableに設定すると、異なる表のデータは異なるKafkaトピックに書き込まれます。 noneに設定すると、異なる表のデータは |
プロパティ・ファイルについては、これ以降のセクションで説明します。
アダプタ・プロパティ・ファイルからのKafkaハンドラの構成例を次に示します。
gg.handlerlist = kafkahandler gg.handler.kafkahandler.Type = kafka gg.handler.kafkahandler.KafkaProducerConfigFile = custom_kafka_producer.properties gg.handler.kafkahandler.TopicName = oggtopic gg.handler.kafkahandler.Format = avro_op gg.handler.kafkahandler.SchemaTopicName = oggSchemaTopic gg.handler.kafkahandler.ProducerRecordClass = com.company.kafka.CustomProducerRecord gg.handler.kafkahandler.SchemaPrClassName = com.company.kafkaProdRec.SchemaRecord gg.handler.kafkahandler.Mode = tx gg.handler.kafkahandler.BlockingSend = true
Kafka統合のレプリケーション構成とJavaアダプタ・プロパティ・ファイルの例は、次のディレクトリにあります。
GoldenGate_install_directory/AdapterExamples/big-data/kafka
Kafkaハンドラは、メッセージをKafkaに公開するために、Kafkaプロデューサ構成ファイルにアクセスする必要があります。Kafkaプロデューサ構成ファイルのファイル名は、Kafkaハンドラ・プロパティの次の構成によって制御されます。
gg.handler.kafkahandler.KafkaProducerConfigFile=custom_kafka_producer.properties
Kafkaハンドラは、Javaのクラスパスを使用してKafkaプロデューサ構成ファイルを検索およびロードしようとします。したがって、Javaのクラスパスに、Kafkaプロデューサ構成ファイルがあるディレクトリが含まれている必要があります。
Kafkaプロデューサ構成ファイルには、Kafka専有のプロパティがあります。0.8.2.0のKafkaプロデューサ・インタフェース・プロパティについては、Kafkaドキュメントに構成情報が記載されています。Kafkaハンドラは、これらのプロパティを使用してKafkaブローカのホストおよびポートを解決し、Kafkaプロデューサ・クライアントとKafkaブローカとの間のインタフェースの動作は、Kafkaプロデューサ構成ファイルにあるプロパティによって制御されます。
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 # 100KB per partition batch.size = 102400 linger.ms = 10000 max.request.size = 5024000 send.buffer.bytes = 5024000
この項では、Kafkaハンドラの操作のモードの詳細について説明します。
Kafkaハンドラは、Kafka ProducerRecordクラスのインスタンスをKafkaプロデューサAPIに送信し、次にそれがProducerRecordをKafkaトピックに公開します。Kafka ProducerRecordは、実質的に1つのKafkaメッセージの実装です。ProducerRecordには、キーと値の2つのコンポーネントがあります。キーと値のどちらも、Kafkaハンドラによってバイト配列として表されます。この項では、Kafkaハンドラがデータを公開する方法について説明します。
トランザクション・モード
トランザクション・モードは、Kafkaハンドラの次の構成によって示されます。
gg.handler.name.Mode=tx
トランザクション・モードでは、ソースOracle GoldenGate証跡ファイルからのトランザクションにおける各操作のシリアライズ・データが連結されます。連結された操作データの内容は、KafkaのProducerRecordオブジェクトの値です。Kafka ProducerRecordオブジェクトのキーはNULLです。結果として、Kafkaメッセージはには1からNまでの操作のデータが含まれます。Nは、トランザクションにおける操作の数です。グループ化されたトランザクションの場合、グループ化されたトランザクションの全操作の全データが1つのKafkaメッセージとして連結されます。そのため、Kafkaメッセージは大量の操作のデータを含んで非常に大きいメッセージになる可能性がある。
操作モード
操作モードは、Kafkaハンドラの次の構成によって示されます。
gg.handler.name.Mode=op
操作モードでは、各操作のシリアライズ・データは個々のProducerRecordオブジェクトに値として配置されます。ProducerRecordキーは、ソース操作の完全修飾表名です。ProducerRecordは、KafkaプロデューサAPIを使用してすぐに送信されます。つまり、着信する操作と、生成されるKafkaメッセージの数との間には1対1の関係があります。
Kafkaハンドラは、ブロック・モード(同期)または非ブロック・モード(非同期)で、Kafkaにメッセージを送信できます。
ブロック・モード
ブロック・モードは、Kafkaハンドラの次の構成によって設定されます。
gg.handler.name.BlockingSend=true
このモードでは、メッセージは同期ベースでKafkaに配信されます。Kafkaハンドラは、現在のメッセージが目的のトピックに書き込まれて確認が受信されるまで、次のメッセージを送信しません。ブロック・モードでは、メッセージ配信が最も確実に保証されますが、負荷が高いためパフォーマンスは下がります。
ブロック・モードでは、Kafkaプロデューサにlinger.ms変数は設定しないでください。そうすると、KafkaプロデューサはKafkaブローカにメッセージを送信する前にタイムアウト期間ずっと待機します。この場合、Kafkaブローカに送信されるメッセージをKafkaプロデューサがバッファすると同時に、メッセージが送信されたという確認を、Kafkaハンドラは待機しています。したがって、これらの設定は矛盾します。
非ブロック・モード
非ブロック・モードは、Kafkaハンドラの次の構成によって設定されます。
gg.handler.name.BlockingSend=false
このモードでは、メッセージは非同期ベースでKafkaに配信されます。Kafkaメッセージは、確認を待たずに順次送信されます。Kafkaプロデューサ・クライアントは、スループットを高めるために着信メッセージをバッファできます。
トランザクションがコミットされるたびに、Kafkaプロデューサに対するブロック・コールを呼び出し、内部でバッファしている可能性があるKafkaプロデューサ・クライアントの操作をすべてフラッシュします。そのため、Kafkaハンドラは安全なチェックポイントでデータの損失をゼロにします。トランザクション・コミットを呼び出すたびに、最悪の場合でも最大でlinger.msの間ブロックが実行されます。ミリ秒間隔の単位で、linger.msの時間は短く設定することをお薦めします。
KafkaプロデューサがデータをKafkaブローカにフラッシュするタイミングは、Kafkaプロデューサ構成ファイルにある変更可能な多数の構成プロパティによって制御できます。Kafkaプロデューサによるメッセージの一括送信を有効にするには、Kafkaプロデューサ構成ファイルでbatch.sizeとlinger.msの両方のKafkaプロデューサ・プロパティを設定する必要があります。Kafkaへの送信前にバッファする最大バイト数はbatch.sizeによって制御され、データ送信前に待機する最大時間(ミリ秒)はlinger.ms変数によって制御されます。batch.sizeに達するか、linger.msの時間が経過するか、どちらか早いほうの条件が満たされると、データはKafkaに送信されます。batch.size変数のみを設定すると、メッセージはただちにKafkaに送信されます。
Kafkaハンドラではソース証跡ファイルからの操作データを、操作データの対応する表名に基づいて個々のトピックに公開できます。この機能を使用して、ソース証跡ファイルからの操作データをソース表名別に分類できます。この機能は、Java Adapterプロパティ・ファイルで次のように構成を設定して有効にします。
gg.handler.kafka.topicPartitioning=table gg.handler.kafka.mode=op
モードはopに設定する必要があります。また使用するKafkaトピックはソース操作の完全修飾の表名です。
Kafkaハンドラを使用して複数のトピックに公開できます。たとえば、gg.handler.kafkahandler.topicPartitioningプロパティをtableに設定してテーブルごとに1つのトピックを公開できます。
トピックは自動的に作成されて、完全修飾の表名と同じトピック名が使用されます。
Kafkaブローカ設定
トピックの自動作成を有効にするには、Kafkaブローカ構成でauto.create.topics.enableプロパティをtrueに設定します。このプロパティのデフォルト値はtrueです。
Kafkaブローカ構成でauto.create.topics.enableプロパティがfalseに設定されている場合、Replicatプロセスを開始する前に必要なすべてのトピックを手動で作成する必要があります。
スキーマの伝播
すべての表のスキーマ・データが、schemaTopicNameプロパティで構成されたスキーマ・トピックに提供されます。詳細は、「スキーマの伝播」を参照してください。
注意: 複数のトピックがopモードでのみサポートされます。たとえば、gg.handler.kafkahandler.topicPartitioningがtableに設定されている場合は、gg.handler.kafkahandler.modeをopに設定する必要があります。
Kafkaハンドラには、スキーマをスキーマ・トピックに公開する機能があります。現在、スキーマ公開が有効なのは、Avro行フォーマッタとAvro操作フォーマッタのみです。KafkaハンドラのschemaTopicNameプロパティが設定されている場合、スキーマは次のイベントで公開されます。
特定の表のAvroスキーマは、その表に対する最初の操作が発生するときに公開されます。
Kafkaハンドラがメタデータ変更イベントを受信すると、スキーマはフラッシュされます。特定の表について生成されるAvroスキーマは、その表に対する次の操作が発生するときに公開されます。
Avroのラッピング機能を有効にしている場合、汎用のラッパーAvroスキーマは、任意の操作が最初に発生するときに公開されます。汎用のラッパーAvroスキーマの機能は、Avroフォーマッタの構成で有効にできます。正確な手順については、このドキュメントの「Avro行フォーマッタ」または「Avro操作フォーマッタ」の項を参照してください。
KafkaのProducerRecord値はスキーマになり、キーは完全修飾の表名になります。
Kafka上でAvroを使用すると、Avroスキーマに対するAvroメッセージの直接依存性が原因で問題が発生する可能性があります。Avroメッセージはバイナリなので、人間が読める形式ではありません。Avroメッセージをデシリアライズするには、まず受信者に正しいAvroスキーマが必要です。ソース・データベースからの各表が個々のAvroスキーマになるため、問題になる可能性があります。Oracle GoldenGate証跡ファイルに複数の表からの操作が含まれる場合、Kafkaメッセージの受信者が、個々のメッセージをデシリアライズするためにどのAvroスキーマを使用するかを判断することはできません。この問題を解決するために、特殊なAvroメッセージを汎用のAvroメッセージ・ラッパーでラップする機能が用意されています。この汎用のAvroラッパーを使用すれば、完全修飾の表名、スキーマ文字列のハッシュコード、ラップされたAvroメッセージを確認することができます。受信者は、完全修飾の表名と、スキーマ文字列のハッシュコードを使用して、ラップされたメッセージに関連付けられたスキーマを解決し、そのスキーマを使用して、ラップされたメッセージをデシリアライズできます。
この項では、トラブルシューティングのオプションについて説明します。
コマンドラインのKafkaプロデューサを使用すると、ダミー・データをKafkaトピックに書き込むことができ、Kafkaコンシューマを使用してこのデータをKafkaトピックから読み取ることができます。これを利用して、Kafkaトピックの設定および読取り/書込み権限を検証することができます。詳細は、次のサイトでオンラインのKafkaドキュメントを参照してください
非常に一般的なのが、Javaクラスパスに関する問題です。この問題は、log4jログ・ファイルでClassNotFoundExceptionとして表れるのが一般的ですが、gg.classpath変数に誤字がある場合には、クラスパスの解決エラーになることもあります。Kafkaクライアント・ライブラリは、Oracle GoldenGate for Big Data製品に付属しません。JavaとKafkaクライアント・ライブラリを適切に解決するために、適切なバージョンのKafkaクライアント・ライブラリを取得し、Javaアダプタ・プロパティ・ファイルでgg.classpathプロパティを適切に構成する作業は、ユーザーが行います。
Oracle GoldenGate for Big Data Kafka Handlerは、Kafka 0.8.2で新しく導入された推奨のKafkaプロデューサAPIを使用します。0.8.2より古いバージョンのKafkaに接続しようとすると、ランタイム・エラーになります。この問題の回避策はありません。顧客はKafka 0.8.2またはそれ以上に統合する必要があります。
この問題は一般的に、次の例外として発生します。
ERROR 2015-11-11 11:49:08,482 [main] Error loading the kafka producer properties
gg.handler.kafkahandler.KafkaProducerConfigFile構成変数で、Kafkaプロデューサ構成ファイルの名前が正しく設定されていることを検証する必要があります。gg.classpath変数もチェックし、クラスパスにKafkaプロデューサ・プロパティ・ファイルのパスが含まれていることと、プロパティ・ファイルのパスの末尾にワイルドカード(*)が含まれていないことを確認してください。
この問題は、KafkaハンドラがKafkaに接続できない場合に発生します。この問題は、次の警告として発生します。
WARN 2015-11-11 11:25:50,784 [kafka-producer-network-thread | producer-1] WARN (Selector.java:276) - Error in I/O with localhost/127.0.0.1 java.net.ConnectException: Connection refused
最終的には、接続の再試行時間が経過し、Kafkaハンドラは異常終了します。Kafkaブローカが稼働しており、Kafkaプロデューサ・プロパティ・ファイルで指定されているホストとポートが正しいことを確認してください。Kafkaブローカをホストしているマシンでネットワーク・シェル・コマンド(netstat -lなど)を使用すると、Kafkaが所定のポートでリスニングしていることを検証できます。
gg.handler.{name}.BlockingSend=trueの場合、Kafka構成ファイルでlinger.ms設定は使用しないことをお薦めします。使用すると、ブロックへの送信のたびに、少なくともlinger.msの時間が発生するため、パフォーマンスに大きく影響するためです。問題は、Kafkaハンドラ構成とKafkaプロデューサ・構成が競合していることです。このような構成の結果、Kafkaハンドラが送信確認を待つ一方で、Kafkaプロデューサは送信前に次のメッセージを待機することになり、一時的なデッドロックが発生します。このデッドロックは、linger.msの時間が過ぎると解決されます。こうしたシナリオが、送信されるメッセージのたびに繰り返されます。
最高のパフォーマンスを得るためには、Kafkaプロデューサを非ブロック(非同期)コールを使用し、トランザクション・モードで動作するようにKafkaハンドラを設定することをお薦めします。これには、Javaアダプタ・ファイルで次のように設定してください。
gg.handler.{name}.Mode = tx
gg.handler.{name}.BlockingSend = false
また、Kafkaプロデューサ・プロパティ・ファイルでbatch.sizeとlinger.msの値も設定することを推奨します。batch.sizeとlinger.msに設定する値は、ユースケース別の状況によって大きく異なります。一般的には、値が大きいほどスループットは向上しますが、レイテンシが大きくなります。これらのパラメータの値を小さくすると、レイテンシは小さくなりますが、全体的なスループットは低下します。ユースケースによって、ソース証跡データから大量の入力データを伴う場合、batch.sizeとlinger.msは許容できる範囲で大きく設定するようにしてください。
Replicat変数をGROUPTRANSOPSを使用しても、パフォーマンスは向上します。その推奨設定は10000です。
ソース証跡ファイルからシリアライズした操作を個々のKafkaメッセージで配信するのはお客様が行いますが、そのときKafkaハンドラは操作モードに設定されている必要があります。
gg.handler.{name}.Mode = op
そのため、Kafkaメッセージの数が増え、パフォーマンスに悪影響が生じます。
Kafka 0.8.2.2以前では、セキュリティをサポートしていません。Kafka 0.9.0.0にSSL/TLSまたはKerberosを介したセキュリティが導入されました。Oracle GoldenGate KafkaハンドラはSSL/TLSまたはKerberos使用して保護できます。Kafkaプロデューサ・クライアント・ライブラリは、それらのライブラリを利用する統合からのセキュリティ機能の抽象化を提供します。Oracle GoldenGate Kafkaハンドラは、セキュリティ機能から効率的に抽象化されます。セキュリティを有効にするには、Kafkaクラスタのセキュリティを設定し、マシンを接続して、Kafkaプロデューサ・プロパティ・ファイル(Oracle GoldenGate Kafkaハンドラで処理のために使用する)を必要なセキュリティ・プロパティを使用して構成する必要があります。Kafkaクラスタの保護に関する詳細は、次の場所にあるKafkaドキュメントを参照してください。
http://kafka.apache.org/documentation.html#security_configclients
Oracle GoldenGate for Big Data Kafka Handlerは、Kafka 0.8.2.0で新しく導入された推奨のKafkaプロデューサ・インタフェースを実装します。Kafkaハンドラは、0.8.1.0以下のバージョンのKafkaとは互換性がありません。
Kafkaハンドラは次のバージョンのApache Kafkaと互換します。
0.9.0.x
0.8.2.x
HDP 2.4 (Kafka 0.9.0)
HDP 2.3 (Kafka 0.8.2.0)
Cloudera Distribution of Apache Kafka 2.0.x (Kafka 0.9.0.0)
Cloudera Distribution of Apache Kafka 1.x (Kafka 0.8.2.0)
メタデータ変更イベントが、Kafkaハンドラによって処理されるようになりました。ただし、これが関係するのは、スキーマ・トピックを構成しており、使用するフォーマッタがスキーマ伝播をサポートしている場合のみです(現在は、Avro行フォーマッタとAvro操作フォーマッタ)。スキーマが変更されている表で次に操作が発生するとき、更新されたスキーマがスキーマ・トピックに公開されます。
メタデータ変更イベントをサポートするには、ソース・データベースで変更を取得するOracle GoldenGateプロセスが、DDL変更と証跡でのメタデータの両方をサポートする必要があります。GoldenGateでは、すべてのデータベース実装でDDLレプリケーションがサポートされているわけではありません。DDLレプリケーションがサポートされているかどうかを確認するには、データベース実装ごとのOracle GoldenGateドキュメントを参照することをお薦めします。
Kafkaプロデューサ構成ファイルは、圧縮の使用をサポートしています。構成オプションの1つがSnappyです。Snappyは、オープン・ソースの圧縮および圧縮解除(コーデック)ライブラリで、他のコーデック・ライブラリよりパフォーマンスが高い傾向があります。ただしSnappyには、Snappy jarがすべてのプラットフォームには対応していないという欠点があります。Snappyは、Linuxシステムではユニバーサルに動作するようですが、他のUNIXおよびWindowsの実装では動作が保証されていません。Snappyの圧縮を使用する場合は、Snappyを使用する圧縮を実装する前に、必要なシステムすべてでSnappyをテストするといいでしょう。必要なシステムの一部にSnappyが対応しない場合は、別のコーデック・ライブラリを使用することをお薦めします。
