ホーム
製品
Red Hat build of Debezium
2.3.7
Debezium ユーザーガイド
8.3. Debezium PostgreSQL コネクターのデータ変更イベントの説明

8.3. Debezium PostgreSQL コネクターのデータ変更イベントの説明

Debezium PostgreSQL コネクターは、行レベルの INSERT、UPDATE、および DELETE 操作ごとにデータ変更イベントを生成します。各イベントにはキーと値が含まれます。キーと値の構造は、変更されたテーブルによって異なります。

Debezium および Kafka Connect は、イベントメッセージの継続的なストリーム を中心として設計されています。ただし、これらのイベントの構造は時間の経過とともに変化する可能性があり、コンシューマーによる処理が困難になることがあります。これに対応するために、各イベントにはコンテンツのスキーマが含まれます。スキーマレジストリーを使用している場合は、コンシューマーがレジストリーからスキーマを取得するために使用できるスキーマ ID が含まれます。これにより、各イベントが自己完結型になります。

以下のスケルトン JSON は、変更イベントの基本となる 4 つの部分を示しています。ただし、アプリケーションで使用するために選択した Kafka Connect コンバーターの設定方法によって、変更イベントのこれら 4 部分の表現が決定されます。schema フィールドは、変更イベントが生成されるようにコンバーターを設定した場合のみ変更イベントに含まれます。同様に、イベントキーおよびイベントペイロードは、変更イベントが生成されるようにコンバーターを設定した場合のみ変更イベントに含まれます。JSON コンバーターを使用し、変更イベントの基本となる 4 つの部分すべてを生成するように設定すると、変更イベントの構造は次のようになります。

Copy to Clipboard Toggle word wrap

{
 "schema": { 
   ...
  },
 "payload": { 
   ...
 },
 "schema": { 
   ...
 },
 "payload": { 
   ...
 },
}

{
 "schema": {


   ...
  },
 "payload": {


   ...
 },
 "schema": {


   ...
 },
 "payload": {


   ...
 },
}

表8.7 変更イベントの基本内容の概要
項目	フィールド名	説明
1	`schema`	最初の `schema` フィールドはイベントキーの一部です。イベントキーの `payload` の部分の内容を記述する Kafka Connect スキーマを指定します。つまり、最初の `schema` フィールドは、変更されたテーブルのプライマリーキーの構造、またはテーブルにプライマリーキーがない場合は変更されたテーブルの一意キーの構造を記述します。 `message.key.columns`コネクター設定プロパティーを設定すると、テーブルのプライマリーキーをオーバーライドできます。この場合、最初の schema フィールドはそのプロパティーによって識別されるキーの構造を記述します。
2	`payload`	最初の `payload` フィールドはイベントキーの一部です。前述の `schema` フィールドによって記述された構造を持ち、変更された行のキーが含まれます。
3	`schema`	2 つ目の `schema` フィールドはイベント値の一部です。イベント値の `payload` の部分の内容を記述する Kafka Connect スキーマを指定します。つまり、2 つ目の `schema` は変更された行の構造を記述します。通常、このスキーマには入れ子になったスキーマが含まれます。
4	`payload`	2 つ目の `payload` フィールドはイベント値の一部です。前述の `schema` フィールドによって記述された構造を持ち、変更された行の実際のデータが含まれます。

デフォルトの動作では、コネクターによって、変更イベントレコードがイベントの元のテーブル名前が同じトピックにストリーミングされます。

注記

Kafka 0.10 以降では、任意でイベントキーおよび値を タイムスタンプ とともに記録できます。このタイムスタンプはメッセージが作成された (プロデューサーによって記録) 時間または Kafka によってログに買い込まれた時間を示します。

警告

PosgreSQL コネクターは、すべての Kafka Connect スキーマ名が Avro スキーマ名の形式に準拠するようにします。つまり、論理サーバー名はアルファベットまたはアンダースコア (a-z、A-Z、または _) で始まる必要があります。論理サーバー名の残りの各文字と、スキーマ名とテーブル名の各文字は、アルファベット、数字、またはアンダースコア ( a-z、A-Z、0-9、または _) でなければなりません。無効な文字がある場合は、アンダースコアに置き換えられます。

論理サーバー名、スキーマ名、またはテーブル名に無効な文字が含まれ、名前を区別する唯一の文字が無効であると、無効な文字はすべてアンダースコアに置き換えられるため、予期せぬ競合が発生する可能性があります。

