Red Hat Summit Red Hat Summitサポートコンソール開発者トライアルの開始お問い合わせ

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

変更イベントの値はキーよりも若干複雑です。キーと同様に、値には schema セクションと payload セクションがあります。schema セクションには、ネストされたフィールドを含む、payload セクションの Envelope 構造を記述するスキーマが含まれます。データを作成、更新、または削除する操作のすべての変更イベントには、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
);
Copy to Clipboard Toggle word wrap

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

作成イベント

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

{
  "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": "mydatabase.MYSCHEMA.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": "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",
3 

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

  },
  "payload": {
5 

    "before": null,
6 

    "after": {
7 

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

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

    "ts_ms": 1559729471739
10 

  }
}
Copy to Clipboard Toggle word wrap
Expand
表3.4 作成 イベント値フィールドの説明
項目フィールド名説明

1

schema

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

2

name

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

mydatabase.MYSCHEMA.CUSTOMERS.Value はペイロードのbefore および after フィールドのスキーマです。このスキーマは、顧客 テーブルに固有です。コネクターは、MYSCHEMA.CUSTOMERS テーブルのすべての行に対してこのスキーマを使用します。

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

3

name

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

4

name

mydatabase.MYSCHEMA.CUSTOMERS.Envelope は、ペイロードの全体的な構造のスキーマです。mydatabase はデータベース、MYSCHEMA はスキーマ、CUSTOMER S はテーブルです。

5

payload

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

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

6

before

イベント発生前の行の状態を指定する任意のフィールド。この例では、op フィールドが create の c になる場合、この変更イベントは新規コンテンツ用であるため、before フィールドは null になります。

7

after

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

8

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 との間の遅延を判断できます。

更新イベント

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

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

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

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

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

    "ts_ms": 1559729998706
5 

  }
}
Copy to Clipboard Toggle word wrap
Expand
表3.5 更新 イベント値フィールドの説明
項目フィールド名説明

1

before

イベント発生前の行の状態を指定する任意のフィールド。更新 イベント値の before フィールドには、データベースのコミット前に各テーブル列のフィールドと、その列にあった値が含まれます。この例では、EMAIL の値が john.doe@example.com であることに注意してください。

2

after

イベント発生後の行の状態を指定する任意のフィールド。構造 の前後に 比較して、この行への更新内容を判断できます。この例では、EMAIL の値が noreply@example.com になりました。

3

source

イベントのソースメタデータを記述する必須のフィールド。ソースフィールド 構造には create イベントと同じフィールドが含まれますが、一部の値が異なります。たとえば、更新 イベントサンプルの 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": { ... },
  },
  "payload": {
    "before": {
1 

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

    "source": {
3 

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

    "ts_ms": 1559730450205
5 

  }
}
Copy to Clipboard Toggle word wrap
Expand
表3.6 削除 イベント値フィールドの説明
項目フィールド名説明

1

before

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

2

after

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

3

source

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

  • 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)イベントを出力します。

トップに戻る
Red Hat logoGithubredditYoutubeTwitter

詳細情報

試用、購入および販売

コミュニティー

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

Red Hat をお使いのお客様が、信頼できるコンテンツが含まれている製品やサービスを活用することで、イノベーションを行い、目標を達成できるようにします。 最新の更新を見る.

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

Red Hat では、コード、ドキュメント、Web プロパティーにおける配慮に欠ける用語の置き換えに取り組んでいます。このような変更は、段階的に実施される予定です。詳細情報: Red Hat ブログ.

会社概要

Red Hat は、企業がコアとなるデータセンターからネットワークエッジに至るまで、各種プラットフォームや環境全体で作業を簡素化できるように、強化されたソリューションを提供しています。

Theme

© 2025 Red Hat