第5章 MySQL の Debezium コネクター

MySQL には、データベースにコミットされた順序ですべての操作を記録するバイナリーログ (binlog) があります。これには、テーブルスキーマの変更やテーブルのデータの変更が含まれます。MySQL はレプリケーションとリカバリーに binlog を使用します。

Debezium MySQL コネクターは binlog を読み取り、行レベルの INSERTUPDATE、および DELETE 操作の変更イベントを生成し、変更イベントを Kafka トピックに出力します。クライアントアプリケーションはこれらの Kafka トピックを読み取ります。

MySQL は通常、指定期間後に binlogs をパージするように設定されているため、MySQL コネクターは各データベースの最初の整合性スナップショット を実行します。MySQL コネクターは、スナップショットが作成された時点から binlog を読み取ります。

このコネクターと互換性のある MySQL データベースのバージョンについては、Debezium でサポートされる設定ページを参照してください。

Debezium MySQL コネクターの使用に関する情報および手順は、以下のように整理されています。

5.1. Debezium MySQL コネクターの仕組み

コネクターがサポートする MySQL トポロジーの概要は、アプリケーションを計画するときに役立ちます。Debezium MySQL コネクターを最適に設定および実行するには、コネクターによるテーブルの構造の追跡方法、スキーマ変更の公開方法、スナップショットの実行方法、および Kafka トピック名の決定方法を理解しておくと便利です。

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

5.1.1. Debezium コネクターでサポートされる MySQL トポロジー

Debezium MySQL コネクターは以下の MySQL トポロジーをサポートします。

スタンドアロン
単一の MySQL サーバーを使用する場合は、Debezium MySQL コネクターがサーバーを監視できるように、binlog を有効 (および任意で GTID を有効) にする必要があります。バイナリーログも増分 バックアップ として使用できるため、これは多くの場合で許容されます。この場合、MySQL コネクターは常にこのスタンドアロン MySQL サーバーインスタンスに接続し、それに従います。
プライマリーおよびレプリカ

Debezium MySQL コネクターはプライマリーサーバーまたはレプリカの 1 つ (レプリカの binlog が有効になっている場合) に従うことができますが、コネクターはサーバーが認識できるクラスターのみで変更を確認できます。通常、これはマルチプライマリートポロジー以外では問題ではありません。

コネクターは、サーバーの binlog の位置を記録します。この位置は、クラスターの各サーバーごとに異なります。そのため、コネクターは 1 つの MySQL サーバーインスタンスのみに従う必要があります。このサーバーに障害が発生した場合、サーバーを再起動またはリカバリーしないと、コネクターは継続できません。

高可用性クラスター
MySQL にはさまざまな 高可用性 ソリューションが存在し、問題や障害の耐性をつけ、即座に回復することが大変容易になります。ほとんどの HA MySQL クラスターは GTID を使用します。そのため、レプリカはあらゆるプライマリーサーバーの変更をすべて追跡できます。
マルチプライマリー

ネットワークデータベース (NDB) クラスターのレプリケーション は、複数のプライマリーアーバーからそれぞれをレプリケートする 1 つ以上の MySQL レプリカを使用します。これは、複数の MySQL クラスターのレプリケーションを集約する強力な方法です。このトポロジーには GTID を使用する必要があります。

Debezium MySQL コネクターはこれらのマルチプライマリー MySQL レプリカをソースとして使用することができ、新しいレプリカが古いレプリカに追い付けば、異なるマルチプライマリー MySQL レプリカにフェイルオーバーできます。つまり、新しいレプリカには最初のレプリカで確認されたすべてのトランザクションが含まれます。これは、新しいマルチプライマリー MySQL レプリカへの再接続を試み、binlog で適切な場所を見つけようとする際に、特定の GTID ソースが含まれるまたは除外されるようにコネクターを設定できるため、コネクターがデータベースやテーブルのサブセットのみを使用している場合でも機能します。

ホステッド

Debezium MySQL コネクターが Amazon RDS や Amazon Aurora などのホステッドオプションを使用するためのサポートがあります。

これらのホステッドオプションではグローバル読み取りロックが許可されないため、テーブルレベルロックを使用して 整合性スナップショット を作成します。

5.1.2. Debezium MySQL コネクターによるデータベーススキーマの変更の処理方法