詳細は以下を参照してください。

8.3.1. Debezium PostgreSQL の変更イベントのキー

指定のテーブルでは、変更イベントのキーは、イベントが作成された時点でテーブルのプライマリーキーの各列のフィールドが含まれる構造を持ちます。また、テーブルの REPLICA IDENTITYが FULL または USING INDEX に設定されている場合は、各ユニークキー制約のフィールドがあります。

public データベーススキーマに定義されている customers テーブルと、そのテーブルの変更イベントキーの例を見てみましょう。

テーブルの例

Copy to Clipboard Toggle word wrap

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)
);

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)
);

変更イベントキーの例

topic.prefix コネクター設定プロパティーに PostgreSQL_server の値がある場合、この定義がある限り customers テーブルの変更イベントはすべて同じキー構造を持ち、JSON では以下のようになります。

Copy to Clipboard Toggle word wrap

{
  "schema": { 
    "type": "struct",
    "name": "PostgreSQL_server.public.customers.Key", 
    "optional": false, 
    "fields": [ 
          {
              "name": "id",
              "index": "0",
              "schema": {
                  "type": "INT32",
                  "optional": "false"
              }
          }
      ]
  },
  "payload": { 
      "id": "1"
  },
}

{
  "schema": {


    "type": "struct",
    "name": "PostgreSQL_server.public.customers.Key",


    "optional": false,


    "fields": [


          {
              "name": "id",
              "index": "0",
              "schema": {
                  "type": "INT32",
                  "optional": "false"
              }
          }
      ]
  },
  "payload": {


      "id": "1"
  },
}

表8.8 変更イベントキーの説明
項目	フィールド名	説明
1	`schema`	キーのスキーマ部分は、キーの `payload` 部分の内容を記述する Kafka Connect スキーマを指定します。
2	`PostgreSQL_server.inventory.customers.Key`	キーのペイロードの構造を定義するスキーマの名前。このスキーマは、変更されたテーブルのプライマリーキーの構造を記述します。キースキーマ名の形式は connector-name.database-name.table-name.`Key` です。この例では、以下のようになります。 `PostgreSQL_server` はこのイベントを生成したコネクターの名前です。 `inventory` は変更されたテーブルが含まれるデータベースです。 `customers` は更新されたテーブルです。
3	`任意`	イベントキーの `payload` フィールドに値が含まれる必要があるかどうかを示します。この例では、キーのペイロードに値が必要です。テーブルにプライマリーキーがない場合は、キーの payload フィールドの値は任意です。
4	`fields`	各フィールドの名前、インデックス、およびスキーマなど、`payload` で想定される各フィールドを指定します。
5	`payload`	この変更イベントが生成された行のキーが含まれます。この例では、キーには値が`1`の `id` フィールドが 1 つ含まれます。

注記

column.exclude.list および column.include.list コネクター設定プロパティーは、テーブル列のサブセットのみをキャプチャーできるようにしますが、プライマリーキーまたは一意キーのすべての列は常にイベントのキーに含まれます。

警告

テーブルにプライマリーキーまたは一意キーがない場合は、変更イベントのキーは null になります。プライマリーキーや一意キーの制約がないテーブルの行は一意に識別できません。

8.3.2. Debezium PostgreSQL 変更イベントの値

変更イベントの値はキーよりも若干複雑です。キーと同様に、値には schema セクションと payload セクションがあります。schema セクションには、入れ子のフィールドを含む、 Envelope セクションの payload 構造を記述するスキーマが含まれています。データを作成、更新、または削除する操作のすべての変更イベントには、Envelope 構造を持つ値 payload があります。

変更イベントキーの例を紹介するために使用した、同じサンプルテーブルについて考えてみましょう。

Copy to Clipboard Toggle word wrap

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)
);

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 テーブルにデータを作成する操作に対して、コネクターによって生成される変更イベントの値の部分を示しています。

Copy to Clipboard Toggle word wrap

{
    "schema": { 
        "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", 
                "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", 
                "field": "source"
            },
            {
                "type": "string",
                "optional": false,
                "field": "op"
            },
            {
                "type": "int64",
                "optional": true,
                "field": "ts_ms"
            }
        ],
        "optional": false,
        "name": "PostgreSQL_server.inventory.customers.Envelope" 
    },
    "payload": { 
        "before": null, 
        "after": { 
            "id": 1,
            "first_name": "Anne",
            "last_name": "Kretchmar",
            "email": "annek@noanswer.org"
        },
        "source": { 
            "version": "2.3.7.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", 
        "ts_ms": 1559033904863 
    }
}

