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

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

例えば、UPDATE 変更イベントの構造の一部は次のようになります。

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

この複雑なフォーマットは、システムで発生する変更に関するほとんどの情報を提供します。しかし、その他のコネクターや Kafka エコシステムの他の要素では、通常、以下のような単純なフォーマットのデータが要求されます。

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

コンシューマーが必要とする Kafka レコードのフォーマットを提供するには、イベントフラット化 SMT を設定します。

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

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

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

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

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

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

コネクターの設定に SMT 設定の詳細を追加して、Kafka Connect ソースコネクターまたはシンクコネクターに Debezium イベントフラット化 SMT を設定します。デフォルトの動作を得るためには、.properties ファイルで、以下のように指定します。

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 を設定する際に指定することのできるオプションを説明します。