13.7. Debezium の変更イベントからステート after ソースレコードを抽出する

Debezium は、複雑な構造を持つデータ変更イベントを生成します。それぞれイベントは、以下の 3 つの部分で構成されます。

以下の例は、UPDATE 変更イベントのメッセージ構造の一部を示しています。

{
	"op": "u",
	"source": {
		...
	},
	"ts_ms" : "...",
	"before" : {
		"field1" : "oldvalue1",
		"field2" : "oldvalue2"
	},
	"after" : {
		"field1" : "newvalue1",
		"field2" : "newvalue2"
	}
}

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

イベントフラット化 SMT が前の例のメッセージを処理した後、メッセージ形式が単純化され、次にあるメッセージが生成されます。

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

イベントフラットニング SMT は、Kafka レコードの Debezium 変更イベントから after フィールドを抽出します。SMT は元の変更イベントを after フィールドのみで置き換え、シンプルな Kafka レコードを作成します。

Debezium コネクターまたは Debezium コネクターから出力されるメッセージを使用する sink コネクターに、イベントフラット化 SMT を設定することができます。sink コネクターにイベントフラット化を設定するメリットは、Apache Kafka に保存されるレコードに Debezium の変更イベント全体が含まれることです。SMT を元のコネクターまたは sink コネクターに適用するかどうかの判断は、特定のユースケースによります。

以下の操作のいずれかを実行するように変換を設定することができます。

データベースの DELETE 操作により、Debezium は 2 つの Kafka レコードを生成します。

before 行のデータを含むレコードをドロップする代わりに、イベントフラットニング SMT が以下のいずれかを行うように設定することができます。

同様に、tombstone レコードを破棄する代わりに、イベントフラット化 SMT を設定して tombstone レコードをストリームに維持することができます。

コネクターの設定に SMT 設定の詳細を追加して、Kafka Connect ソースコネクターまたは sink コネクターに Debezium イベントフラット化 SMT を設定します。たとえば、変換のデフォルトの動作を取得するには、次の例のように、オプションを指定せずにコネクター設定に追加します。

transforms=unwrap,...
transforms.unwrap.type=io.debezium.transforms.ExtractNewRecordState

他の Kafka Connect のコネクター設定と同様に、transforms= にコンマで区切られた複数の SMT エイリアスを設定し、Kafka Connect に SMT を適用させたい順番に設定することができます。

次の .properties の例では、いくつかのイベントフラットニング SMT オプションを設定しています。

transforms=unwrap,...
transforms.unwrap.type=io.debezium.transforms.ExtractNewRecordState
transforms.unwrap.drop.tombstones=false
transforms.unwrap.delete.handling.mode=rewrite
transforms.unwrap.add.fields=table,lsn

イベントフラット化 SMT を設定して、元の変更イベントメタデータを簡素化した Kafka レコードに追加できます。たとえば、簡素化したレコードのヘッダーまたは値に、次のいずれかの項目を含めることができます。

簡略化された Kafka レコードのヘッダーにメタデータを追加するには、add.header オプションを指定します。簡略化された Kafka レコードの値にメタデータを追加するには、add.fields オプションを指定します。これらのオプションには、それぞれ変更イベントフィールド名のコンマ区切りリストを設定します。スペースは指定しないでください。フィールド名が重複している場合、それらのフィールドの 1 つのメタデータを追加するには、フィールドと共に構造体を指定します。以下に例を示します。

transforms=unwrap,...
transforms.unwrap.type=io.debezium.transforms.ExtractNewRecordState
transforms.unwrap.add.fields=op,table,lsn,source.ts_ms
transforms.unwrap.add.headers=db
transforms.unwrap.delete.handling.mode=rewrite

この設定では、簡素化した Kafka レコードには以下のような内容が含まれます。

{
 ...
	"__op" : "c",
	"__table": "MY_TABLE",
	"__lsn": "123456789",
	"__source_ts_ms" : "123456789",
 ...
}

また、簡略化された Kafka のレコードには、__db ヘッダーが付いています。

簡素化した Kafka レコードでは、SMT はメタデータフィールド名の前にダブルアンダースコアを追加します。また、構造体を指定すると、SMT は構造体名とフィールド名の間にアンダースコアを挿入します。

DELETE 操作用のシンプルな Kafka レコードにメタデータを追加するには、delete.handling.mode=rewrite も設定する必要があります。

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

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

次の表で、イベントフラット化 SMT を設定する際に指定することのできるオプションを説明します。