{
    "schema": {


        "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",


                "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",


                "field": "source"
            },
            {
                "type": "string",
                "optional": false,
                "field": "op"
            },
            {
                "type": "int64",
                "optional": true,
                "field": "ts_ms"
            }
        ],
        "optional": false,
        "name": "PostgreSQL_server.inventory.customers.Envelope"


    },
    "payload": {


        "before": null,


        "after": {


            "id": 1,
            "first_name": "Anne",
            "last_name": "Kretchmar",
            "email": "annek@noanswer.org"
        },
        "source": {


            "version": "2.3.7.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",


        "ts_ms": 1559033904863

}
}

表8.9 作成イベント値フィールドの説明
項目	フィールド名	説明
1	`schema`	値のペイロードの構造を記述する、値のスキーマ。変更イベントの値スキーマは、コネクターが特定のテーブルに生成するすべての変更イベントで同じになります。
2	`name`	`スキーマ` セクションで、各 `name` フィールドは、値のペイロードのフィールドに対するスキーマを指定します。 `PostgreSQL_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` である場合、この変更イベントは新しい内容に対するものであるため、`before` は `null` になります。注記このフィールドを利用できるかどうかは、各テーブルの `REPLICA IDENTITY` 設定によって異なります。
7	`after`	イベント発生後の行の状態を指定する任意のフィールド。この例では、`after` フィールドには、新しい行の `id`、`first_name`、`last_name`、および `email` 列の値が含まれます。
8	`比較元`	イベントのソースメタデータを記述する必須のフィールド。このフィールドには、イベントの発生元、イベントの発生順序、およびイベントが同じトランザクションの一部であるかどうかなど、このイベントと他のイベントを比較するために使用できる情報が含まれています。ソースメタデータには以下が含まれています。 Debezium バージョンコネクター型および名前新しい行が含まれるデータベースおよびテーブル追加のオフセット情報を文字列化した JSON 配列。最初の値は常に最後にコミットされた LSN で、2 番目の値は常に現在の LSN です。いずれの値も `null` である可能性があります。スキーマ名イベントがスナップショットの一部であるか操作が実行されたトランザクションの ID データベースログの操作のオフセットデータベースに変更が加えられた時点のタイムスタンプ
9	`op`	コネクターによってイベントが生成される原因となった操作の型を記述する必須文字列。この例では、`c` は操作によって行が作成されたことを示しています。有効な値は以下のとおりです。 `c` = create `u` = update `d` = delete `r` = read (読み取り、スナップショットのみに適用) `t` = truncate `m` = message
10	`ts_ms`	コネクターがイベントを処理した時間を表示する任意のフィールド。この時間は、Kafka Connect タスクを実行している JVM のシステムクロックを基にします。 `source` オブジェクトで、`ts_ms` は変更がデータベースに加えられた時間を示します。`payload.source.ts_ms` の値を `payload.ts_ms` の値と比較することにより、ソースデータベースの更新と Debezium との間の遅延を判断できます。

更新イベント

サンプル customers テーブルにある更新の変更イベントの値には、そのテーブルの作成イベントと同じスキーマがあります。同様に、イベント値のペイロードは同じ構造を持ちます。ただし、イベント値ペイロードでは更新イベントに異なる値が含まれます。以下は、コネクターによって customers テーブルでの更新に生成されるイベントの変更イベント値の例になります。

Copy to Clipboard Toggle word wrap