データベースクライアントがデータベースのクエリーを行うと、クライアントはデータベースの現在のスキーマを使用します。しかし、データベーススキーマはいつでも変更が可能です。そのため、挿入、更新、または削除の操作が記録されるたびに、コネクターはどのスキーマであるかを特定できる必要があります。また、テーブルのスキーマが変更される前に記録された、比較的古いイベントをコネクターが処理するので、コネクターは現在のスキーマだけを使用することはできません。

スキーマの変更後に発生する変更を正しく処理するために、MySQL にはデータの行レベルの変更だけでなく、データベースに適用される DDL ステートメントも含まれます。コネクターは binlog を読み取り、DDL ステートメントを見つけると、それらの DDL ステートメントを解析し、各テーブルのスキーマのインメモリー表現を更新します。コネクターはこのスキーマ表現を使用して、挿入、更新、または削除の操作時にテーブルの構造を特定し、適切な変更イベントを生成します。別のデータベース履歴 Kafka トピックでは、コネクターは各 DDL ステートメントがある binlog の場所とともにすべての DDL ステートメントを記録します。

コネクターが正常にクラッシュまたは停止された後にコネクターが再起動されると、コネクターは特定の場所 (特定の時点) から binlog の読み取りを開始します。コネクターは、データベース履歴の Kafka トピックを読み取り、コネクターが起動する binlog の時点まですべての DDL ステートメントを解析することで、この時点で存在したテーブル構造を再ビルドします。

このデータベース履歴トピックはコネクターのみが使用します。コネクターは任意で、コンシューマーアプリケーション向けの異なるトピックへのスキーマ変更イベントの生成を表示でき ます。

MySQL コネクターが、gh-ost または pt-online-schema-change などのスキーマ変更ツールが適用されるテーブルで変更をキャプチャーすると、移行プロセス中にヘルパーテーブルが作成されます。これらのヘルパーテーブルへの変更をキャプチャーするようにコネクターを設定する必要があります。ヘルパーテーブル用に生成されたレコードがコンシューマーに必要ない場合は、メッセージ変換を 1 回適用して、除去できます。

Debezium イベントレコードを受信する トピックのデフォルト名 を参照してください。

5.1.3. Debezium MySQL コネクターによるデータベーススキーマの変更の公開方法

Debezium MySQL コネクターを設定すると、データベースのキャプチャーされたテーブルに適用されるスキーマの変更を記述するスキーマ変更イベントを生成できます。コネクターは、スキーマ変更イベントをすべて <serverName> という名前の Kafka トピックに書き込みます。serverNamedatabase.server.name 設定プロパティーに指定されたコネクターの名前になります。コネクターがスキーマ変更トピックに送信するメッセージには、ペイロードと、任意で変更イベントメッセージのスキーマが含まれます。

スキーマ変更イベントメッセージのペイロードには、以下の要素が含まれます。

ddl
スキーマの変更につながる SQL CREATEALTER、または DROP ステートメントを提供します。
databaseName
DDL ステートメントが適用されるデータベースの名前。databaseName の値は、メッセージキーとして機能します。
pos
ステートメントが表示される binlog の位置。
tableChanges
スキーマの変更後のテーブルスキーマ全体の構造化表現。tableChanges フィールドには、テーブルの各列のエントリーなどのアレイが含まれます。構造化された表現は JSON または Avro 形式でデータを表示するため、コンシューマーは DDL パーサーを介して最初にメッセージを処理しなくてもメッセージを簡単に読み取りできます。
重要

キャプチャーモードであるテーブルでは、コネクターはスキーマ変更トピックにスキーマ変更の履歴だけでなく、内部データベース履歴トピックにも格納します。内部データベース履歴トピックはコネクターのみの使用を対象としており、使用するアプリケーションによる直接使用を目的としていません。スキーマ変更に関する通知が必要なアプリケーションが、スキーマ変更トピックからの情報のみを使用するようにしてください。

重要

データベース履歴トピックをパーティションに分割しないでください。データベース履歴トピックが正しく機能するには、コネクターが出力するイベントレコードの一貫したグローバル順序を維持する必要があります。

トピックがパーティション間で分割されないようにするには、以下のいずれかの方法を使用してトピックのパーティション数を設定します。

  • データベース履歴トピックを手動で作成する場合は、パーティション数を 1 に指定します。
  • Apache Kafka ブローカーを使用してデータベース履歴トピックを自動的に作成する場合に、トピックが作成されるので、Kafka num.partitions 設定オプションの値を 1 に設定します。
警告

コネクターがスキーマ変更トピックに出力するメッセージの形式は、初期の状態であり、通知なしに変更される可能性があります。

