13.8. Debezium MongoDB 変更イベントから after 状態のソースドキュメント抽出

Debezium MongoDB コネクターは、構造が複雑な変更イベントを生成します。各イベントメッセージには、以下の部分が含まれます。

以下のフラグメントは、MongoDB の insert 操作後にコネクターが発行する create 変更イベントの基本構造を示しています。

{
  "op": "c",
  "after": "{\"field1\":\"newvalue1\",\"field2\":\"newvalue1\"}",
  "source": { ... }
}

前の例の after フィールドの複雑な形式では、ソースデータベースで発生した変更に関する詳細情報を提供します。ただし、一部のコンシューマーは、ネストされた値が含まれるメッセージを処理できません。元のメッセージの複雑なネストされたフィールドを、より単純で普遍的、かつ互換性のある構造に変換するには、MongoDB のイベントフラット化 SMT を使用します。SMT は、以下の例のようにメッセージ内のネストされたフィールドの構造をフラット化します。

{
  "field1" : "newvalue1",
  "field2" : "newvalue2"
}

Debezium MongoDB コネクターによって生成されたメッセージのデフォルト構造の詳細は、コネクターのドキュメント を参照してください。

MongoDB のイベントフラット化 SMT は、Debezium MongoDB コネクターによって発行された create または update 変更イベントメッセージから after フィールドを抽出します。SMT は、元の変更イベントメッセージを処理した後、after フィールドの内容のみを含む、単純化バージョンを生成します。

ユースケースに応じて、ExtractNewDocumentState SMT を Debezium MongoDB コネクター、または Debezium コネクターが生成するメッセージを消費する sink コネクターに適用できます。SMT を Debezium MongoDB コネクターに適用すると、SMT はコネクターが Apache Kafka に送信する前に出力するメッセージを変更します。Kafka が完全な Debezium 変更イベントメッセージを元の形式で確実に保持するには、SMT を sink コネクターに適用します。

イベントフラット化 SMT を使用して MongoDB コネクターから送信されたメッセージを処理すると、SMT は元のメッセージのレコードの構造を、通常の sink コネクターで使用できるように、適切に型指定された Kafka Connect レコードに変換します。たとえば、SMT は元のメッセージの after 情報を表す JSON 文字列を、任意のコンシューマーが処理できるスキーマ構造に変換します。

オプションとして、MongoDB のイベントフラット化 SMT を設定して、処理中に他の方法でメッセージを変更できます。詳細は 設定トピックを参照してください。

Debezium MongoDB コネクターによって発行されたメッセージを消費する sink コネクター用に、MongoDB のイベントフラット化 (ExtractNewDocumentState) SMT を設定します。

詳細は以下のセクションを参照してください。

transforms=unwrap,...
transforms.unwrap.type=io.debezium.connector.mongodb.transforms.ExtractNewDocumentState

transforms=unwrap,...
transforms.unwrap.type=io.debezium.connector.mongodb.transforms.ExtractNewDocumentState
transforms.unwrap.drop.tombstones=false
transforms.unwrap.delete.handling.mode=drop
transforms.unwrap.add.headers=op

デフォルトでは、イベントフラット化 SMT は MongoDB 配列を Apache Kafka Connect または Apache Avro スキーマと互換性のある配列に変換します。MongoDB 配列には複数のタイプの要素を含めることができますが、Kafka 配列のすべての要素が同じタイプである必要があります。

SMT が環境のニーズを満たす方法で配列をエンコードするようにするには、array.encoding 設定オプションを指定できます。以下の例は、配列エンコーディングを指定するための設定を示しています。

transforms=unwrap,...
transforms.unwrap.type=io.debezium.connector.mongodb.transforms.ExtractNewDocumentState
transforms.unwrap.array.encoding=<array|document>

設定に応じて、SMT は、次のエンコード方法のいずれかを使用して、ソースメッセージ内の配列の各インスタンスを処理します。

次の例は、Debezium MongoDB コネクターが、異種データ型で構成される配列を含むデータベースドキュメントを表す方法を示しています。

