7.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 IDENTITYUPDATE および 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.7.2.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
    }
}

表7.6 作成 イベント値フィールドの説明

項目フィールド名説明

1

schema

値のペイロードの構造を記述する、値のスキーマ。変更イベントの値スキーマは、コネクターが特定のテーブルに生成するすべての変更イベントで同じになります。

2

name

schema セクションでは、各name フィールドが、値のペイロード内のフィールドのスキーマを指定します。

Postgre SQL_server.inventory.customers.Value は、ペイロードの before および after フィールドのスキーマです。このスキーマは customers テーブルに固有です。

before および after フィールドのスキーマ名はlogicalName.tableName.Value の形式で、スキーマ名がデータベースで一意になるようにします。つまり、Avro コンバーター を使用する場合、各論理ソースの各テーブルの Avro スキーマには独自の進化と履歴があります。

3

name

io.debezium.connector.postgresql.Source は、ペイロードの source フィールドのスキーマです。このスキーマは、PostgreSQL コネクターに固有のものです。コネクターは生成するすべてのイベントにこれを使用します。

4

name

PostgreSQL_server.inventory.customers.Envelope は、ペイロードの全体的な構造のスキーマで、PostgreSQL_server はコネクター名、inventory はデータベース、customers はテーブルを指します。

5

payload

値の実際のデータ。これは、変更イベントが提供する情報です。

イベントの JSON 表現はそれが記述する行よりもはるかに大きいように見えることがあります。これは、JSON 表現にはメッセージのスキーマ部分とペイロード部分を含める必要があるためです。しかし、Avro コンバーター を使用すると、コネクターが Kafka トピックにストリーミングするメッセージのサイズを大幅に小さくすることができます。

6

before

イベント発生前の行の状態を指定する任意のフィールド。この例のように、op フィールドが create (作成) の c である場合、この変更イベントは新しい内容に対するものであるため、beforenull になります。

注記

このフィールドを利用できるかどうかは、各テーブルの REPLICA IDENTITY 設定によって異なります。

7

after

イベント発生後の行の状態を指定する任意のフィールド。この例では、after フィールドには、新しい行の idfirst_namelast_name、および email 列の値が含まれます。

8

source

イベントのソースメタデータを記述する必須のフィールド。このフィールドには、イベントの発生元、イベントの発生順序、およびイベントが同じトランザクションの一部であるかどうかなど、このイベントと他のイベントを比較するために使用できる情報が含まれています。ソースメタデータには以下が含まれています。

  • Debezium バージョン
  • コネクター型および名前
  • 新しい行が含まれるデータベースおよびテーブル
  • 追加のオフセット情報を文字列化した JSON 配列。最初の値は常に最後にコミットされた LSN で、2 番目の値は常に現在の LSN です。いずれの値も null である可能性があります。
  • スキーマ名
  • イベントがスナップショットの一部であるか
  • 操作が実行されたトランザクションの ID
  • データベースログの操作のオフセット
  • データベースに変更が加えられた時点のタイムスタンプ

9

op

コネクターによってイベントが生成される原因となった操作の型を記述する必須文字列。この例では、c は操作によって行が作成されたことを示しています。有効な値は以下のとおりです。

  • c = create
  • u = update
  • d = delete
  • r = read (読み取り、スナップショットのみに適用)

10

ts_ms

コネクターがイベントを処理した時間を表示する任意のフィールド。この時間は、Kafka Connect タスクを実行している JVM のシステムクロックを基にします。

source オブジェクトで、ts_ms は変更がデータベースに加えられた時間を示します。payload.source.ts_ms の値を payload.ts_ms の値と比較することにより、ソースデータベースの更新と Debezium との間の遅延を判断できます。

更新イベント

サンプル 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.7.2.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
    }
}

表7.7 更新 イベント値フィールドの説明

項目フィールド名説明

1

before

データベースをコミットする前に行にあった値が含まれる任意のフィールド。この例では、テーブルの REPLICA IDENTITY 設定がデフォルトでは DEFAULT であるため、プライマリーキー列 id のみが存在します。+ 更新 イベントに、行のすべてのコラムの以前の値が含まれるようにするには、ALTER TABLE customers REPLICA IDENTITY FULL を実行し、customers テーブルを変更する必要があります。

2

after

イベント発生後の行の状態を指定する任意のフィールド。この例では、first_name 値は Anne Marie です。

3

source

イベントのソースメタデータを記述する必須のフィールド。source フィールド構造には 作成 イベントと同じフィールドがありますが、一部の値が異なります。ソースメタデータには以下が含まれています。

  • Debezium バージョン
  • コネクター型および名前
  • 新しい行が含まれるデータベースおよびテーブル
  • スキーマ名
  • イベントがスナップショットの一部である場合 (update イベントの場合は常にfalse)
  • 操作が実行されたトランザクションの ID
  • データベースログの操作のオフセット
  • データベースに変更が加えられた時点のタイムスタンプ

4

op

操作の型を記述する必須の文字列。更新 イベントの値では、 op フィールドの値は u で、更新によってこの行が変更したことを示します。

5

ts_ms

コネクターがイベントを処理した時間を表示する任意のフィールド。この時間は、Kafka Connect タスクを実行している JVM のシステムクロックを基にします。

source オブジェクトで、ts_ms は変更がデータベースに加えられた時間を示します。payload.source.ts_ms の値を payload.ts_ms の値と比較することにより、ソースデータベースの更新と Debezium との間の遅延を判断できます。