例: MySQL コネクタースキーマ変更トピックに出力されるメッセージ

以下の例は、JSON 形式の一般的なスキーマ変更メッセージを示しています。メッセージには、テーブルスキーマの論理表現が含まれます。

{
  "schema": {
  ...
  },
  "payload": {
        "source": {  // (1)
        "version": "1.9.7.Final",
        "connector": "mysql",
        "name": "dbserver1",
        "ts_ms": 0,
        "snapshot": "false",
        "db": "inventory",
        "sequence": null,
        "table": "customers",
        "server_id": 0,
        "gtid": null,
        "file": "mysql-bin.000003",
        "pos": 219,
        "row": 0,
        "thread": null,
        "query": null
    },
    "databaseName": "inventory", // (2)
    "schemaName": null,
    "ddl": "ALTER TABLE customers ADD COLUMN middle_name VARCHAR(2000)", // (3)
    "tableChanges": [ // (4)
        {
        "type": "ALTER", // (5)
        "id": "\"inventory\".\"customers\"",  // (6)
        "table": { // (7)
            "defaultCharsetName": "latin1",
            "primaryKeyColumnNames": [  // (8)
                "id"
            ],
            "columns": [ // (9)
                {
                "name": "id",
                "jdbcType": 4,
                "nativeType": null,
                "typeName": "INT",
                "typeExpression": "INT",
                "charsetName": null,
                "length": 11,
                "scale": null,
                "position": 1,
                "optional": false,
                "autoIncremented": true,
                "generated": true
            },
            {
                "name": "first_name",
                "jdbcType": 12,
                "nativeType": null,
                "typeName": "VARCHAR",
                "typeExpression": "VARCHAR",
                "charsetName": "latin1",
                "length": 255,
                "scale": null,
                "position": 2,
                "optional": false,
                "autoIncremented": false,
                "generated": false
            },                        {
                "name": "last_name",
                "jdbcType": 12,
                "nativeType": null,
                "typeName": "VARCHAR",
                "typeExpression": "VARCHAR",
                "charsetName": "latin1",
                "length": 255,
                "scale": null,
                "position": 3,
                "optional": false,
                "autoIncremented": false,
                "generated": false
            },
            {
                "name": "email",
                "jdbcType": 12,
                "nativeType": null,
                "typeName": "VARCHAR",
                "typeExpression": "VARCHAR",
                "charsetName": "latin1",
                "length": 255,
                "scale": null,
                "position": 4,
                "optional": false,
                "autoIncremented": false,
                "generated": false
            },
            {
                "name": "middle_name",
                "jdbcType": 12,
                "nativeType": null,
                "typeName": "VARCHAR",
                "typeExpression": "VARCHAR",
                "charsetName": "latin1",
                "length": 2000,
                "scale": null,
                "position": 5,
                "optional": true,
                "autoIncremented": false,
                "generated": false
            }
          ]
        }
      }
    ]
  },
  "payload": {
    "databaseName": "inventory",
    "ddl": "CREATE TABLE products ( id INTEGER NOT NULL AUTO_INCREMENT PRIMARY KEY, name VARCHAR(255) NOT NULL, description VARCHAR(512), weight FLOAT ); ALTER TABLE products AUTO_INCREMENT = 101;",
    "source" : {
      "version": "1.9.7.Final",
      "name": "mysql-server-1",
      "server_id": 0,
      "ts_ms": 0,
      "gtid": null,
      "file": "mysql-bin.000003",
      "pos": 154,
      "row": 0,
      "snapshot": true,
      "thread": null,
      "db": null,
      "table": null,
      "query": null
    }
  }
}

表5.1 スキーマ変更トピックに出力されたメッセージのフィールドの説明

項目フィールド名説明

1

source

source フィールドは、コネクターがテーブル固有のトピックに書き込む標準のデータ変更イベントとして設定されます。このフィールドは、異なるトピックでイベントを関連付けるのに役立ちます。

2

databaseName
schemaName

変更が含まれるデータベースとスキーマを識別します。databaseName フィールドの値は、レコードのメッセージキーとして使用されます。

3

ddl

このフィールドには、スキーマの変更を行う DDL が含まれます。ddl フィールドには複数の DDL ステートメントが含まれることがあります。各ステートメントは、databaseName フィールドのデータベースに適用されます。ステートメントは、データベースに適用された順序で示されます。

