ホーム
製品
Red Hat Integration
2022.Q1
Debezium ユーザーガイド
3.3.2. Debezium Db2 変更イベントの値

3.3.2. Debezium Db2 変更イベントの値

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

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

テーブルの例

CREATE TABLE customers (
 ID INTEGER IDENTITY(1001,1) NOT NULL PRIMARY KEY,
 FIRST_NAME VARCHAR(255) NOT NULL,
 LAST_NAME VARCHAR(255) NOT NULL,
 EMAIL VARCHAR(255) NOT NULL UNIQUE
);

CREATE TABLE customers (
 ID INTEGER IDENTITY(1001,1) NOT NULL PRIMARY KEY,
 FIRST_NAME VARCHAR(255) NOT NULL,
 LAST_NAME VARCHAR(255) NOT NULL,
 EMAIL VARCHAR(255) NOT NULL UNIQUE
);

Copy to Clipboard

Toggle word wrap

customers テーブルのすべての変更イベントのイベント値部分は同じスキーマを指定します。イベント値のペイロードは、イベント型によって異なります。

作成イベント

以下の例は、customers テーブルにデータを作成する操作に対して、コネクターによって生成される変更イベントの値の部分を示しています。

{
  "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": "mydatabase.MYSCHEMA.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": "mydatabase.MYSCHEMA.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": "string",
            "optional": true,
            "field": "change_lsn"
          },
          {
            "type": "string",
            "optional": true,
            "field": "commit_lsn"
          },
        ],
        "optional": false,
        "name": "io.debezium.connector.db2.Source",  
        "field": "source"
      },
      {
        "type": "string",
        "optional": false,
        "field": "op"
      },
      {
        "type": "int64",
        "optional": true,
        "field": "ts_ms"
      }
    ],
    "optional": false,
    "name": "mydatabase.MYSCHEMA.CUSTOMERS.Envelope"  
  },
  "payload": {  
    "before": null,  
    "after": {  
      "ID": 1005,
      "FIRST_NAME": "john",
      "LAST_NAME": "doe",
      "EMAIL": "john.doe@example.org"
    },
    "source": {  
      "version": "1.7.2.Final",
      "connector": "db2",
      "name": "myconnector",
      "ts_ms": 1559729468470,
      "snapshot": false,
      "db": "mydatabase",
      "schema": "MYSCHEMA",
      "table": "CUSTOMERS",
      "change_lsn": "00000027:00000758:0003",
      "commit_lsn": "00000027:00000758:0005",
    },
    "op": "c",  
    "ts_ms": 1559729471739  
  }
}

{
  "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": "mydatabase.MYSCHEMA.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": "mydatabase.MYSCHEMA.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": "string",
            "optional": true,
            "field": "change_lsn"
          },
          {
            "type": "string",
            "optional": true,
            "field": "commit_lsn"
          },
        ],
        "optional": false,
        "name": "io.debezium.connector.db2.Source",


        "field": "source"
      },
      {
        "type": "string",
        "optional": false,
        "field": "op"
      },
      {
        "type": "int64",
        "optional": true,
        "field": "ts_ms"
      }
    ],
    "optional": false,
    "name": "mydatabase.MYSCHEMA.CUSTOMERS.Envelope"


  },
  "payload": {


    "before": null,


    "after": {


      "ID": 1005,
      "FIRST_NAME": "john",
      "LAST_NAME": "doe",
      "EMAIL": "john.doe@example.org"
    },
    "source": {


      "version": "1.7.2.Final",
      "connector": "db2",
      "name": "myconnector",
      "ts_ms": 1559729468470,
      "snapshot": false,
      "db": "mydatabase",
      "schema": "MYSCHEMA",
      "table": "CUSTOMERS",
      "change_lsn": "00000027:00000758:0003",
      "commit_lsn": "00000027:00000758:0005",
    },
    "op": "c",


    "ts_ms": 1559729471739

}
}

Copy to Clipboard

Toggle word wrap

Expand

