第6章 Oracle 用 Debezium コネクター (開発者プレビュー)

通常、Oracle サーバーの redo ログは、データベースの完全な履歴を保持しないように設定されています。そのため、Debezium Oracle コネクターはデータベースの全履歴をログから取得できません。コネクターがデータベースの現在の状態のベースラインを確立できるようにするには、コネクターが最初に起動すると、データベースの初期 整合性スナップショット を実行します。

snapshot.mode コネクター設定プロパティーの値を設定することで、コネクターがスナップショットを作成する方法をカスタマイズできます。デフォルトでは、コネクターのスナップショットモードは initial に設定されます。

スナップショットプロセスが開始されたら、コネクターの障害、リバランス、またはその他の理由でプロセスが中断された場合には、コネクターの再起動後にプロセスが再開されます。コネクターによって初期スナップショットが完了した後、ステップ 3 で読み取られた位置からストリーミングが続行され、更新を見逃しません。何らかの理由でコネクターが再び停止した場合、再起動後に、以前停止した場所から変更のストリーミングが再開されます。

Debezium Oracle コネクターは、Debezium Oracle コネクターが 1 つのテーブルのすべての INSERT、UPDATE、および DELETE 操作のイベントを 1 つの Kafka トピックに書き込むことです。Kafka トピックの命名規則は次のとおりです。

serverName.schemaName.tableName

たとえば、Met ment がサーバー名で、インベントリーが スキーマ名で、データベースに名前、顧客 、および製品 が含まれるテーブルが含まれる場合、Debezium Oracle コネクターはイベントを以下の Kafka トピックに出力します。1 つはデータベースのテーブルごとに 1 つずつです。

fulfillment.inventory.orders
fulfillment.inventory.customers
fulfillment.inventory.products

Debezium Oracle コネクターは、スキーマ変更の履歴をデータベース履歴トピックに保存します。このトピックは内部コネクターの状態を反映するため、直接使用しないでください。アプリケーションがスキーマの変更に関する通知を必要とする場合は、パブリックスキーマの変更トピックから情報を取得する必要があります。コネクターは、スキーマ変更イベントを <serverName> という名前の Kafka トピックに書き込みます。serverName は、database.server.name 設定プロパティーに指定されたコネクターの名前になります。

Debezium は、新しいテーブルからデータをストリーミングするたびに、このトピックに新しいメッセージを出力します。

メッセージには、テーブルスキーマの論理表現が含まれます。