クライアントは、複数のデータベースに適用される複数の DDL ステートメントを送信できます。MySQL がこれらをアトミックに適用する場合、コネクターは DDL ステートメントを順番に取得し、データベース別にグループ化して、各グループにスキーマ変更イベントを作成します。MySQL がこれらを個別に適用すると、コネクターは各ステートメントに対して個別のスキーマ変更イベントを作成します。

4

tableChanges

DDL コマンドによって生成されるスキーマの変更が含まれる 1 つ以上の項目の配列。

5

type

変更の種類を説明します。値は以下のいずれかになります。

CREATE
テーブルの作成
ALTER
テーブルの変更
DROP
テーブルの削除

6

id

作成、変更、または破棄されたテーブルの完全な識別子。テーブルの名前が変更されると、この識別子は<old>,<new> のテーブル名が連結されます。

7

table

適用された変更後のテーブルメタデータを表します。

8

primaryKeyColumnNames

テーブルのプライマリーキーを設定する列のリスト。

9

変更されたテーブルの各列のメタデータ。

スキーマ履歴トピック も参照してください。

5.1.4. Debezium MySQL コネクターによるデータベーススナップショットの実行方法

Debezium MySQL コネクターが最初に起動すると、データベースの最初の 整合性スナップショット が実行されます。以下のフローは、コネクターによってこのスナップショットが作成される方法を示しています。このフローは、デフォルト initial のスナップショットモード用です。その他のスナップショットモードの詳細は、MySQL コネクター snapshot.mode 設定プロパティー を参照してください。

表5.2 グローバル読み取りロックを使用して最初のスナップショットを実行するためのワークフロー

ステップアクション

1

他のデータベースクライアントによる 書き込み をブロックするグローバル読み取りロックを取得します。

スナップショット自体は、コネクターによる binlog の位置やテーブルスキーマの読み取りを妨害する可能性のある DDL を他のクライアントが適用しないように防ぐことはありません。コネクターは binlog の位置を読み取る間にグローバル読み取りロックを保持し、後のステップで説明するように、ロックを解除します。

2

繰り返し可能な読み取りセマンティクス でトランザクションを開始し、トランザクション内の後続の読み取りがすべて 整合性スナップショット に対して実行されるようにします。

3

現在の binlog の位置を読み取ります。

4

コネクターが変更をキャプチャーするように設定されたデータベースとテーブルのスキーマを読み取ります。

5

グローバル読み取りロックを解放します。他のデータベースクライアントがデータベースに書き込みできるようになりました。

6

該当する場合は、DDL の変更をスキーマ変更トピックに書き込みます。これには、必要な DROP…​ および CREATE…​ DDL ステートメントがすべて含まれます。

7

データベーステーブルをスキャンします。コネクターは、行ごとに、CREATE イベントを関係するテーブル固有の Kafka トピックに出力します。

8

トランザクションをコミットします。

9

コネクターオフセットの完了済みスナップショットを記録します。

コネクターの再起動

最初のスナップショット の実行中にコネクターが失敗または停止したり、再分散された場合、コネクターの再起動後に新しいスナップショットが実行されます。この 最初のスナップショット が完了すると、Debezium MySQL コネクターは binlog の同じ位置から再起動するため、更新が見逃されることはありません。

コネクターが長時間停止した場合、MySQL が古い binlog ファイルをパージし、コネクターの位置が失われる可能性があります。位置が失われた場合、コネクターは 最初のスナップショット を開始位置に戻します。Debezium MySQL コネクターのトラブルシューティングに関する詳細は、behavior when things go wrong を参照してください。

グローバル読み取りロックが許可されない

一部の環境では、グローバル読み取りロックが許可されません。Debezium MySQL コネクターがグローバル読み取りロックが許可されないことを検出すると、代わりにテーブルレベルロックを使用して、この方法でスナップショットを実行します。これには、Debezium コネクターのデータベースユーザーに LOCK TABLES 権限が必要になります。

表5.3 テーブルレベルロックを使用して最初のスナップショットを実行するためのワークフロー

ステップアクション

1

テーブルレベルロックを取得します。

2

繰り返し可能な読み取りセマンティクス でトランザクションを開始し、トランザクション内の後続の読み取りがすべて 整合性スナップショット に対して実行されるようにします。

3

データベースとテーブルの名前を読み取り、選別します。

4

現在の binlog の位置を読み取ります。

5

コネクターが変更をキャプチャーするように設定されたデータベースとテーブルのスキーマを読み取ります。