表3.6 作成イベント値フィールドの説明
項目	フィールド名	説明
1	`schema`	値のペイロードの構造を記述する、値のスキーマ。変更イベントの値スキーマは、コネクターが特定のテーブルに生成するすべての変更イベントで同じになります。
2	`name`	`スキーマ` セクションで、各 `name` フィールドは、値のペイロードのフィールドのスキーマを指定します。 `mydatabase.MYSCHEMA.CUSTOMERS.Value` はペイロードの`before` および `after` フィールドのスキーマです。このスキーマは `customers` テーブルに固有です。コネクターは、`MYSCHEMA.CUSTOMERS` テーブルのすべての行に対してこのスキーマを使用します。 `before` および `after` フィールドのスキーマ名は`logicalName.schemaName.tableName.Value` の形式を取るので、スキーマ名がデータベースで一意になるようにします。つまり、Avro コンバーターを使用する場合、各論理ソースの各テーブルの Avro スキーマには独自の進化と履歴があります。
3	`name`	`io.debezium.connector.db2.Source` は、ペイロードの `source` フィールドのスキーマです。このスキーマは Db2 コネクターに固有です。コネクターは生成するすべてのイベントにこれを使用します。
4	`name`	`mydatabase.MYSCHEMA.CUSTOMERS.Envelope` は、ペイロードの全体的な構造のスキーマです。`mydatabase` はデータベース、`MYSCHEMA` はスキーマ、`CUSTOMERS` はテーブルです。
5	`payload`	値の実際のデータ。これは、変更イベントが提供する情報です。イベントの JSON 表現はそれが記述する行よりもはるかに大きいように見えることがあります。これは、JSON 表現にはメッセージのスキーマ部分とペイロード部分を含める必要があるためです。しかし、Avro コンバーターを使用すると、コネクターが Kafka トピックにストリーミングするメッセージのサイズを大幅に小さくすることができます。
6	`before`	イベント発生前の行の状態を指定する任意のフィールド。この例のように、`op` フィールドが create (作成) の `c` である場合、この変更イベントは新しい内容に対するものであるため、`before` は `null` になります。
7	`after`	イベント発生後の行の状態を指定する任意のフィールド。この例では、`after` フィールドには、新しい行の `ID`、`FIRST_NAME`、`LAST_NAME`、および `EMAIL` 列の値が含まれます。
8	`source`	イベントのソースメタデータを記述する必須のフィールド。`source` 構造には、この変更に関する Db2 の情報が示され、トレーサビリティーが提供されます。また、同じトピックや他のトピックの他のイベントと比較する情報もあり、このイベントが他のイベントの前または後に発生したか、あるいはこのイベントが他のイベントと同じコミットの一部であるかを認識できます。ソースメタデータには以下が含まれています。 Debezium バージョンコネクター型および名前データベースに変更が加えられた時点のタイムスタンプイベントが進行中のスナップショットの一部であるかどうか新しい行が含まれるデータベース、スキーマ、およびテーブルの名前変更 LSN コミット LSN (このイベントがスナップショットの一部である場合は省略)
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": {  
      "ID": 1005,
      "FIRST_NAME": "john",
      "LAST_NAME": "doe",
      "EMAIL": "john.doe@example.org"
    },
    "after": {  
      "ID": 1005,
      "FIRST_NAME": "john",
      "LAST_NAME": "doe",
      "EMAIL": "noreply@example.org"
    },
    "source": {  
      "version": "1.7.2.Final",
      "connector": "db2",
      "name": "myconnector",
      "ts_ms": 1559729995937,
      "snapshot": false,
      "db": "mydatabase",
      "schema": "MYSCHEMA",
      "table": "CUSTOMERS",
      "change_lsn": "00000027:00000ac0:0002",
      "commit_lsn": "00000027:00000ac0:0007",
    },
    "op": "u",  
    "ts_ms": 1559729998706  
  }
}

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


      "ID": 1005,
      "FIRST_NAME": "john",
      "LAST_NAME": "doe",
      "EMAIL": "john.doe@example.org"
    },
    "after": {


      "ID": 1005,
      "FIRST_NAME": "john",
      "LAST_NAME": "doe",
      "EMAIL": "noreply@example.org"
    },
    "source": {


      "version": "1.7.2.Final",
      "connector": "db2",
      "name": "myconnector",
      "ts_ms": 1559729995937,
      "snapshot": false,
      "db": "mydatabase",
      "schema": "MYSCHEMA",
      "table": "CUSTOMERS",
      "change_lsn": "00000027:00000ac0:0002",
      "commit_lsn": "00000027:00000ac0:0007",
    },
    "op": "u",


    "ts_ms": 1559729998706

}
}

Copy to Clipboard

Toggle word wrap

Expand

