リアルタイム変更トラッキング機能のデータ形式定義と例 - Lindorm

このトピックでは、サブスクライブされたテーブルのデータが挿入、更新、および削除された後にLindormが生成する操作レコードのデータ形式について説明します。また、操作の種類ごとにサンプルデータも提供します。デフォルトのデータ形式はDebezium Format V2.0です。コンシューマークライアントを設定して、データ形式に基づいてデータを使用できます。

データ形式の説明

次のサンプルコードは、データ形式の例を示しています。次の表では、サンプルコードのフィールドについて説明します。

{ 
  "payload": { 
    "op": "u", 
    "ts_ms": 1465491411815, 
    "before": { 
      "id": 1004,
      "name": "Jane"
    },
    "after": { 
      "id": 1004,
      "name": "Anne"
    },
    "source": { 
      "version": "v1.0",
      "db": "ld-xxxx",
      "namespace": "default",
      "table": "customers",
      "ts_ms": 1465491411807
    }
  },
  "schema": { 
  "type": "struct",
  "fields": [
     {
        "type": "string",
        "optional": false,
        "field": "op"
   }, {
        "type": "int64",
        "optional": false,
        "field": "ts_ms"
   }, {
        "type": "struct",
        "fields": [
          {
            "type": "int32",
            "optional": false,
            "field": "id"
     }, {
            "type": "string",
            "optional": false,
            "field": "name"
          }
        ],
        "optional": true,
        "field": "before"
   }, {
        "type": "struct",
        "fields": [
          {
            "type": "int32",
            "optional": false,
            "field": "id"
     }, {
            "type": "string",
            "optional": false,
            "field": "name"
     }
        ],
        "optional": true,
        "field": "after"
   }, {
        "type": "struct",
        "fields": [
          {
            "type": "string",
            "optional": false,
            "field": "version"
         }, {
            "type": "string",
            "optional": false,
            "field": "db"
     }, {
            "type": "string",
            "optional": false,
            "field": "namespace"
     }, {
            "type": "string",
            "optional": false,
            "field": "table"
     }, {
            "type": "int64",
            "optional": false,
            "field": "ts_ms"
     }
        ],
        "optional": false,
        "field": "source"
   }
    ],
  "optional": false
 }
}

フィールド名	説明
payload.op	値 c は、データが挿入されたことを示します。値 u は、データが更新されたことを示します。値 d は、データが削除されたことを示します。値 r は、すべてのデータがエクスポートされたことを示します。この値は使用できません。
payload.ts_ms	データがKafkaに書き込まれたときのUNIXタイムスタンプ。
payload.before	挿入、更新、または削除操作が実行される前に、対応する行に格納されているデータ。
payload.after	挿入、更新、または削除操作が実行された後に、対応する行に格納されている最新のデータ。
payload.source	操作に関する追加情報。 version: Lindormデータベースのバージョン番号。 db: ソース Lindorm インスタンスのID。 namespace: テーブルが属する名前空間。 table: Lindormデータベーステーブルの名前。 ts_ms: Lindormデータベーステーブルのデータが挿入、更新、または削除された時刻。
schema	スキーマフィールドは、使用されるデータのペイロードに基づいて自動的に生成されます。スキーマフィールドの値はJSON形式です。スキーマフィールドは、使用されるデータのフィールド (各フィールドのデータ型を含む) を記述するために使用されます。デフォルトでは、各操作レコードにはスキーマフィールドが含まれます。スキーマフィールドの内容は再帰構造です。 field: フィールドの名前を示します。 type: フィールドのデータ型を示します。 name: スキーマの説明を示します。 fields: 再帰順に格納されているネストされたフィールドを含みます。 optional: フィールドがオプションかどうかを示します。

説明

HBaseテーブルの操作レコードのデータ形式は、SQLテーブルのデータ形式と同じです。ただし、HBaseテーブルとSQLテーブルには次の構造的な違いがあります。

HBaseテーブルは、データベースに生のバイナリデータを格納します。データ変更トラッキング機能を使用して使用されるデータは、Base64でエンコードされた文字列です。
HBaseテーブルは列ファミリーを使用します。したがって、非主キー列の列名は列ファミリー_列名の形式です。主キー列の列名はROWに固定されています。

サンプルデータ

SQL

Lindormデータベースで次のスキーマを使用するテーブルを作成します。

CREATE TABLE customers (id VARCHAR,first_name VARCHAR,last_name VARCHAR, PRIMARY KEY(id));

データ挿入レコードのサンプルデータ

{
  "schema": {}, 
  "payload": { 
    "op": "c", 
    "ts_ms": 1465491411815, 
    "before": null, 
    "after": { 
      "id": "1004",
      "first_name": "Anne",
      "last_name": "Kretchmar"
    },
    "source": { 
      "version": "v1.0",
      "db": "ld-xxxx",
      "namespace": "default",
      "table": "customers",
      "ts_ms": 1465491411807
    }
  }
}

データ更新レコードのサンプルデータ

{
  "schema": {},
  "payload": { 
    "op": "u", 
    "ts_ms": 1465491411815,
    "before": {
      "id": "1004",
      "first_name": "Anne Marie",
      "last_name": "Kretchmar"
    }, 
    "after": {
      "id": "1004",
      "first_name": "Anne",
      "last_name": "Kretchmar"
    },
    "source": {
      "version": "v1.0",
      "db": "ld-xxxx",
      "namespace": "default",
      "table": "customers",
      "ts_ms": 1465491411807
    }
  }
}

行削除レコードのサンプルデータ

{
  "schema": {},
  "payload": { 
    "op": "d", 
    "ts_ms": 1465491411815,
    "before": { 
      "id": "1004",
      "first_name": "Anne Marie",
      "last_name": "Kretchmar"
    }, 
    "after": null,
    "source": {
      "version": "v1.0",
      "db": "ld-xxxx",
      "namespace": "default",
      "table": "customers",
      "ts_ms": 1465491411807
    }
  }
}

列削除レコードのサンプルデータ

{
  "schema": {},
  "payload": { 
    "op": "u", 
    "ts_ms": 1465491411815,
    "before": { 
      "id": "1004",
      "first_name": "Anne Marie",
      "last_name": "Kretchmar"
    }, 
    "after": { 
      "id": "1004",
      "first_name": "Anne Marie"
    }, 
    "source": {
      "version": "v1.0",
      "db": "ld-xxxx",
      "namespace": "default",
      "table": "customers",
      "ts_ms": 1465491411807
    }
  }
}

HBase

データ挿入レコードのサンプルデータ

Put put = new Put(Bytes.toBytes("user1")); // Putオブジェクトを作成し、行キーを"user1"に設定します。
put.addColumn(Bytes.toBytes("f"), Bytes.toBytes("name"), Bytes.toBytes("lucky")); // 列ファミリー"f"、列修飾子"name"、値"lucky"を持つ列を追加します。
table.put(put); // データをテーブルに書き込みます。

データ更新レコードのサンプルデータ

{
    "schema": {},
    "payload": {
        "op": "c",
        "ts_ms": 1725258859839,
        "before": null,
        "after": {
            "ROW": "dXNlcjE=",
            "f_name": "bHVja3k="
        },
        "source": {
            "version": "v2.0",
            "db": "ld-xxxx",
            "namespace": "default",
            "table": "customers",
            "ts_ms": 1725258833727
        }
    }
}