6

該当する場合は、DDL の変更をスキーマ変更トピックに書き込みます。これには、必要な DROP…​ および CREATE…​ DDL ステートメントがすべて含まれます。

7

データベーステーブルをスキャンします。コネクターは、行ごとに、CREATE イベントを関係するテーブル固有の Kafka トピックに出力します。

8

トランザクションをコミットします。

9

テーブルレベルロックを解除します。

10

コネクターオフセットの完了済みスナップショットを記録します。

5.1.4.1. アドホックスナップショット

デフォルトでは、コネクターは初回スナップショット操作の開始後にのみ実行されます。通常の状況では、この最初のスナップショットが作成されると、コネクターではスナップショットプロセスは繰り返し処理されません。コネクターがキャプチャーする今後の変更イベントデータはストリーミングプロセス経由でのみ行われます。

ただし、場合によっては、最初のスナップショット中にコネクターを取得したデータが古くなったり、失われたり、または不完全となったり可能性があります。テーブルデータを再キャプチャーするメカニズムを提供するため、Debezium にはアドホックスナップショットを実行するオプションがあります。データベースで以下が変更されたことで、アドホックスナップショットが実行される場合があります。

  • コネクター設定は、異なるテーブルセットをキャプチャーするように変更されます。
  • Kafka トピックを削除して、再構築する必要があります。
  • 設定エラーや他の問題が原因で、データの破損が発生します。

アドホックと呼ばれるスナップショット を開始することで、以前にスナップショットをキャプチャーしたテーブルのスナップショットを再実行できます。アドホックスナップショットには、シグナルテーブル を使用する必要があります。シグナルリクエストを Debezium シグナルテーブルに送信して、アドホックスナップショットを開始します。

既存のテーブルのアドホックスナップショットを開始すると、コネクターはテーブルにすでに存在するトピックにコンテンツを追加します。既存のトピックが削除された場合には、トピックの自動作成 が有効になっているのであれば、Debezium は自動的にトピックを作成できます。

アドホックのスナップショットシグナルは、スナップショットに追加するテーブルを指定します。スナップショットは、データベースの内容全体をキャプチャーしたり、データベース内のテーブルのサブセットのみをキャプチャーしたりできます。

execute-snapshot メッセージをシグナルテーブルに送信してキャプチャーするテーブルを指定します。以下の表で説明されているように、run-snapshot シグナルのタイプを incremental に設定し、スナップショットに追加するテーブルの名前を指定します。

表5.4 アドホックの execute-snapshot シグナルレコードの例

フィールドデフォルト

type

incremental

実行するスナップショットのタイプを指定します。
タイプの設定は任意です。現在要求できるのは、incremental スナップショットのみです。

data-collections

該当なし

スナップショットを作成するテーブルの完全修飾名が含まれる配列。
名前の形式は signal.data.collection 設定オプションと同じです。

アドホックスナップショットのトリガー

execute-snapshot シグナルタイプのエントリーをシグナルテーブルに追加して、アドホックスナップショットを開始します。コネクターがメッセージを処理した後に、スナップショット操作を開始します。スナップショットプロセスは、最初と最後のプライマリーキーの値を読み取り、これらの値を各テーブルの開始ポイントおよびエンドポイントとして使用します。テーブルのエントリー数と設定されたチャンクサイズに基づいて、Debezium はテーブルをチャンクに分割し、チャンクごとに 1 度に 1 つずつスナップショットを順番に作成していきます。

現在、execute-snapshot アクションタイプは 増分スナップショット のみをトリガーします。詳細は、スナップショットの増分を参照してください。

5.1.4.2. 増分スナップショット

スナップショットを柔軟に管理するため、Debezium には 増分スナップショット と呼ばれる補助スナップショットメカニズムが含まれています。増分スナップショットは、Debezium コネクターにシグナルを送信するための Debezium メカニズムに依存します。

増分スナップショットでは、最初のスナップショットのように、データベースの完全な状態を一度にすべてキャプチャーする代わりに、一連の設定可能なチャンクで各テーブルを段階的にキャプチャーします。スナップショットがキャプチャーするテーブルと、各チャンクのサイズ を指定できます。チャンクのサイズにより、データベース上の各フェッチ操作中にスナップショットで収集される行数が決まります。増分スナップショットのデフォルトのチャンクサイズは 1 KB です。