表3.7 更新イベント値フィールドの説明
項目	フィールド名	説明
1	`before`	イベント発生前の行の状態を指定する任意のフィールド。更新イベント値の `before` フィールドには、各テーブル列のフィールドと、データベースのコミット前にその列にあった値が含まれます。この例では、EMAIL の値が `EMAIL` value is `john.doe@example.com` であることに注意してください。
2	`after`	イベント発生後の行の状態を指定する任意のフィールド。`before` と `after` の構造を比較すると、この行への更新内容を判断できます。この例では、`EMAIL`の値が `noreply@example.com` となっています。
3	`source`	イベントのソースメタデータを記述する必須のフィールド。`source` フィールド構造には作成イベントと同じフィールドが含まれますが、一部の値が異なります。たとえば、更新イベントサンプルの LSN は異なります。この情報を使用して、このイベントを他のイベントと比較し、このイベントが他のイベントの前または後に発生したか、あるいはこのイベントが他のイベントと同じコミットの一部であるかを認識できます。ソースメタデータには以下が含まれています。 Debezium バージョンコネクター型および名前データベースに変更が加えられた時点のタイムスタンプイベントが進行中のスナップショットの一部であるかどうか新しい行が含まれるデータベース、スキーマ、およびテーブルの名前変更 LSN コミット LSN (このイベントがスナップショットの一部である場合は省略)
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)、およびそれに続く行の新しいキーを持つイベントです。

削除イベント

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

{
  "schema": { ... },
  },
  "payload": {
    "before": {  
      "ID": 1005,
      "FIRST_NAME": "john",
      "LAST_NAME": "doe",
      "EMAIL": "noreply@example.org"
    },
    "after": null,  
    "source": {  
      "version": "1.7.2.Final",
      "connector": "db2",
      "name": "myconnector",
      "ts_ms": 1559730445243,
      "snapshot": false,
      "db": "mydatabase",
      "schema": "MYSCHEMA",
      "table": "CUSTOMERS",
      "change_lsn": "00000027:00000db0:0005",
      "commit_lsn": "00000027:00000db0:0007"
    },
    "op": "d",  
    "ts_ms": 1559730450205  
  }
}

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


      "ID": 1005,
      "FIRST_NAME": "john",
      "LAST_NAME": "doe",
      "EMAIL": "noreply@example.org"
    },
    "after": null,


    "source": {


      "version": "1.7.2.Final",
      "connector": "db2",
      "name": "myconnector",
      "ts_ms": 1559730445243,
      "snapshot": false,
      "db": "mydatabase",
      "schema": "MYSCHEMA",
      "table": "CUSTOMERS",
      "change_lsn": "00000027:00000db0:0005",
      "commit_lsn": "00000027:00000db0:0007"
    },
    "op": "d",


    "ts_ms": 1559730450205

}
}

Copy to Clipboard

Toggle word wrap

Expand

表3.8 削除イベント値フィールドの説明
項目	フィールド名	説明
1	`before`	イベント発生前の行の状態を指定する任意のフィールド。削除イベント値の `before` フィールドには、データベースのコミットで削除される前に行にあった値が含まれます。
2	`after`	イベント発生後の行の状態を指定する任意のフィールド。削除イベント値の `after` フィールドは `null` で、行が存在しないことを示します。
3	`source`	イベントのソースメタデータを記述する必須のフィールド。削除イベント値の `source` フィールド構造は、同じテーブルの作成および更新イベントと同じになります。多くの `source` フィールドの値も同じです。削除イベント値では、`ts_ms` および LSN フィールドの値や、その他の値が変更された可能性があります。ただし、削除イベント値の `source` フィールドは、同じメタデータを提供します。 Debezium バージョンコネクター型および名前データベースに変更が加えられた時点のタイムスタンプイベントが進行中のスナップショットの一部であるかどうか新しい行が含まれるデータベース、スキーマ、およびテーブルの名前変更 LSN コミット LSN (このイベントがスナップショットの一部である場合は省略)
4	`op`	操作の型を記述する必須の文字列。`op` フィールドの値は `d` で、行が削除されたことを示します。
5	`ts_ms`	コネクターがイベントを処理した時間を表示する任意のフィールド。この時間は、Kafka Connect タスクを実行している JVM のシステムクロックを基にします。 `source` オブジェクトで、`ts_ms` は変更がデータベースに加えられた時間を示します。`payload.source.ts_ms` の値を `payload.ts_ms` の値と比較することにより、ソースデータベースの更新と Debezium との間の遅延を判断できます。

削除変更イベントレコードは、この行の削除を処理するために必要な情報を持つコンシューマーを提供します。コンシューマーによっては、削除を適切に処理するために古い値が必要になることがあるため、古い値が含まれます。

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

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

トップに戻る

3.3.2. Debezium Db2 変更イベントの値

詳細情報

試用、購入および販売

コミュニティー

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

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

会社概要

Theme

Red Hat legal and privacy links

Red Hat legal and privacy links