{
    "_id": 1,
    "a1": [
        {
            "a": 1,
            "b": "none"
        },
        {
            "a": "c",
            "d": "something"
        }
    ]
}

{
    "_id": 1,
    "a1": {
        "_0": {
            "a": 1,
            "b": "none"
        },
        "_1": {
            "a": "c",
            "d": "something"
        }
    }
}

document エンコーディングオプションを使用すると、SMT は異種要素で構成される任意の配列を処理できます。ただし、このオプションを使用する前に、sink コネクターと他のダウンストリームコンシューマーが複数のデータ型を含む配列を処理できることを常に確認してください。

データベース操作に埋め込みドキュメントが含まれる場合、Debezium MongoDB コネクターは、元のドキュメントの階層構造を反映する Kafka イベントレコードを発行します。つまり、イベントメッセージは、ネストされたフィールド構造のセットとして、ネスト化ドキュメントを表します。ダウンストリームコネクターがネスト化された構造が含まれるメッセージを処理できない環境では、イベントフラット化 SMT を設定して、メッセージ内の階層構造をフラット化できます。フラットなメッセージ構造は、テーブルのようなストレージに適しています。

ネストされた構造をフラット化するように SMT を設定するには、flatten.struct 設定オプションを true に設定します。変換されたメッセージでは、フィールド名がドキュメントソースと一致するよう構築されます。SMT は、親ドキュメントフィールドの名前とネストされたドキュメントフィールドの名前を連結して、フラット化された各フィールドの名前を変更します。名前のコンポーネントを区切るには、flatten.struct.delimiter オプションで定義される区切り文字を使用します。struct.delimiter のデフォルト値はアンダースコア文字 (_) です。

次の例は、SMT がネストされた構造をフラット化するかどうかを指定する設定です。

transforms=unwrap,...
transforms.unwrap.type=io.debezium.connector.mongodb.transforms.ExtractNewDocumentState
transforms.unwrap.flatten.struct=<true|false>
transforms.unwrap.flatten.struct.delimiter=<string>

以下の例は、MongoDB コネクターが出力するイベントメッセージを示しています。メッセージには、ネストされた 2 つのドキュメント b と c のフィールドを含むドキュメント a のフィールドが含まれます。

{
    "_id": 1,
    "a": {
            "b": 1,
            "c": "none"
    },
    "d": 100
}

以下の例のメッセージは、MongoDB の SMT が前のメッセージでネスト化された構造をフラット化した後に表示した出力です。

{
    "_id": 1,
    "a_b": 1,
    "a_c": "none",
    "d": 100
}

結果のメッセージでは、元のメッセージでネストされた b フィールドと c フィールドがフラット化され、名前が変更されます。名前が変更されたフィールドは、親ドキュメント a の名前を、ネストされたドキュメントの名前 a_b および a_c と連結することによって形成されます。新しいフィールド名のコンポーネントは、struct.delimiter 設定プロパティーの設定で定義されているようにアンダースコアで区切られます。

MongoDB では、$unset Operator と $rename Operator の両方がドキュメントからフィールドを削除します。MongoDB コレクションはスキーマレスであるため、更新でドキュメントからフィールドが削除されると、更新済みドキュメントで抜けているフィールドの名前を推測できません。Debezium は、削除されたフィールドの名前をリストする removedFields 要素を含む更新メッセージを発行して、削除済みフィールドに関する情報を必要とする可能性がある sink コネクターまたはその他のコンシューマーをサポートします。

以下の例は、フィールド a を削除する操作の更新メッセージの一部を示しています。

"payload": {
  "op": "u",
  "ts_ms": "...",
  "before": "{ ... }",
  "after": "{ ... }",
  "updateDescription": {
    "removedFields": ["a"],
    "updatedFields": null,
    "truncatedArrays": null
  }
}

上記の例では、before および after はドキュメントが更新される前と後のソースドキュメントの状態を表します。これらのフィールドは、コネクターの capture.mode が次のリストで説明されているように設定されている場合にのみ、コネクターが発行するイベントメッセージに存在します。