増分スナップショットが進むと、Debezium はウォーターマークを使用して進捗を追跡し、キャプチャーする各テーブル行のレコードを管理します。この段階的なアプローチでは、標準の初期スナップショットプロセスと比較して、以下の利点があります。

  • スナップショットが完了するまで、ストリーミングストリーミングを延期する代わりに、ストリームしたデータキャプチャーと並行して増分スナップショットを実行できます。コネクターはスナップショットプロセス全体で変更ログからのほぼリアルタイムイベントをキャプチャーし続け、他の操作はブロックしません。
  • 増分スナップショットの進捗が中断された場合は、データを失うことなく再開できます。プロセスが再開すると、スナップショットは最初からテーブルをキャプチャーするのではなく、停止した時点から開始します。
  • いつでも増分スナップショットを実行し、必要に応じてプロセスを繰り返してデータベースの更新に適合できます。たとえば、コネクター設定を変更してテーブルを table.include.list プロパティーに追加した後にスナップショットを再実行します。

増分スナップショットプロセス

増分スナップショットを実行する場合には、Debezium は各テーブルをプライマリーキー別に分類して、設定されたチャンクサイズ に基づいてテーブルをチャンクに分割します。チャンクごとに作業し、テーブルの行ごとにチャンクでキャプチャーします。キャプチャーする行ごとに、スナップショットは READ イベントを出力します。そのイベントは、対象となるチャンクのスナップショットを開始する時の行の値を表します。

スナップショットの作成が進むにつれ、他のプロセスがデータベースへのアクセスを継続し、テーブルレコードが変更される可能性があります。このような変更を反映させるように、通常通りに INSERTUPDATEDELETE 操作がトランザクションログにコミットされます。同様に、継続中の Debezium ストリーミングプロセスは、これらの変更イベントを検出し、対応する変更イベントレコードを Kafka に出力します。

Debezium を使用してプライマリーキーが同じレコード間での競合を解決する方法

場合によっては、ストリーミングプロセスが出力する UPDATE または DELETE イベントを順番に受信できます。つまり、ストリーミングプロセスは、スナップショットがその行の READ イベントが含まれるチャンクをキャプチャーする前に、テーブルの行を変更するイベントを生成する可能性があります。スナップショットが最終的に対象の行にあった READ イベントを出力すると、その値はすでに置き換えられています。Debezium は、シーケンスが到達する増分スナップショットイベントが正しい論理順序で処理されるように、競合を解決するためにバッファースキームを使用します。スナップショットのイベント間で競合が発生し、ストリームされたイベントが解決されてからでないと、Debezium はイベントのレコードを Kafka に送信しません。

スナップショットウィンドウ

遅れて入ってきた READ イベントと、同じテーブルの行を変更するストリーミングイベント間の競合の解決を容易にするために、Debezium は スナップショットウィンドウ と呼ばれるものを使用します。スナップショットウィンドウは、増分スナップショットが指定のテーブルチャンクのデータをキャプチャーしている途中に、間隔を決定します。チャンクのスナップショットウィンドウを開く前に、Debezium は通常の動作に従い、トランザクションログから直接ターゲットの Kafka トピックにイベントをダウンストリームに出力します。ただし、特定のチャンクのスナップショットが開放された瞬間から終了するまで、Debezium は重複除去のステップを実行して、プライマリーキーが同じイベント間での競合を解決します。

データコレクションごとに、Debezium は 2 種類のイベントを出力し、それらの両方のレコードを単一の宛先 Kafka トピックに保存します。テーブルから直接キャプチャーするスナップショットレコードは、READ 操作として出力されます。その間、ユーザーはデータコレクションのレコードの更新を続け、各コミットを反映するようにトランザクションログが更新されるので、Debezium は変更ごとに UPDATE または DELETE 操作を出力します。

スナップショットウィンドウが開放され、Debezium がスナップショットチャンクの処理を開始すると、スナップショットレコードをメモリーバッファーに提供します。スナップショットウィンドウ中に、バッファー内の READ イベントのプライマリーキーは、受信ストリームイベントのプライマリーキーと比較されます。一致するものが見つからない場合、ストリーミングされたイベントレコードが Kafka に直接送信されます。Debezium が一致を検出すると、バッファーされた READ イベントを破棄し、ストリーミングされたレコードを宛先トピックに書き込みます。これは、ストリーミングされたイベントが静的スナップショットイベントよりも論理的に優先されるためです。チャンクのスナップショットウィンドウが終了すると、バッファーに含まれるのは、関連するトランザクションログイベントが存在しない READ イベントのみです。Debezium は、これらの残りの READ イベントをテーブルの Kafka トピックに出力します。

