7.3.2. Debezium PostgreSQL 変更イベントの値
変更イベントの値はキーよりも若干複雑です。キーと同様に、値には schema セクションと payload セクションがあります。schema セクションには、ネストされたフィールドを含む、payload セクションの Envelope 構造を記述するスキーマが含まれます。データを作成、更新、または削除する操作のすべての変更イベントには、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操作の再送信イベントには、テーブルのすべてのコラムの以前の値が含まれます。 -
INDEXindex-name:UPDATEおよびDELETE操作の発生したイベントには、指定されたインデックスに含まれるコラムの以前の値が含まれます。UPDATEイベントには、更新された値を持つインデックス化された列も含まれます。
作成イベント
以下の例は、顧客 テーブルのデータを作成する操作に対してコネクターによって生成される変更イベントの値の部分を示しています。
{
"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.5.4.Final",
"connector": "postgresql",
"name": "PostgreSQL_server",
"ts_ms": 1559033904863,
"snapshot": true,
"db": "postgres",
"sequence": "[\"24023119\",\"24023128\"]"
"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 のシステムクロックを基にします。 |
更新イベント
サンプル 顧客 テーブルの更新の変更イベントの値には、そのテーブルの 作成 イベントと同じスキーマがあります。同様に、イベント値のペイロードは同じ構造を持ちます。ただし、イベント値ペイロードでは 更新 イベントに異なる値が含まれます。以下は、コネクターが 顧客 テーブルの更新に対して生成するイベントの変更イベント値の例になります。
{
"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.5.4.Final",
"connector": "postgresql",
"name": "PostgreSQL_server",
"ts_ms": 1559033904863,
"snapshot": false,
"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": { ... },
"payload": {
"before": { 1
"id": 1
},
"after": null, 2
"source": { 3
"version": "1.5.4.Final",
"connector": "postgresql",
"name": "PostgreSQL_server",
"ts_ms": 1559033904863,
"snapshot": false,
"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.5.4.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 のシステムクロックを基にします。 |
single TRUNCATE ステートメントが複数のテーブルに適用される場合、切り捨てられたテーブルごとに 1 つの 切り捨て(truncate) 変更イベントレコードが出力されます。
切り捨て (truncate) イベントは、テーブル全体に加えた変更を表し、メッセージキーを持たないので、単一のパーティションを持つトピックを使用しない限り、テーブルに関する変更イベント (作成、更新 など) とそのテーブルの 切り捨て (truncate) イベントの順番は保証されません。たとえば、これらのイベントが異なるパーティションから読み取られる場合、コンシューマーは 更新 イベントを 切り捨て (truncate) イベントの後でのみ受け取る可能性があります。