完全なドキュメントをキャプチャーするようにコネクターが設定されていると仮定すると、ExtractNewDocumentState SMT が $unset イベントの update メッセージを受信すると、SMT は、次の例に示すように、削除されたフィールドに null 値があることを表し、メッセージを再エンコードします。

{
    "id": 1,
    "a": null
}

完全なドキュメントをキャプチャーするように設定されていないコネクターの場合、SMT が $unset 操作の更新イベントを受信すると、以下の出力メッセージが生成されます。

{
   "a": null
}

SMT がイベントメッセージをフラット化した後、生成されるメッセージでは、イベントを生成した操作が create、update、または初期スナップショットの read であるかを判断できなくなります。通常、delete 操作に関する情報を公開するか、削除に伴うイベントを書き換えるようにコネクターを設定して、削除操作を識別できます。イベントメッセージで廃棄および書き換えに関する情報を公開するようにコネクターを設定する方法は、drop.tombstones および delete.handling.mode プロパティーを参照してください。

イベントメッセージでデータベース操作のタイプを報告させるには、SMT は次の要素のいずれかに op フィールドを追加します。

たとえば、元の操作のタイプを示すヘッダープロパティーを追加するには、次の例のように、変換を追加してから、add.headers プロパティーをコネクター設定に追加します。

transforms=unwrap,...
transforms.unwrap.type=io.debezium.connector.mongodb.transforms.ExtractNewDocumentState
transforms.unwrap.add.headers=op

上記の設定に基づいて、SMT は op ヘッダーをメッセージに追加し、文字列の値を割り当てて操作のタイプを特定することでイベントタイプを報告します。割り当てられた文字列の値は、元の MongoDB 変更イベントメッセージ の op フィールドの値に基づいています。

MongoDB のイベントフラット化 SMT は、元の変更イベントメッセージから簡素化されたメッセージにメタデータフィールドを追加できます。追加されたメタデータフィールドには、2 つのアンダースコア ("__") が接頭辞として付けられます。イベントレコードにメタデータを追加すると、変更イベントが発生したコレクションの名前などのコンテンツを含めたり、レプリカセット名などのコネクター固有のフィールドを含めたりすることができます。現在、SMT は、source、transaction、および updateDescription の変更イベントのサブ構造フィールドのみを追加できます。

MongoDB 変更イベントの構造に関する詳細は、MongoDB コネクターのドキュメント を参照してください。

たとえば、次の設定を指定して、変更イベントのレプリカセット名 (rs) とコレクション名を、最終的なフラット化されたイベントレコードに追加できます。

transforms=unwrap,...
transforms.unwrap.type=io.debezium.connector.mongodb.transforms.ExtractNewDocumentState
transforms.unwrap.add.fields=rs,collection

上記の設定により、フラット化されたレコードに次のコンテンツが追加されます。

{ "__rs" : "rs0", "__collection" : "my-collection", ... }

SMT でメタデータフィールドを delete イベントに追加する場合には、delete.handling.mode オプションの値を rewrite に設定します。

データベースの変更が発生したときに Debezium コネクターが出力する変更イベントメッセージの他に、コネクターはハートビートメッセージなど、他のタイプのメッセージとスキーマ変更およびトランザクションに関するメタデータメッセージも出力します。これらの他のメッセージの構造は、SMT が処理するように設計された変更イベントメッセージの構造とは異なるため、目的のデータ変更メッセージのみを処理するようにコネクターを SMT を選んで適用することが推奨されます。

SMT を選択的に適用する方法の詳細は、変換用の SMT 述語の設定 を参照してください。

以下の表は、MongoDB イベントフラット化 SMT の設定オプションを示しています。

version:VERSION, connector:CONNECTOR, source.ts_ms:EVENT_TIMESTAMP

version:VERSION, connector:CONNECTOR, source.ts_ms:EVENT_TIMESTAMP