コネクターは各スナップショットチャンクにプロセスを繰り返します。

増分スナップショットのトリガー

現在、増分スナップショットを開始する唯一の方法は、アドホックスナップショットシグナル をソースデータベースのシグナルテーブルに送信することです。SQL INSERT クエリーとしてテーブルにシグナルを送信します。Debezium がシグナルテーブルの変更を検出すると、シグナルを読み取り、要求されたスナップショット操作を実行します。

送信するクエリーはスナップショットに追加するテーブルを指定し、必要に応じてスナップショット操作の種類を指定します。現在、スナップショット操作で唯一の有効なオプションはデフォルト値の incremental だけです。

スナップショットに追加するテーブルを指定するには、テーブルを一覧表示する data-collectionsアレイを指定します (例:
{"data-collections": ["public.MyFirstTable", "public.MySecondTable"]})。

増分スナップショットシグナルの data-collections アレイにはデフォルト値がありません。data-collections アレイが空である場合には、アクションが不要であり、スナップショットを実行しないことが、Debezium で検出されます。

注記

スナップショットに含めるテーブルの名前に、データベース、スキーマ、またはテーブルの名前にドット (.) が含まれている場合、そのテーブルを data-collections 配列に追加するには、名前の各パートを二重引用符でエスケープする必要があります。

たとえば、以下のようなテーブルを含めるには public スキーマに存在し、その名前が My.Table を持つテーブルを含めるには、次の形式を使用します。"public"."My.Table"

前提条件

  • シグナルが有効になっている

    • シグナルデータコレクションがソースのデータベースに存在し、コネクターはこれをキャプチャーするように設定されています。
    • シグナルデータコレクションは signal.data.collection プロパティーで指定されます。

手順

  1. SQL クエリーを送信し、アドホック増分スナップショット要求をシグナルテーブルに追加します。

    INSERT INTO _<signalTable>_ (id, type, data) VALUES (_'<id>'_, _'<snapshotType>'_, '{"data-collections": ["_<tableName>_","_<tableName>_"],"type":"_<snapshotType>_"}');

    以下に例を示します。

    INSERT INTO myschema.debezium_signal (id, type, data) VALUES('ad-hoc-1', 'execute-snapshot', '{"data-collections": ["schema1.table1", "schema2.table2"],"type":"incremental"}');

    コマンドの idtype、および data パラメーターの値は、シグナルテーブルのフィールド に対応します。

    以下の表では、これらのパラメーターについて説明しています。

    表5.5 シグナルテーブルに増分スナップショットシグナルを送信する SQL コマンドのフィールドの説明

    説明

    myschema.debezium_signal

    ソースデータベースにあるシグナルテーブルの完全修飾名を指定します。

    ad-hoc-1

    id パラメーターは、シグナルリクエストの ID 識別子として割り当てられる任意の文字列を指定します。
    この文字列を使用して、シグナルテーブルのエントリーへのログメッセージを特定します。Debezium はこの文字列を使用しません。代わりに、スナップショット作成中に、Debezium は独自の ID 文字列をウォーターマークシグナルとして生成します。

    execute-snapshot

    type パラメーターを指定し、シグナルがトリガーする操作を指定します。

    data-collections

    スナップショットに含めるテーブル名の配列を指定するシグナルの dataフィールドの必須コンポーネント。
    配列は、signal.data.collection 設定プロパティーにコネクターのシグナルテーブルの名前を指定するときに使用する形式で、完全修飾名別にテーブルを一覧表示します。

    incremental

    実行するスナップショット操作の種類指定するシグナルの data フィールドの任意のtype コンポーネント。
    現在、唯一の有効なオプションはデフォルト値 incremental だけです。
    シグナルテーブルに送信する SQL クエリーでの type 値の指定は任意です。
    値を指定しない場合には、コネクターは増分スナップショットを実行します。

以下の例は、コネクターによってキャプチャーされる増分スナップショットイベントの JSON を示しています。

例: 増分スナップショットイベントメッセージ

{
    "before":null,
    "after": {
        "pk":"1",
        "value":"New data"
    },
    "source": {
        ...
        "snapshot":"incremental" 1
    },
    "op":"r", 2
    "ts_ms":"1620393591654",
    "transaction":null
}

項目フィールド名説明

1

snapshot