注記

行のプライマリーキー/一意キーの列を更新すると、行のキーの値が変更されます。キーが変更されると、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.7.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": "d", 4
        "ts_ms": 1465581902461 5
    }
}

表7.8 削除 イベント値フィールドの説明

項目フィールド名説明

1

before

イベント発生前の行の状態を指定する任意のフィールド。削除 イベント値の before フィールドには、データベースのコミットで削除される前に行にあった値が含まれます。

この例では、テーブルの REPLICA IDENTITY 設定が DEFAULT であるため、before フィールドにはプライマリーキー列のみが含まれます。

2

after

イベント発生後の行の状態を指定する任意のフィールド。削除 イベント値の after フィールドは null で、行が存在しないことを示します。

3

source

イベントのソースメタデータを記述する必須のフィールド。削除 イベント値の source フィールド構造は、同じテーブルの 作成 および 更新 イベントと同じになります。多くの source フィールドの値も同じです。削除 イベント値では、ts_ms および lsn フィールドの値や、その他の値が変更された可能性があります。ただし、削除 イベント値の source フィールドは、同じメタデータを提供します。

  • Debezium バージョン
  • コネクター型および名前
  • 削除された行が含まれていたデータベースとテーブル
  • スキーマ名
  • イベントがスナップショットの一部であるか (常に 削除 イベントは false)
  • 操作が実行されたトランザクションの ID
  • データベースログの操作のオフセット
  • データベースに変更が加えられた時点のタイムスタンプ

4

op

操作の型を記述する必須の文字列。op フィールドの値は d で、行が削除されたことを示します。

5

ts_ms

コネクターがイベントを処理した時間を表示する任意のフィールド。この時間は、Kafka Connect タスクを実行している JVM のシステムクロックを基にします。

source オブジェクトで、ts_ms は変更がデータベースに加えられた時間を示します。payload.source.ts_ms の値を payload.ts_ms の値と比較することにより、ソースデータベースの更新と Debezium との間の遅延を判断できます。

削除 変更イベントレコードは、この行の削除を処理するために必要な情報を持つコンシューマーを提供します。

警告

プライマリーキーを持たないテーブルに対して生成された 削除 イベントをコンシューマーが処理できるようにするには、テーブルの REPLICA IDENTITYFULL に設定します。テーブルに主キーがなく、テーブルの REPLICA IDENTITYDEFAULT または NOTHING に設定されている場合、削除 イベントの before フィールドはありません。

PostgreSQL コネクターイベントは、Kafka のログコンパクション と動作するように設計されています。ログコンパクションにより、少なくとも各キーの最新のメッセージが保持される限り、一部の古いメッセージを削除できます。これにより、トピックに完全なデータセットが含まれ、キーベースの状態のリロードに使用できるようにするとともに、Kafka がストレージ領域を確保できるようにします。

廃棄 (tombstone) イベント

行が削除された場合でも、Kafka は同じキーを持つ以前のメッセージをすべて削除できるため、削除 イベントの値はログコンパクションで動作します。ただし、Kafka が同じキーを持つすべてのメッセージを削除するには、メッセージの値が null である必要があります。これを可能にするには、PostgreSQL コネクターは、値が null 値以外の同じキーを持つ特別な 廃棄 イベントが含まれる 削除 イベントに従います。

切り捨て (truncate) イベント

切り捨て (truncate) 変更イベントは、テーブルが切り捨てられていることを伝えます。この場合のメッセージキーは null で、メッセージの値は以下のようになります。

{
    "schema": { ... },
    "payload": {
        "source": { 1
            "version": "1.7.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
    }
}

表7.9 切り捨て (truncate) イベント値フィールドの説明

項目フィールド名説明

1

source

イベントのソースメタデータを記述する必須のフィールド。切り捨て (truncate) イベント値の source フィールド構造は、同じテーブルの 作成更新、および 削除 イベントと同じで、以下のメタデータを提供します。

  • Debezium バージョン
  • コネクター型および名前
  • 新しい行が含まれるデータベースおよびテーブル
  • スキーマ名
  • イベントがスナップショットの一部であるか (常に 削除 イベントは false)
  • 操作が実行されたトランザクションの ID
  • データベースログの操作のオフセット
  • データベースに変更が加えられた時点のタイムスタンプ

2

op

操作の型を記述する必須の文字列。op フィールドの値は t で、このテーブルが切り捨てされたことを示します。

3

ts_ms

コネクターがイベントを処理した時間を表示する任意のフィールド。この時間は、Kafka Connect タスクを実行している JVM のシステムクロックを基にします。

source オブジェクトで、ts_ms は変更がデータベースに加えられた時間を示します。payload.source.ts_ms の値を payload.ts_ms の値と比較することにより、ソースデータベースの更新と Debezium との間の遅延を判断できます。

1 つの TRUNCATE ステートメントが複数のテーブルに適用された場合、切り捨てられたテーブルごとに 1 つの切り捨て (truncate) 変更イベントレコードが出力されます。

切り捨て (truncate) イベントは、テーブル全体に加えた変更を表し、メッセージキーを持たないので、単一のパーティションを持つトピックを使用しない限り、テーブルに関する変更イベント (作成更新 など) とそのテーブルの 切り捨て (truncate) イベントの順番は保証されません。たとえば、これらのイベントが異なるパーティションから読み取られる場合、コンシューマーは 更新 イベントを 切り捨て (truncate) イベントの後でのみ受け取る可能性があります。