Overview
このガイドでは、変更ストリームと MongoDB Kafka ソース コネクタでその変更ストリームがどのように使用されるかについて学習できます。
変更ストリーム
変更ストリームは、データ変更に関するリアルタイム更新を受信できる MongoDB の機能です。 変更ストリームは変更イベント ドキュメント を返します。
変更イベント ドキュメントには、MongoDB の配置で発生した変更とその変更に関連するメタデータを記述するための冪等な指示が含まれています。 変更イベント ドキュメントは、 oplog 内のデータを使用して生成されます。
重要
変更ストリームは MongoDB レプリカセットとシャーディングされたクラスターでのみ実行
スタンドアロンの MongoDB インスタンスでは変更ストリームを生成できません。
変更ストリームのすべての構成オプションのリストを表示するには、「変更ストリームのプロパティ」ページを参照してください。
変更ストリームの詳細については、次のリソースを参照してください。
Change StreamsMongoDBマニュアルの「 」
oplogの詳細については、MongoDB レプリカセットoplog に関する のマニュアル エントリを参照してください。
集計
集計パイプラインを使用して、ソース コネクタの変更ストリームを構成します。 集計パイプラインを使用して次の操作を含むタスクを実行するように connector 変更ストリームを構成できます。
操作タイプによる変更イベントのフィルタリング
プロジェクト固有のフィールド
フィールドの値を更新
フィールドを追加する
変更ストリームによって生成されたデータ量を試行する
変更ストリームで使用できる集計演算子については、MongoDB マニュアルの「変更ストリーム出力の修正」ガイドを参照してください。
集計パイプラインを使用して変更ストリームを変更する例については、次のページを参照してください。
既存のデータのコピーの使用例
イベント構造の変更
すべてのフィールドの説明を含む、変更イベント ドキュメントの完全な構造については、 MongoDB マニュアル を参照してください。
注意
完全なドキュメント オプション
変更操作によって作成または変更されたドキュメントのみを Kafka Connect が受信する場合は、 publish.full.document.only=trueオプションを使用します。 詳細については、「変更ストリームのプロパティ」ページを参照してください。
パフォーマンス
oplog は、インデックスを使用できない特別な Cappedコレクション です。 この制限の詳細については、「 Change Streams の運用に関する推奨事項 」を参照してください。
変更ストリームのパフォーマンスを向上させるには、MongoDB クラスターにより高速なディスクを使用し、WiredTiger キャッシュのサイズを増やします。 WiredTiger キャッシュの設定方法については、「 WiredTiger ストレージ エンジン 」のガイドを参照してください。
ソース コネクタ
ソース コネクタは、MongoDB で 1 つの変更ストリームを開き、その変更ストリームから Kafka にデータを送信することで機能します。ソース コネクタは実行中は変更ストリームを維持し、停止すると変更ストリームを閉じます。
ソース コネクタの変更ストリームを構成するために使用できるオプションを表示するには、「変更ストリーム プロパティ」ページを参照してください。
再開トークン
connectorは、オフセットとして再開トークンを使用します。 オフセットは、処理されたソース データを追跡するために Apache Kafka コネクタが Apache Kafka トピックに保存する値です。 コネクタは、再起動またはクラッシュから回復する必要がある場合に、そのオフセット値を使用します。 再開トークンは、MongoDB oplog 内の変更イベント ドキュメントの_idフィールドを参照するデータの一部です。
コネクタを初めて起動する場合など、ソース コネクタにオフセットがない場合、コネクタは新しい変更ストリームを開始します。 connector が最初の変更イベント ドキュメントを受け取り、そのドキュメントを Apache Kafka に公開すると、connector はそのドキュメントの再開トークンをオフセットとして保存します。
ソース コネクタのオフセットの再開トークン値が MongoDB 配置の oplog のエントリに対応しない場合、コネクタの再開トークンは無効です。 無効な再開トークンから回復する方法については、無効なトークンのトラブルシューティング ガイドを参照してください。
再開トークンの詳細については、次のリソースを参照してください。
MongoDB マニュアルの「変更ストリームの再開」
MongoDB マニュアルの変更イベント
オフセットの詳細については、次のリソースを参照してください。
Kafka Connect
offset.storage.topic構成オプションのドキュメントKafka Connect
OffsetStorageReaderAPIドキュメント