{
    "schema": { ... },
    "payload": {
        "before": { 
            "id": 1
        },
        "after": { 
            "id": 1,
            "first_name": "Anne Marie",
            "last_name": "Kretchmar",
            "email": "annek@noanswer.org"
        },
        "source": { 
            "version": "2.3.7.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", 
        "ts_ms": 1465584025523  
    }
}

{
    "schema": { ... },
    "payload": {
        "before": {


            "id": 1
        },
        "after": {


            "id": 1,
            "first_name": "Anne Marie",
            "last_name": "Kretchmar",
            "email": "annek@noanswer.org"
        },
        "source": {


            "version": "2.3.7.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",


        "ts_ms": 1465584025523

}
}

表8.10 更新イベント値フィールドの説明
項目	フィールド名	説明
1	`before`	データベースをコミットする前に行にあった値が含まれる任意のフィールド。この例では、テーブルの `REPLICA IDENTITY` 設定がデフォルトでは `DEFAULT` であるため、プライマリーキー列 `id` のみが存在します。+ 更新イベントに、行のすべてのコラムの以前の値が含まれるようにするには、`ALTER TABLE customers REPLICA IDENTITY FULL` を実行し、`customers` テーブルを変更する必要があります。
2	`after`	イベント発生後の行の状態を指定する任意のフィールド。この例では、`first_name` 値は `Anne Marie` です。
3	`比較元`	イベントのソースメタデータを記述する必須のフィールド。`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 が含まれます。このヘッダーの値は、更新された行にあった以前の (古い) プライマリーキーです。

delete イベント

削除変更イベントの値は、同じテーブルの作成および更新イベントと同じ schema の部分になります。サンプル customers テーブルの削除イベントの payload 部分は以下のようになります。

Copy to Clipboard Toggle word wrap

{
    "schema": { ... },
    "payload": {
        "before": { 
            "id": 1
        },
        "after": null, 
        "source": { 
            "version": "2.3.7.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", 
        "ts_ms": 1465581902461 
    }
}

{
    "schema": { ... },
    "payload": {
        "before": {


            "id": 1
        },
        "after": null,


        "source": {


            "version": "2.3.7.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",


        "ts_ms": 1465581902461

}
}

表8.11 削除イベント値フィールドの説明
項目	フィールド名	説明
1	`before`	イベント発生前の行の状態を指定する任意のフィールド。delete イベント値の `before` フィールドには、データベースのコミットで削除される前に行にあった値が含まれます。この例では、テーブルの `REPLICA IDENTITY` 設定が `DEFAULT` であるため、`before` フィールドにはプライマリーキー列のみが含まれます。
2	`after`	イベント発生後の行の状態を指定する任意のフィールド。削除イベント値の `after` フィールドは `null` で、行が存在しないことを示します。
3	`比較元`	イベントのソースメタデータを記述する必須のフィールド。削除イベント値の `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 IDENTITY を FULL に設定します。テーブルにプライマリーキーがなく、テーブルの REPLICA IDENTITY が DEFAULT または NOTHING に設定されている場合、削除イベントの before フィールドはありません。

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

廃棄 (tombstone) イベント

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

切り捨て (truncate) イベント

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

Copy to Clipboard Toggle word wrap

{
    "schema": { ... },
    "payload": {
        "source": { 
            "version": "2.3.7.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", 
        "ts_ms": 1559033904961 
    }
}

{
    "schema": { ... },
    "payload": {
        "source": {


            "version": "2.3.7.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",


        "ts_ms": 1559033904961

}
}

表8.12 切り捨て (truncate) イベント値フィールドの説明
項目	フィールド名	説明
1	`比較元`	イベントのソースメタデータを記述する必須のフィールド。切り捨て (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) イベントの後でのみ受け取る可能性があります。

メッセージイベント

このイベントタイプは、Postgres 14+ の pgoutput プラグインでのみサポートされています (Postgres ドキュメント)。

メッセージイベントは、一般的にpg_logical_emit_message 関数を使用して、汎用の論理デコードメッセージが WAL に直接挿入されたことを通知します。メッセージキーは、ここでは prefix という名前の 1 つのフィールドを持つ Struct で、メッセージを挿入する際に指定された接頭辞を持ちます。トランザクションメッセージの場合、メッセージの値は以下のようになります。

Copy to Clipboard Toggle word wrap

{
    "schema": { ... },
    "payload": {
        "source": { 
            "version": "2.3.7.Final",
            "connector": "postgresql",
            "name": "PostgreSQL_server",
            "ts_ms": 1559033904863,
            "snapshot": false,
            "db": "postgres",
            "schema": "",
            "table": "",
            "txId": 556,
            "lsn": 46523128,
            "xmin": null
        },
        "op": "m", 
        "ts_ms": 1559033904961, 
        "message": { 
            "prefix": "foo",
            "content": "Ymfy"
        }
    }
}

{
    "schema": { ... },
    "payload": {
        "source": {


            "version": "2.3.7.Final",
            "connector": "postgresql",
            "name": "PostgreSQL_server",
            "ts_ms": 1559033904863,
            "snapshot": false,
            "db": "postgres",
            "schema": "",
            "table": "",
            "txId": 556,
            "lsn": 46523128,
            "xmin": null
        },
        "op": "m",


        "ts_ms": 1559033904961,


        "message": {


            "prefix": "foo",
            "content": "Ymfy"
        }
    }
}

他のイベントタイプとは異なり、非トランザクションメッセージは、関連する BEGIN や END のトランザクションイベントを持ちません。メッセージの値は、非取引メッセージの場合は以下のようになります。

Copy to Clipboard Toggle word wrap

{
    "schema": { ... },
    "payload": {
        "source": { 
            "version": "2.3.7.Final",
            "connector": "postgresql",
            "name": "PostgreSQL_server",
            "ts_ms": 1559033904863,
            "snapshot": false,
            "db": "postgres",
            "schema": "",
            "table": "",
            "lsn": 46523128,
            "xmin": null
        },
        "op": "m", 
        "ts_ms": 1559033904961 
        "message": { 
            "prefix": "foo",
            "content": "Ymfy"
    }
}

{
    "schema": { ... },
    "payload": {
        "source": {


            "version": "2.3.7.Final",
            "connector": "postgresql",
            "name": "PostgreSQL_server",
            "ts_ms": 1559033904863,
            "snapshot": false,
            "db": "postgres",
            "schema": "",
            "table": "",
            "lsn": 46523128,
            "xmin": null
        },
        "op": "m",


        "ts_ms": 1559033904961


        "message": {


            "prefix": "foo",
            "content": "Ymfy"
    }
}

表8.13 message イベント値フィールドの説明
項目	フィールド名	説明
1	`比較元`	イベントのソースメタデータを記述する必須のフィールド。メッセージイベント値では、`source` フィールド構造は、いかなるメッセージイベントの `table` または `scheme` 情報も持たず、メッセージイベントがトランザクションである場合にのみ、`tx Id` を持つことになります。 Debezium バージョンコネクター型および名前データベース名スキーマ名 (message イベントの場合は常に `""`) テーブル名 (message イベントの場合は常に `""`) イベントがスナップショットの一部である場合 (メッセージイベントの場合は常に `false`) 操作が行われたトランザクションの ID (非トランザクションのメッセージイベントの場合は `null`) データベースログの操作のオフセットトランザクションメッセージ。メッセージが WAL に挿入された時のタイムスタンプトランザクション以外のメッセージ。コネクターがメッセージに遭遇したときのタイムスタンプ
2	`op`	操作の型を記述する必須の文字列。`op` フィールドの値は `m` で、これがメッセージイベントであることを示しています。
3	`ts_ms`	コネクターがイベントを処理した時間を表示する任意のフィールド。この時間は、Kafka Connect タスクを実行している JVM のシステムクロックを基にします。トランザクションメッセージイベントの場合、`source` オブジェクトの `ts_ms` 属性は、トランザクションメッセージイベントにおいて、データベースで変更が行われた時間を示します。`payload.source.ts_ms` の値を `payload.ts_ms` の値と比較することにより、ソースデータベースの更新と Debezium との間の遅延を判断できます。非トランザクションメッセージイベントの場合、`source` オブジェクトの `ts_ms` は、コネクターがメッセージイベントに遭遇した時間を示し、`payload.ts_ms` は、コネクターがイベントを処理した時間を示します。この違いは、Postgres の一般的な論理メッセージ形式にはコミットのタイムスタンプが存在せず、非トランザクションの論理メッセージには (タイムスタンプ情報を持つ) `BEGIN` イベントが先行していないことに起因します。
4	`message`	メッセージのメタデータを格納するフィールド接頭辞 (テキスト) コンテンツ (バイナリープロセスモードの設定に基づいてエンコードされたバイト配列)

トップに戻る

8.3. Debezium PostgreSQL コネクターのデータ変更イベントの説明

8.3.1. Debezium PostgreSQL の変更イベントのキー

8.3.2. Debezium PostgreSQL 変更イベントの値

詳細情報

試用、購入および販売

コミュニティー

Red Hat ドキュメントについて

多様性を受け入れるオープンソースの強化

会社概要

Theme

Red Hat legal and privacy links

Red Hat legal and privacy links