3.3.2. Debezium PostgreSQL 変更イベントの値
変更イベントの値はキーよりも若干複雑です。キーと同様に、値には schema
セクションと payload
セクションがあります。schema
セクションには、入れ子のフィールドを含む、 Envelope
セクションの payload
構造を記述するスキーマが含まれています。データを作成、更新、または削除する操作のすべての変更イベントには、Envelope 構造を持つ値 payload があります。
変更イベントキーの例を紹介するために使用した、同じサンプルテーブルについて考えてみましょう。
CREATE TABLE customers ( id SERIAL, first_name VARCHAR(255) NOT NULL, last_name VARCHAR(255) NOT NULL, email VARCHAR(255) NOT NULL, PRIMARY KEY(id) );
この表への変更に対する変更イベントの値は、REPLICA IDENTITY
設定およびイベントの目的である操作により異なります。
詳細は、以下を参照してください。
Replica identity
REPLICA IDENTITY は UPDATE
および DELETE
イベントの論理デコードプラグインで利用可能な情報量を決定する PostgreSQL 固有のテーブルレベルの設定です。具体的には、REPLICA IDENTITY
の設定は、UPDATE
または DELETE
イベントが発生するたびに、関係するテーブル列の以前の値に利用可能な情報 (ある場合) を制御します。
REPLICA IDENTITY
には 4 つの可能性があります。
DEFAULT
- テーブルにプライマリーキーがある場合に、UPDATE
およびDELETE
イベントにテーブルのプライマリーキー列の以前の値が含まれることがデフォルトの動作になります。UPDATE
イベントでは、値が変更されたプライマリーキー列のみが存在します。テーブルにプライマリーキーがない場合、コネクターはそのテーブルの
UPDATE
またはDELETE
イベントを出力しません。プライマリーキーのないテーブルの場合、コネクターは 作成 イベントのみを出力します。通常、プライマリーキーのないテーブルは、テーブルの最後にメッセージを追加するために使用されます。そのため、UPDATE
およびDELETE
イベントは便利ではありません。-
NOTHING
:UPDATE
およびDELETE
操作の出力されたイベントにはテーブル列の以前の値に関する情報は含まれません。 -
FULL
:UPDATE
およびDELETE
操作の出力されたイベントには、テーブルの列すべての以前の値が含まれます。 -
INDEX
index-name:UPDATE
およびDELETE
操作の発生したイベントには、指定されたインデックスに含まれる列の以前の値が含まれます。UPDATE
イベントには、更新された値を持つインデックス化された列も含まれます。
作成 イベント
以下の例は、customers
テーブルにデータを作成する操作に対して、コネクターによって生成される変更イベントの値の部分を示しています。
{ "schema": { 1 "type": "struct", "fields": [ { "type": "struct", "fields": [ { "type": "int32", "optional": false, "field": "id" }, { "type": "string", "optional": false, "field": "first_name" }, { "type": "string", "optional": false, "field": "last_name" }, { "type": "string", "optional": false, "field": "email" } ], "optional": true, "name": "PostgreSQL_server.inventory.customers.Value", 2 "field": "before" }, { "type": "struct", "fields": [ { "type": "int32", "optional": false, "field": "id" }, { "type": "string", "optional": false, "field": "first_name" }, { "type": "string", "optional": false, "field": "last_name" }, { "type": "string", "optional": false, "field": "email" } ], "optional": true, "name": "PostgreSQL_server.inventory.customers.Value", "field": "after" }, { "type": "struct", "fields": [ { "type": "string", "optional": false, "field": "version" }, { "type": "string", "optional": false, "field": "connector" }, { "type": "string", "optional": false, "field": "name" }, { "type": "int64", "optional": false, "field": "ts_ms" }, { "type": "boolean", "optional": true, "default": false, "field": "snapshot" }, { "type": "string", "optional": false, "field": "db" }, { "type": "string", "optional": false, "field": "schema" }, { "type": "string", "optional": false, "field": "table" }, { "type": "int64", "optional": true, "field": "txId" }, { "type": "int64", "optional": true, "field": "lsn" }, { "type": "int64", "optional": true, "field": "xmin" } ], "optional": false, "name": "io.debezium.connector.postgresql.Source", 3 "field": "source" }, { "type": "string", "optional": false, "field": "op" }, { "type": "int64", "optional": true, "field": "ts_ms" } ], "optional": false, "name": "PostgreSQL_server.inventory.customers.Envelope" 4 }, "payload": { 5 "before": null, 6 "after": { 7 "id": 1, "first_name": "Anne", "last_name": "Kretchmar", "email": "annek@noanswer.org" }, "source": { 8 "version": "1.4.2.Final", "connector": "postgresql", "name": "PostgreSQL_server", "ts_ms": 1559033904863, "snapshot": true, "db": "postgres", "schema": "public", "table": "customers", "txId": 555, "lsn": 24023128, "xmin": null }, "op": "c", 9 "ts_ms": 1559033904863 10 } }
項目 | フィールド名 | 説明 |
---|---|---|
1 |
| 値のペイロードの構造を記述する、値のスキーマ。変更イベントの値スキーマは、コネクターが特定のテーブルに生成するすべての変更イベントで同じになります。 |
2 |
|
|
3 |
|
|
4 |
|
|
5 |
|
値の実際のデータ。これは、変更イベントが提供する情報です。 |
6 |
|
イベント発生前の行の状態を指定する任意のフィールド。この例のように、 注記
このフィールドを利用できるかどうかは、各テーブルの |
7 |
|
イベント発生後の行の状態を指定する任意のフィールド。この例では、 |
8 |
| イベントのソースメタデータを記述する必須のフィールド。このフィールドには、イベントの発生元、イベントの発生順序、およびイベントが同じトランザクションの一部であるかどうかなど、このイベントと他のイベントを比較するために使用できる情報が含まれています。ソースメタデータには以下が含まれています。
|
9 |
|
コネクターによってイベントが生成される原因となった操作の型を記述する必須文字列。この例では、
|
10 |
|
コネクターがイベントを処理した時間を表示する任意のフィールド。この時間は、Kafka Connect タスクを実行している JVM のシステムクロックを基にします。 |
更新イベント
サンプル customers
テーブルにある更新の変更イベントの値には、そのテーブルの 作成 イベントと同じスキーマがあります。同様に、イベント値のペイロードは同じ構造を持ちます。ただし、イベント値ペイロードでは 更新 イベントに異なる値が含まれます。以下は、コネクターによって customers
テーブルでの更新に生成されるイベントの変更イベント値の例になります。
{ "schema": { ... }, "payload": { "before": { 1 "id": 1 }, "after": { 2 "id": 1, "first_name": "Anne Marie", "last_name": "Kretchmar", "email": "annek@noanswer.org" }, "source": { 3 "version": "1.4.2.Final", "connector": "postgresql", "name": "PostgreSQL_server", "ts_ms": 1559033904863, "snapshot": null, "db": "postgres", "schema": "public", "table": "customers", "txId": 556, "lsn": 24023128, "xmin": null }, "op": "u", 4 "ts_ms": 1465584025523 5 } }
項目 | フィールド名 | 説明 |
---|---|---|
1 |
|
データベースをコミットする前に行にあった値が含まれる任意のフィールド。この例では、テーブルの |
2 |
|
イベント発生後の行の状態を指定する任意のフィールド。この例では、 |
3 |
|
イベントのソースメタデータを記述する必須のフィールド。
|
4 |
|
操作の型を記述する必須の文字列。更新 イベントの値では、 |
5 |
|
コネクターがイベントを処理した時間を表示する任意のフィールド。この時間は、Kafka Connect タスクを実行している JVM のシステムクロックを基にします。 |
行のプライマリーキー/一意キーの列を更新すると、行のキーの値が変更されます。キーが変更されると、3 つのイベントが Debezium によって出力されます。3 つのイベントとは、DELETE
イベント、行の古いキーを持つ 廃棄 (tombstone)、およびそれに続く行の新しいキーを持つイベントです。詳細は次のセクションで説明します。
プライマリーキーの更新
行のプライマリーキーフィールドを変更する UPDATE
操作は、プライマリーキーの変更と呼ばれます。プライマリーキーの変更では、UPDATE
イベントレコードの代わりにコネクターが古いキーの DELETE
イベントレコードと、新しい (更新された) キーの CREATE
イベントレコードを出力します。これらのイベントには通常の構造と内容があり、イベントごとにプライマリーキーの変更に関連するメッセージヘッダーがあります。
-
DELETE
イベントレコードには、メッセージヘッダーとして__debezium.newkey
が含まれます。このヘッダーの値は、更新された行の新しいプライマリーキーです。 -
CREATE
イベントレコードには、メッセージヘッダーとして__debezium.oldkey
が含まれます。このヘッダーの値は、更新された行にあった以前の (古い) プライマリーキーです。
削除 イベント
削除 変更イベントの値は、同じテーブルの 作成 および 更新 イベントと同じ schema
の部分になります。サンプル customers
テーブルの 削除 イベントの payload
部分は以下のようになります。
{ "schema": { ... }, "payload": { "before": { 1 "id": 1 }, "after": null, 2 "source": { 3 "version": "1.4.2.Final", "connector": "postgresql", "name": "PostgreSQL_server", "ts_ms": 1559033904863, "snapshot": null, "db": "postgres", "schema": "public", "table": "customers", "txId": 556, "lsn": 46523128, "xmin": null }, "op": "d", 4 "ts_ms": 1465581902461 5 } }
項目 | フィールド名 | 説明 |
---|---|---|
1 |
|
イベント発生前の行の状態を指定する任意のフィールド。削除 イベント値の |
2 |
|
イベント発生後の行の状態を指定する任意のフィールド。削除 イベント値の |
3 |
|
イベントのソースメタデータを記述する必須のフィールド。削除 イベント値の
|
4 |
|
操作の型を記述する必須の文字列。 |
5 |
|
コネクターがイベントを処理した時間を表示する任意のフィールド。この時間は、Kafka Connect タスクを実行している JVM のシステムクロックを基にします。 |
削除 変更イベントレコードは、この行の削除を処理するために必要な情報を持つコンシューマーを提供します。
プライマリーキーを持たないテーブルに対して生成された 削除 イベントをコンシューマーが処理できるようにするには、テーブルの REPLICA IDENTITY
を FULL
に設定します。テーブルに主キーがなく、テーブルの REPLICA IDENTITY
が DEFAULT
または NOTHING
に設定されている場合、削除 イベントの before
フィールドはありません。
PostgreSQL コネクターイベントは、Kafka のログコンパクション と動作するように設計されています。ログコンパクションにより、少なくとも各キーの最新のメッセージが保持される限り、一部の古いメッセージを削除できます。これにより、トピックに完全なデータセットが含まれ、キーベースの状態のリロードに使用できるようにするとともに、Kafka がストレージ領域を確保できるようにします。
廃棄 (tombstone) イベント
行が削除された場合でも、Kafka は同じキーを持つ以前のメッセージをすべて削除できるため、削除 イベントの値はログコンパクションで動作します。ただし、Kafka が同じキーを持つすべてのメッセージを削除するには、メッセージの値が null
である必要があります。これを可能にするには、PostgreSQL コネクターは、値が null
値以外の同じキーを持つ特別な 廃棄 イベントが含まれる 削除 イベントに従います。
切り捨て (truncate) イベント
切り捨て (truncate) 変更イベントは、テーブルが切り捨てられていることを伝えます。この場合のメッセージキーは null
で、メッセージの値は以下のようになります。
{ "schema": { ... }, "payload": { "source": { 1 "version": "1.4.2.Final", "connector": "postgresql", "name": "PostgreSQL_server", "ts_ms": 1559033904863, "snapshot": false, "db": "postgres", "schema": "public", "table": "customers", "txId": 556, "lsn": 46523128, "xmin": null }, "op": "t", 2 "ts_ms": 1559033904961 3 } }
項目 | フィールド名 | 説明 |
---|---|---|
1 |
|
イベントのソースメタデータを記述する必須のフィールド。切り捨て (truncate) イベント値の
|
2 |
|
操作の型を記述する必須の文字列。 |
3 |
|
コネクターがイベントを処理した時間を表示する任意のフィールド。この時間は、Kafka Connect タスクを実行している JVM のシステムクロックを基にします。 |
1 つの TRUNCATE
ステートメントが複数のテーブルに適用された場合、切り捨てられたテーブルごとに 1 つの切り捨て (truncate) 変更イベントレコードが出力されます。
切り捨て (truncate) イベントは、テーブル全体に加えた変更を表し、メッセージキーを持たないので、単一のパーティションを持つトピックを使用しない限り、テーブルに関する変更イベント (作成、更新 など) とそのテーブルの 切り捨て (truncate) イベントの順番は保証されません。たとえば、これらのイベントが異なるパーティションから読み取られる場合、コンシューマーは 更新 イベントを 切り捨て (truncate) イベントの後でのみ受け取る可能性があります。