実行するスナップショット操作タイプを指定します。
現在、唯一の有効なオプションはデフォルト値 incremental だけです。
シグナルテーブルに送信する SQL クエリーでの type 値の指定は任意です。
値を指定しない場合には、コネクターは増分スナップショットを実行します。

2

op

イベントタイプを指定します。
スナップショットイベントの値は r で、READ 操作を示します。

5.1.5. Debezium MySQL 変更イベントレコードを受信する Kafka トピックのデフォルト名

デフォルトでは、MySQL コネクターは、テーブルで発生するすべての INSERTUPDATEDELETE 操作の変更イベントを、そのテーブルに固有の単一の Apache Kafka トピックに書き込みます。

コネクターは以下の規則を使用して変更イベントトピックに名前を付けます。

serverName.databaseName.tableName

fulfillment はサーバー名、inventory はデータベース名で、データベースに orderscustomers、および productsという名前のテーブルが含まれるとします。Debezium MySQL コネクターは、データベースのテーブルごとに 1 つずつ、3 つの Kafka トピックにイベントを出力します。

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

以下のリストは、デフォルト名のコンポーネントの定義を示しています。

serverName
database.server.name コネクター設定プロパティーで指定したサーバーの論理名です。
schemaName
操作が発生したスキーマの名前。
tableName
操作が発生したテーブルの名前。

コネクターは同様の命名規則を適用して、内部データベース履歴トピック (スキーマ変更トピックトランザクションメタデータトピック) にラベルを付けます。

デフォルトのトピック名が要件を満たさない場合は、カスタムトピック名を設定できます。カスタムトピック名を設定するには、論理トピックルーティング SMT に正規表現を指定します。論理トピックルーティング SMT を使用してトピックの命名をカスタマイズする方法は、トピックルーティング を参照してください。

トランザクションメタデータ

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

Debezium がトランザクションメタデータを受信する場合の制限

Debezium は、コネクターのデプロイ後に発生するトランザクションに対してのみメタデータを登録し、受信します。コネクターをデプロイする前に発生するトランザクションのメタデータは利用できません。

Debezium は、すべてのトランザクションで BEGIN および END 区切り文字のトランザクション境界イベントを生成します。トランザクション境界イベントには以下のフィールドが含まれます。

status
BEGIN または END
id
一意のトランザクション識別子の文字列表現。
event_count (END イベント用)
トランザクションによって出力されるイベントの合計数。
data_collections (END イベント用)
data_collectionevent_count 要素のペアの配列。これは、コネクターがデータコレクションから発信された変更に対して出力するイベントの数を示します。

{
  "status": "BEGIN",
  "id": "0e4d5dcd-a33b-11ea-80f1-02010a22a99e:10",
  "event_count": null,
  "data_collections": null
}

{
  "status": "END",
  "id": "0e4d5dcd-a33b-11ea-80f1-02010a22a99e:10",
  "event_count": 2,
  "data_collections": [
    {
      "data_collection": "s1.a",
      "event_count": 1
    },
    {
      "data_collection": "s2.a",
      "event_count": 1
    }
  ]
}

transaction.topic オプションで上書きされない限り、コネクターはトランザクションイベントを <database.server.name>.transaction トピックに出力します。

変更データイベントのエンリッチメント

トランザクションメタデータを有効にすると、データメッセージ Envelope は新しい transaction フィールドでエンリッチされます。このフィールドは、複合フィールドの形式ですべてのイベントに関する情報を提供します。

  • id - 一意のトランザクション識別子の文字列表現。
  • total_order - トランザクションによって生成されたすべてのイベントを対象とするイベントの絶対位置。
  • data_collection_order - トランザクションによって出力されたすべてのイベントを対象とするイベントのデータコレクションごとの位置。

以下は、メッセージの例になります。

{
  "before": null,
  "after": {
    "pk": "2",
    "aa": "1"
  },
  "source": {
...
  },
  "op": "c",
  "ts_ms": "1580390884335",
  "transaction": {
    "id": "0e4d5dcd-a33b-11ea-80f1-02010a22a99e:10",
    "total_order": "1",
    "data_collection_order": "1"
  }
}

GTID が有効ではないシステムの場合は、binlog のファイル名と binlog の位置の組み合わせを使用してトランザクション識別子が作成されます。たとえば、トランザクション BEGIN イベントに対応する binlog のファイル名と位置が mysql-bin.000002 および 1913 の場合には、Debezium が構築したトランザクション識別子は file=mysql-bin.000002,pos=1913 になります。