{
  "schema": {
  ...
  },
  "payload": {
    "source": {
      "version": "1.5.4.Final",
      "connector": "oracle",
      "name": "server1",
      "ts_ms": 1588252618953,
      "snapshot": "true",
      "db": "ORCLPDB1",
      "schema": "DEBEZIUM",
      "table": "CUSTOMERS",
      "txId" : null,
      "scn" : "1513734",
      "commit_scn": "1513734",
      "lcr_position" : null
    },
    "databaseName": "ORCLPDB1", 1
    "schemaName": "DEBEZIUM", //
    "ddl": "CREATE TABLE \"DEBEZIUM\".\"CUSTOMERS\" \n   (    \"ID\" NUMBER(9,0) NOT NULL ENABLE, \n    \"FIRST_NAME\" VARCHAR2(255), \n    \"LAST_NAME" VARCHAR2(255), \n    \"EMAIL\" VARCHAR2(255), \n     PRIMARY KEY (\"ID\") ENABLE, \n     SUPPLEMENTAL LOG DATA (ALL) COLUMNS\n   ) SEGMENT CREATION IMMEDIATE \n  PCTFREE 10 PCTUSED 40 INITRANS 1 MAXTRANS 255 \n NOCOMPRESS LOGGING\n  STORAGE(INITIAL 65536 NEXT 1048576 MINEXTENTS 1 MAXEXTENTS 2147483645\n  PCTINCREASE 0 FREELISTS 1 FREELIST GROUPS 1\n  BUFFER_POOL DEFAULT FLASH_CACHE DEFAULT CELL_FLASH_CACHE DEFAULT)\n  TABLESPACE \"USERS\" ", 2
    "tableChanges": [ 3
      {
        "type": "CREATE", 4
        "id": "\"ORCLPDB1\".\"DEBEZIUM\".\"CUSTOMERS\"", 5
        "table": { 6
          "defaultCharsetName": null,
          "primaryKeyColumnNames": [ 7
            "ID"
          ],
          "columns": [ 8
            {
              "name": "ID",
              "jdbcType": 2,
              "nativeType": null,
              "typeName": "NUMBER",
              "typeExpression": "NUMBER",
              "charsetName": null,
              "length": 9,
              "scale": 0,
              "position": 1,
              "optional": false,
              "autoIncremented": false,
              "generated": false
            },
            {
              "name": "FIRST_NAME",
              "jdbcType": 12,
              "nativeType": null,
              "typeName": "VARCHAR2",
              "typeExpression": "VARCHAR2",
              "charsetName": null,
              "length": 255,
              "scale": null,
              "position": 2,
              "optional": false,
              "autoIncremented": false,
              "generated": false
            },
            {
              "name": "LAST_NAME",
              "jdbcType": 12,
              "nativeType": null,
              "typeName": "VARCHAR2",
              "typeExpression": "VARCHAR2",
              "charsetName": null,
              "length": 255,
              "scale": null,
              "position": 3,
              "optional": false,
              "autoIncremented": false,
              "generated": false
            },
            {
              "name": "EMAIL",
              "jdbcType": 12,
              "nativeType": null,
              "typeName": "VARCHAR2",
              "typeExpression": "VARCHAR2",
              "charsetName": null,
              "length": 255,
              "scale": null,
              "position": 4,
              "optional": false,
              "autoIncremented": false,
              "generated": false
            }
          ]
        }
      }
    ]
  }
}

コネクターがスキーマ変更トピックに送信するメッセージは、スキーマ変更が含まれるデータベースの名前と同じメッセージキーを使用します。以下の例では、payload フィールドにキーが含まれます。

{
  "schema": {
    "type": "struct",
    "fields": [
      {
        "type": "string",
        "optional": false,
        "field": "databaseName"
      }
    ],
    "optional": false,
    "name": "io.debezium.connector.oracle.SchemaChangeKey"
  },
  "payload": {
    "databaseName": "ORCLPDB1"
  }
}

Debezium は、トランザクションメタデータ境界を表し、データ変更イベントメッセージをエンリッチするイベントを生成できます。

データベーストランザクションは、BEGIN キーワードと END キーワードの間に囲まれているステートメントブロックによって表されます。Debezium は、すべてのトランザクションで BEGIN 区切り文字と END 区切り文字のトランザクション境界イベントを生成します。トランザクション境界イベントには以下のフィールドが含まれます。

以下の例は、典型的なトランザクション境界メッセージを示しています。

{
  "status": "BEGIN",
  "id": "5.6.641",
  "event_count": null,
  "data_collections": null
}

{
  "status": "END",
  "id": "5.6.641",
  "event_count": 2,
  "data_collections": [
    {
      "data_collection": "ORCLPDB1.DEBEZIUM.CUSTOMER",
      "event_count": 1
    },
    {
      "data_collection": "ORCLPDB1.DEBEZIUM.ORDER",
      "event_count": 1
    }
  ]
}

トランザクションイベントは、<database.server.name>.transaction という名前のトピックに書き込まれます。

{
  "before": null,
  "after": {
    "pk": "2",
    "aa": "1"
  },
  "source": {
...
  },
  "op": "c",
  "ts_ms": "1580390884335",
  "transaction": {
    "id": "5.6.641",
    "total_order": "1",
    "data_collection_order": "1"
  }
}