Debezium ユーザーガイド
Debezium 1.4 の使用
概要
はじめに
Debezium は、データベースの行レベルの変更をキャプチャーする分散サービスのセットで、アプリケーションがそれらの変更を認識し、応答できるようにします。Debezium は、各データベーステーブルにコミットされたすべての行レベルの変更を記録します。各アプリケーションは、対象のトランザクションログを読み取り、発生した順序ですべての操作を確認します。
本ガイドでは、以下の Debezium コネクターの使用方法について説明します。
多様性を受け入れるオープンソースの強化
Red Hat では、コード、ドキュメント、Web プロパティーにおける配慮に欠ける用語の置き換えに取り組んでいます。まずは、マスター (master)、スレーブ (slave)、ブラックリスト (blacklist)、ホワイトリスト (whitelist) の 4 つの用語の置き換えから始めます。この取り組みは膨大な作業を要するため、今後の複数のリリースで段階的に用語の置き換えを実施して参ります。詳細は、Red Hat CTO である Chris Wright のメッセージ をご覧ください。
第1章 Debezium の概要
Debezium は、データベースの変更をキャプチャーする分散サービスのセットです。アプリケーションはこれらの変更を利用し、応答できます。Debezium は、各データベーステーブルの行レベルの変更を 1 つずつ変更イベントレコードにキャプチャーし、これらのレコードを Kafka トピックにストリーミングします。これらのストリームはアプリケーションによって読み取られ、変更イベントレコードは生成された順に提供されます。
詳細は、以下を参照してください。
1.1. Debezium の機能
Debezium は、Apache Kafka Connect のソースコネクターのセットです。各コネクターは、CDC (Change Data Capture) のデータベースの機能を使用して、異なるデータベースから変更を取り込みます。ログベースの CDC は、ポーリングや二重書き込みなどのその他の方法とは異なり、Debezium によって実装されます。
- すべてのデータ変更がキャプチャーされたことを確認します。
- 頻度の高いポーリングに必要な CPU 使用率の増加を防ぎながら、非常に低遅延な変更イベントを生成します。たとえば、MySQL または PostgreSQL の場合、遅延はミリ秒の範囲内になります。
- Last Updated (最終更新日時) の列など、データモデルへの変更は必要ありません。
- 削除をキャプチャー できます。
- データベースの機能や設定に応じて、トランザクション ID や原因となるクエリーなどの古いレコードの状態や追加のメタデータをキャプチャーできます。
詳細は、ブログの記事 Five Advantages of Log-Based Change Data Capture を参照してください。
Debezium コネクターは、さまざまな関連機能やオプションでデータの変更をキャプチャーします。
- スナップショット: コネクターが起動し、すべてのログが存在していない場合は、任意でデータベースの現在の状態の初期スナップショットを取得できます。通常、これは、データベースが一定期間稼働していて、トランザクションのリカバリーやレプリケーションに不要となったトランザクションログを破棄してしまった場合に該当します。スナップショットを実行するためのモードは複数あります。使用しているコネクターのドキュメントを参照してください。
- フィルター: キャプチャーされたスキーマ、テーブル、およびコラムは include または exclude リストフィルターで設定できます。
- マスク:たとえば、機密データが含まれている場合など、特定の列からの値はマスクできます。
- 監視: ほとんどのコネクターは JMX を使用して監視できます。
使用準備が整った メッセージ変換:
- メッセージルーティング
- コンテンツベースルーティング
- リレーショナルコネクターの新しいレコード状態の抽出
- フィルターリング
- トランザクションアウトボックステーブルからの イベントのルーティング
各コネクターのドキュメントには、コネクター機能と設定オプションの詳細が記載されています。
1.2. Debezium アーキテクチャーの説明
Apache Kafka Connect を使用して Debezium をデプロイします。Kafka Connect は、以下を実装および操作するためのフレームワークおよびランタイムです。
- レコードを Kafka に送信する Debezium などのソースコネクター
- Kafka トピックから他のシステムにレコードを伝播するシンクコネクター
以下の図は、Debezium をベースとした Change Data Capture パイプラインのアーキテクチャーを示しています。
イメージにあるように、MySQL と PostgresSQL の Debezium コネクターは、この 2 種類のデータベースへの変更をキャプチャーするためにデプロイされます。各 Debezium コネクターは、そのソースデータベースへの接続を確立します。
-
MySQL コネクターは、
binlog
へのアクセスにクライアントライブラリーを使用します。 - PostgreSQL コネクターは論理レプリケーションストリームから読み取ります。
Kafka Connect は、Kafka ブローカー以外の別のサービスとして動作します。
デフォルトでは、1 つのデータベースからの変更が、名前がテーブル名に対応する Kafka トピックに書き込まれます。必要な場合は、Debezium の トピックルーティング変換 を設定すると、宛先トピック名を調整できます。たとえば、以下を実行できます。
- テーブルの名前と名前が異なるトピックへレコードをルーティングする。
- 複数テーブルの変更イベントレコードを単一のトピックにストリーミングする。
変更イベントレコードが Apache Kafka に存在する場合、Kafka Connect エコシステムの異なるコネクターは、Elasticsearch、データウェアハウス、分析システムなどのその他のシステムおよびデータベースや、Infinispan などのキャッシュにレコードをストリーミングできます。選択したシンクコネクターによっては、Debezium の 新しいレコード状態抽出 (Record State Extraction) の変換を設定する必要がある場合があります。この Kafka Connect SMT は、Debezium の変更イベントからシンクコネクターに after
構造を伝播します。これは、デフォルトで伝播される詳細な変更イベントレコードの代わりになります。
第2章 MySQL の Debezium コネクター
MySQL には、データベースにコミットされた順序ですべての操作を記録するバイナリーログ (binlog) があります。これには、テーブルスキーマの変更やテーブルのデータの変更が含まれます。MySQL はレプリケーションとリカバリーに binlog を使用します。
Debezium MySQL コネクターは binlog を読み取り、行レベルの INSERT
、UPDATE
、および DELETE
操作の変更イベントを生成し、変更イベントを Kafka トピックに出力します。クライアントアプリケーションはこれらの Kafka トピックを読み取ります。
MySQL は通常、指定期間後に binlogs をパージするように設定されているため、MySQL コネクターは各データベースの最初の整合性スナップショット を実行します。MySQL コネクターは、スナップショットが作成された時点から binlog を読み取ります。
Debezium MySQL コネクターの使用に関する情報および手順は、以下のように整理されています。
2.1. Debezium MySQL コネクターの仕組み
コネクターがサポートする MySQL トポロジーの概要は、アプリケーションを計画するときに役立ちます。Debezium MySQL コネクターを最適に設定および実行するには、コネクターによるテーブルの構造の追跡方法、スキーマ変更の公開方法、スナップショットの実行方法、および Kafka トピック名の決定方法を理解しておくと便利です。
詳細は以下を参照してください。
2.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 などのホステッドオプションを使用するためのサポートがあります。
これらのホステッドオプションではグローバル読み取りロックが許可されないため、テーブルレベルロックを使用して 整合性スナップショット を作成します。
2.1.2. Debezium MySQL コネクターによるデータベーススキーマの変更の処理方法
データベースクライアントがデータベースのクエリーを行うと、クライアントはデータベースの現在のスキーマを使用します。しかし、データベーススキーマはいつでも変更が可能です。そのため、挿入、更新、または削除の操作が記録されるたびに、コネクターはどのスキーマであるかを特定できる必要があります。また、コネクターが比較的古いイベントを処理し、テーブルのスキーマが変更される前に記録された可能性があるため、コネクターは現在のスキーマのみを使用することはできません。
これに対応するために、MySQL の binlog にはデータの行レベルの変更だけでなく、データベースに適用される DDL ステートメントも含まれます。コネクターは binlog を読み取り、DDL ステートメントを見つけると、それらの DDL ステートメントを解析し、各テーブルのスキーマのインメモリー表現を更新します。コネクターはこのスキーマ表現を使用して、挿入、更新、または削除の操作時にテーブルの構造を特定し、適切な変更イベントを生成します。別のデータベース履歴 Kafka トピックでは、コネクターは各 DDL ステートメントがある binlog の場所とともにすべての DDL ステートメントを記録します。
コネクターが正常にクラッシュまたは停止された後にコネクターが再起動されると、コネクターは特定の場所 (特定の時点) から binlog の読み取りを開始します。コネクターは、データベース履歴の Kafka トピックを読み取り、コネクターが起動する binlog の時点まですべての DDL ステートメントを解析することで、この時点で存在したテーブル構造を再ビルドします。
このデータベース履歴トピックはコネクターのみが使用します。コネクターは任意で、コンシューマーアプリケーション向けの異なるトピックへのスキーマ変更イベントの生成を表示でき ます。
MySQL コネクターが、gh-ost
または pt-online-schema-change
などのスキーマ変更ツールが適用されるテーブルで変更をキャプチャーすると、移行プロセス中にヘルパーテーブルが作成されます。これらのヘルパーテーブルへの変更をキャプチャーするようにコネクターを設定する必要があります。ヘルパーテーブル用に生成されたレコードがコンシューマーに必要ない場合は、メッセージ変換を 1 回適用して、除去できます。
Debezium イベントレコードを受信する トピックのデフォルト名 を参照してください。
2.1.3. Debezium MySQL コネクターによるデータベーススキーマの変更の公開方法
Debezium MySQL コネクターを設定すると、MySQL サーバーのデータベースに適用されるすべての DDL ステートメントが含まれるスキーマ変更イベントを生成できます。コネクターは、これらのイベントを serverName という名前の Kafka トピックに出力します。serverName は、database.server.name
コネクター設定プロパティーによって指定されるコネクターの名前になります。
スキーマ変更イベント の使用を選択した場合、スキーマ変更トピックからレコードを使用するようにしてください。データベース履歴トピックはコネクターのみが使用します。
スキーマ変更トピックに出力されたイベントのグローバルな順序は重要です。したがって、データベース履歴のトピックをパーティション化しないでください。つまり、データベース履歴トピックの作成時にパーティション数として 1
を指定する必要があります。自動トピックの作成に依存する場合は、デフォルトのパーティション数を指定する Kafka の num.partitions
設定オプションが 1
に設定されていることを確認します。
コネクターがスキーマ変更トピックに出力する各レコードには、DDL ステートメントの適用時に接続されたデータベースの名前を含むメッセージキーが含まれています。例を以下に示します。
{ "schema": { "type": "struct", "name": "io.debezium.connector.mysql.SchemaChangeKey", "optional": false, "fields": [ { "field": "databaseName", "type": "string", "optional": false } ] }, "payload": { "databaseName": "inventory" } }
スキーマ変更イベントレコードの値には、DDL ステートメント、ステートメントが適用されたデータベースの名前、および binlog におけるステートメントの位置を含む構造が含まれます。以下に例を示します。
{ "schema": { "type": "struct", "name": "io.debezium.connector.mysql.SchemaChangeValue", "optional": false, "fields": [ { "field": "databaseName", "type": "string", "optional": false }, { "field": "ddl", "type": "string", "optional": false }, { "field": "source", "type": "struct", "name": "io.debezium.connector.mysql.Source", "optional": false, "fields": [ { "type": "string", "optional": true, "field": "version" }, { "type": "string", "optional": false, "field": "name" }, { "type": "int64", "optional": false, "field": "server_id" }, { "type": "int64", "optional": false, "field": "ts_ms" }, { "type": "string", "optional": true, "field": "gtid" }, { "type": "string", "optional": false, "field": "file" }, { "type": "int64", "optional": false, "field": "pos" }, { "type": "int32", "optional": false, "field": "row" }, { "type": "boolean", "optional": true, "default": false, "field": "snapshot" }, { "type": "int64", "optional": true, "field": "thread" }, { "type": "string", "optional": true, "field": "db" }, { "type": "string", "optional": true, "field": "table" }, { "type": "string", "optional": true, "field": "query" } ] } ] }, "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.4.2.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 } } }
ddl
フィールドには複数の DDL ステートメントが含まれることがあります。各ステートメントは、databaseName
フィールドのデータベースに適用されます。ステートメントは、データベースに適用された順序で示されます。source
フィールドは、テーブル固有のトピックに書き込まれた標準のデータ変更イベントとして設定されます。このフィールドは、異なるトピックでイベントを関連付けるのに役立ちます。
.... "payload": { "databaseName": "inventory", "ddl": "CREATE TABLE products ( id INTEGER NOT NULL AUTO_INCREMENT PRIMARY KEY,...)", "source" : { ... } } ....
クライアントは、複数のデータベースに適用される複数の DDL ステートメントを送信できます。MySQL がこれらをアトミックに適用する場合、コネクターは DDL ステートメントを順番に取得し、データベース別にグループ化して、各グループにスキーマ変更イベントを作成します。MySQL がこれらを個別に適用すると、コネクターは各ステートメントに対して個別のスキーマ変更イベントを作成します。
スキーマ履歴トピック も参照してください。
2.1.4. Debezium MySQL コネクターによるデータベーススナップショットの実行方法
Debezium MySQL コネクターが最初に起動すると、データベースの最初の 整合性スナップショット が実行されます。以下のフローは、コネクターによってこのスナップショットが作成される方法を示しています。このフローは、デフォルト initial
のスナップショットモード用です。その他のスナップショットモードの詳細は、MySQL コネクター snapshot.mode
設定プロパティー を参照してください。
ステップ | アクション |
---|---|
1 |
他のデータベースクライアントによる 書き込み をブロックするグローバル読み取りロックを取得します。 |
2 | 繰り返し可能な読み取りセマンティクス でトランザクションを開始し、トランザクション内の後続の読み取りがすべて 整合性スナップショット に対して実行されるようにします。 |
3 | 現在の binlog の位置を読み取ります。 |
4 | コネクターが変更をキャプチャーするように設定されたデータベースとテーブルのスキーマを読み取ります。 |
5 | グローバル読み取りロックを解放します。他のデータベースクライアントがデータベースに書き込みできるようになりました。 |
6 |
該当する場合は、DDL の変更をスキーマ変更トピックに書き込みます。これには、必要な |
7 |
データベーステーブルをスキャンします。コネクターは、行ごとに、 |
8 | トランザクションをコミットします。 |
9 | コネクターオフセットの完了済みスナップショットを記録します。 |
- コネクターの再起動
最初のスナップショット の実行中にコネクターが失敗または停止したり、再分散された場合、コネクターの再起動後に新しいスナップショットが実行されます。この 最初のスナップショット が完了すると、Debezium MySQL コネクターは binlog の同じ位置から再起動するため、更新が見逃されることはありません。
コネクターが長時間停止した場合、MySQL が古い binlog ファイルをパージし、コネクターの位置が失われる可能性があります。位置が失われた場合、コネクターは 最初のスナップショット を開始位置に戻します。Debezium MySQL コネクターのトラブルシューティングに関する詳細は、問題が発生したときの挙動 を参照してください。
- グローバル読み取りロックが許可されない
一部の環境では、グローバル読み取りロックが許可されません。Debezium MySQL コネクターがグローバル読み取りロックが許可されないことを検出すると、代わりにテーブルレベルロックを使用して、この方法でスナップショットを実行します。これには、Debezium コネクターのデータベースユーザーに
LOCK TABLES
権限が必要になります。表2.2 テーブルレベルロックを使用して最初のスナップショットを実行するためのワークフロー ステップ アクション 1
テーブルレベルロックを取得します。
2
繰り返し可能な読み取りセマンティクス でトランザクションを開始し、トランザクション内の後続の読み取りがすべて 整合性スナップショット に対して実行されるようにします。
3
データベースとテーブルの名前を読み取り、選別します。
4
現在の binlog の位置を読み取ります。
5
コネクターが変更をキャプチャーするように設定されたデータベースとテーブルのスキーマを読み取ります。
6
該当する場合は、DDL の変更をスキーマ変更トピックに書き込みます。これには、必要な
DROP…
およびCREATE…
DDL ステートメントがすべて含まれます。7
データベーステーブルをスキャンします。コネクターは、行ごとに、
CREATE
イベントを関係するテーブル固有の Kafka トピックに出力します。8
トランザクションをコミットします。
9
テーブルレベルロックを解除します。
10
コネクターオフセットの完了済みスナップショットを記録します。
2.1.5. Debezium MySQL 変更イベントレコードを受信する Kafka トピックのデフォルト名
デフォルトの動作では、Debezium MySQL コネクターは 1 つのテーブルのINSERT
、UPDATE
、および DELETE
操作すべてのイベントを 1 つの Kafka トピックに書き込みます。Kafka トピックの命名規則は次のとおりです。
serverName.databaseName.tableName
fulfillment
はサーバー名、inventory
はデータベース名で、データベースに orders
、customers
、および products
という名前のテーブルが含まれるとします。Debezium MySQL コネクターは、データベースのテーブルごとに 1 つずつ、3 つの Kafka トピックにイベントを出力します。
fulfillment.inventory.orders fulfillment.inventory.customers fulfillment.inventory.products
2.2. Debezium MySQL コネクターのデータ変更イベントの説明
Debezium MySQL コネクターは、行レベルの INSERT
、UPDATE
、および DELETE
操作ごとにデータ変更イベントを生成します。各イベントにはキーと値が含まれます。キーと値の構造は、変更されたテーブルによって異なります。
Debezium および Kafka Connect は、イベントメッセージの継続的なストリーム を中心として設計されています。ただし、これらのイベントの構造は時間の経過とともに変化する可能性があり、コンシューマーによる処理が困難になることがあります。これに対応するために、各イベントにはコンテンツのスキーマが含まれます。スキーマレジストリーを使用している場合は、コンシューマーがレジストリーからスキーマを取得するために使用できるスキーマ ID が含まれます。これにより、各イベントが自己完結型になります。
以下のスケルトン JSON は、変更イベントの基本となる 4 つの部分を示しています。ただし、アプリケーションで使用するために選択した Kafka Connect コンバーターの設定方法によって、変更イベントのこれら 4 部分の表現が決定されます。schema
フィールドは、変更イベントが生成されるようにコンバーターを設定した場合のみ変更イベントに含まれます。同様に、イベントキーおよびイベントペイロードは、変更イベントが生成されるようにコンバーターを設定した場合のみ変更イベントに含まれます。JSON コンバーターを使用し、変更イベントの基本となる 4 つの部分すべてを生成するように設定すると、変更イベントの構造は次のようになります。
{ "schema": { 1 ... }, "payload": { 2 ... }, "schema": { 3 ... }, "payload": { 4 ... }, }
項目 | フィールド名 | 説明 |
---|---|---|
1 |
|
最初の |
2 |
|
最初の |
3 |
|
2 つ目の |
4 |
|
2 つ目の |
デフォルトでは、コネクターによって、変更イベントレコードがイベントの元のテーブルと同じ名前を持つトピックにストリーミングされます。トピック名 を参照してください。
MySQL コネクターは、すべての Kafka Connect スキーマ名が Avro スキーマ名の形式 に準拠するようにします。つまり、論理サーバー名はアルファベットまたはアンダースコア (a-z、A-Z、または _) で始まる必要があります。論理サーバー名の残りの各文字と、データベース名とテーブル名の各文字は、アルファベット、数字、またはアンダースコア ( a-z、A-Z、0-9、または \_) でなければなりません。無効な文字がある場合は、アンダースコアに置き換えられます。
論理サーバー名、データベース名、またはテーブル名に無効な文字が含まれ、名前を区別する唯一の文字が無効であると、無効な文字はすべてアンダースコアに置き換えられるため、予期せぬ競合が発生する可能性があります。
詳細は以下を参照してください。
2.2.1. Debezium MySQL 変更イベントのキー
変更イベントのキーには、変更されたテーブルのキーのスキーマと、変更された行の実際のキーのスキーマが含まれます。スキーマとそれに対応するペイロードの両方には、コネクターによってイベントが作成された時点において、変更されたテーブルの PRIMARY KEY
(または一意の制約) に存在した各列のフィールドが含まれます。
以下の customers
テーブルについて考えてみましょう。この後に、このテーブルの変更イベントキーの例を示します。
CREATE TABLE customers ( id INTEGER NOT NULL AUTO_INCREMENT PRIMARY KEY, first_name VARCHAR(255) NOT NULL, last_name VARCHAR(255) NOT NULL, email VARCHAR(255) NOT NULL UNIQUE KEY ) AUTO_INCREMENT=1001;
customers
テーブルへの変更をキャプチャーする変更イベントのすべてに、イベントキースキーマがあります。customers
テーブルに前述の定義がある限り、customers
テーブルへの変更をキャプチャーする変更イベントのキー構造はすべて以下のようになります。JSON では、以下のようになります。
{ "schema": { 1 "type": "struct", "name": "mysql-server-1.inventory.customers.Key", 2 "optional": false, 3 "fields": [ 4 { "field": "id", "type": "int32", "optional": false } ] }, "payload": { 5 "id": 1001 } }
項目 | フィールド名 | 説明 |
---|---|---|
1 |
|
キーのスキーマ部分は、キーの |
2 |
|
キーのペイロードの構造を定義するスキーマの名前。このスキーマは、変更されたテーブルのプライマリーキーの構造を記述します。キースキーマ名の形式は connector-name.database-name.table-name.
|
3 |
|
イベントキーの |
4 |
|
各フィールドの名前、型、および必要かどうかなど、 |
5 |
|
この変更イベントが生成された行のキーが含まれます。この例では、キーには値が |
2.2.2. Debezium MySQL 変更イベントの値
変更イベントの値はキーよりも若干複雑です。キーと同様に、値には schema
セクションと payload
セクションがあります。schema
セクションには、入れ子のフィールドを含む、 Envelope
セクションの payload
構造を記述するスキーマが含まれています。データを作成、更新、または削除する操作のすべての変更イベントには、Envelope 構造を持つ値 payload があります。
変更イベントキーの例を紹介するために使用した、同じサンプルテーブルについて考えてみましょう。
CREATE TABLE customers ( id INTEGER NOT NULL AUTO_INCREMENT PRIMARY KEY, first_name VARCHAR(255) NOT NULL, last_name VARCHAR(255) NOT NULL, email VARCHAR(255) NOT NULL UNIQUE KEY ) AUTO_INCREMENT=1001;
このテーブルへの変更に対する変更イベントの値部分には以下について記述されています。
作成 イベント
以下の例は、customers
テーブルにデータを作成する操作に対して、コネクターによって生成される変更イベントの値の部分を示しています。
{ "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": "mysql-server-1.inventory.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": "mysql-server-1.inventory.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": true, "field": "table" }, { "type": "int64", "optional": false, "field": "server_id" }, { "type": "string", "optional": true, "field": "gtid" }, { "type": "string", "optional": false, "field": "file" }, { "type": "int64", "optional": false, "field": "pos" }, { "type": "int32", "optional": false, "field": "row" }, { "type": "int64", "optional": true, "field": "thread" }, { "type": "string", "optional": true, "field": "query" } ], "optional": false, "name": "io.debezium.connector.mysql.Source", 3 "field": "source" }, { "type": "string", "optional": false, "field": "op" }, { "type": "int64", "optional": true, "field": "ts_ms" } ], "optional": false, "name": "mysql-server-1.inventory.customers.Envelope" 4 }, "payload": { 5 "op": "c", 6 "ts_ms": 1465491411815, 7 "before": null, 8 "after": { 9 "id": 1004, "first_name": "Anne", "last_name": "Kretchmar", "email": "annek@noanswer.org" }, "source": { 10 "version": "1.4.2.Final", "connector": "mysql", "name": "mysql-server-1", "ts_ms": 0, "snapshot": false, "db": "inventory", "table": "customers", "server_id": 0, "gtid": null, "file": "mysql-bin.000003", "pos": 154, "row": 0, "thread": 7, "query": "INSERT INTO customers (first_name, last_name, email) VALUES ('Anne', 'Kretchmar', 'annek@noanswer.org')" } } }
項目 | フィールド名 | 説明 |
---|---|---|
1 |
| 値のペイロードの構造を記述する、値のスキーマ。変更イベントの値スキーマは、コネクターが特定のテーブルに生成するすべての変更イベントで同じになります。 |
2 |
|
|
3 |
|
|
4 |
|
|
5 |
|
値の実際のデータ。これは、変更イベントが提供する情報です。 |
6 |
|
コネクターによってイベントが生成される原因となった操作の型を記述する必須文字列。この例では、
|
7 |
|
コネクターがイベントを処理した時間を表示する任意のフィールド。この時間は、Kafka Connect タスクを実行している JVM のシステムクロックを基にします。 |
8 |
|
イベント発生前の行の状態を指定する任意のフィールド。この例のように、 |
9 |
|
イベント発生後の行の状態を指定する任意のフィールド。この例では、 |
10 |
| イベントのソースメタデータを記述する必須のフィールド。このフィールドには、イベントの発生元、イベントの発生順序、およびイベントが同じトランザクションの一部であるかどうかなど、このイベントと他のイベントを比較するために使用できる情報が含まれています。ソースメタデータには以下が含まれています。
|
更新イベント
サンプル customers
テーブルにある更新の変更イベントの値には、そのテーブルの 作成 イベントと同じスキーマがあります。同様に、イベント値のペイロードは同じ構造を持ちます。ただし、イベント値ペイロードでは 更新 イベントに異なる値が含まれます。以下は、コネクターによって customers
テーブルでの更新に生成されるイベントの変更イベント値の例になります。
{ "schema": { ... }, "payload": { "before": { 1 "id": 1004, "first_name": "Anne", "last_name": "Kretchmar", "email": "annek@noanswer.org" }, "after": { 2 "id": 1004, "first_name": "Anne Marie", "last_name": "Kretchmar", "email": "annek@noanswer.org" }, "source": { 3 "version": "1.4.2.Final", "name": "mysql-server-1", "connector": "mysql", "name": "mysql-server-1", "ts_ms": 1465581029100, "snapshot": false, "db": "inventory", "table": "customers", "server_id": 223344, "gtid": null, "file": "mysql-bin.000003", "pos": 484, "row": 0, "thread": 7, "query": "UPDATE customers SET first_name='Anne Marie' WHERE id=1004" }, "op": "u", 4 "ts_ms": 1465581029523 5 } }
項目 | フィールド名 | 説明 |
---|---|---|
1 |
|
イベント発生前の行の状態を指定する任意のフィールド。更新 イベント値の |
2 |
|
イベント発生後の行の状態を指定する任意のフィールド。 |
3 |
|
イベントのソースメタデータを記述する必須のフィールド。
|
4 |
|
操作の型を記述する必須の文字列。更新 イベントの値では、 |
5 |
|
コネクターがイベントを処理した時間を表示する任意のフィールド。この時間は、Kafka Connect タスクを実行している JVM のシステムクロックを基にします。 |
行のプライマリーキー/一意キーの列を更新すると、行のキーの値が変更されます。キーが変更されると、3 つのイベントが Debezium によって出力されます。3 つのイベントとは、DELETE
イベント、行の古いキーを持つ 廃棄 (tombstone)、およびそれに続く行の新しいキーを持つイベントです。詳細は次のセクションで説明します。
プライマリーキーの更新
行のプライマリーキーフィールドを変更する UPDATE
操作は、プライマリーキーの変更と呼ばれます。プライマリーキーの変更では、UPDATE
イベントレコードの代わりにコネクターが古いキーの DELETE
イベントレコードと、新しい (更新された) キーの CREATE
イベントレコードを出力します。これらのイベントには通常の構造と内容があり、イベントごとにプライマリーキーの変更に関連するメッセージヘッダーがあります。
-
DELETE
イベントレコードには、メッセージヘッダーとして__debezium.newkey
が含まれます。このヘッダーの値は、更新された行の新しいプライマリーキーです。 -
CREATE
イベントレコードには、メッセージヘッダーとして__debezium.oldkey
が含まれます。このヘッダーの値は、更新された行にあった以前の (古い) プライマリーキーです。
削除 イベント
削除 変更イベントの値は、同じテーブルの 作成 および 更新 イベントと同じ schema
の部分になります。サンプル customers
テーブルの 削除 イベントの payload
部分は以下のようになります。
{ "schema": { ... }, "payload": { "before": { 1 "id": 1004, "first_name": "Anne Marie", "last_name": "Kretchmar", "email": "annek@noanswer.org" }, "after": null, 2 "source": { 3 "version": "1.4.2.Final", "connector": "mysql", "name": "mysql-server-1", "ts_ms": 1465581902300, "snapshot": false, "db": "inventory", "table": "customers", "server_id": 223344, "gtid": null, "file": "mysql-bin.000003", "pos": 805, "row": 0, "thread": 7, "query": "DELETE FROM customers WHERE id=1004" }, "op": "d", 4 "ts_ms": 1465581902461 5 } }
項目 | フィールド名 | 説明 |
---|---|---|
1 |
|
イベント発生前の行の状態を指定する任意のフィールド。削除 イベント値の |
2 |
|
イベント発生後の行の状態を指定する任意のフィールド。削除 イベント値の |
3 |
|
イベントのソースメタデータを記述する必須のフィールド。削除 イベント値の
|
4 |
|
操作の型を記述する必須の文字列。 |
5 |
|
コネクターがイベントを処理した時間を表示する任意のフィールド。この時間は、Kafka Connect タスクを実行している JVM のシステムクロックを基にします。 |
削除 変更イベントレコードは、この行の削除を処理するために必要な情報を持つコンシューマーを提供します。コンシューマーによっては、削除を適切に処理するために古い値が必要になることがあるため、古い値が含まれます。
MySQL コネクターイベントは、Kafka のログコンパクション と動作するように設計されています。ログコンパクションにより、少なくとも各キーの最新のメッセージが保持される限り、一部の古いメッセージを削除できます。これにより、トピックに完全なデータセットが含まれ、キーベースの状態のリロードに使用できるようにするとともに、Kafka がストレージ領域を確保できるようにします。
廃棄 (tombstone) イベント
行が削除された場合でも、Kafka は同じキーを持つ以前のメッセージをすべて削除できるため、削除 イベントの値はログコンパクションで動作します。ただし、Kafka が同じキーを持つすべてのメッセージを削除するには、メッセージの値が null
である必要があります。これを可能にするために、Debezium の MySQL コネクターは 削除 イベントを出力した後に、null
値以外で同じキーを持つ特別な廃棄 (tombstone) イベントを出力します。
2.3. Debezium MySQL コネクターによるデータ型のマッピング方法
Debezium MySQL コネクターは、行が存在するテーブルのように構造化されたイベントで行への変更を表します。イベントには、各列の値のフィールドが含まれます。その列の MySQL データ型は、イベントの値を表す方法を指定します。
文字列を格納する列は、文字セットと照合順序を使用して MySQL に定義されます。MySQL コネクターは、binlog イベントの列値のバイナリー表現を読み取るときに、列の文字セットを使用します。
コネクターは MySQL データ型を リテラル 型および セマンティック 型の両方にマップできます。
- リテラル型: Kafka Connect スキーマタイプを使用して値がどのように表されるか。
- セマンティック型: Kafka Connect スキーマがどのようにフィールド (スキーマ名) の意味をキャプチャーするか。
詳細は以下を参照してください。
基本型
以下の表は、コネクターによる基本的な MySQL データ型のマッピング方法を示しています。
MySQL 型 | リテラル型 | セマンティック型 |
---|---|---|
|
| 該当なし |
|
| 該当なし |
|
|
|
|
| 該当なし |
|
| 該当なし |
|
| 該当なし |
|
| 該当なし |
|
| 該当なし |
|
| 該当なし |
|
| 該当なし |
|
| 該当なし |
|
| 該当なし |
|
| 該当なし |
|
|
該当なし |
|
|
該当なし |
|
|
該当なし |
|
| 該当なし |
|
|
該当なし |
|
| 該当なし |
|
|
該当なし |
|
| 該当なし |
|
|
該当なし |
|
| 該当なし |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
時間型
TIMESTAMP
データ型を除き、MySQL の時間型は time.precision.mode
コネクター設定プロパティーの値によって異なります。デフォルト値が CURRENT_TIMESTAMP
または NOW
として指定される TIMESTAMP
列では、Kafka Connect スキーマのデフォルト値として値 1970-01-01 00:00:00
が使用されます。
MySQL では、DATE, `DATETIME
、および TIMESTAMP
列のゼロ値を許可していますが、これはゼロ値が null 値よりも優先されることがあるからです。MySQL コネクターは、列定義で null 値が許可される場合はゼロの値を null 値として表し、列で null 値が許可されない場合はエポック日として表します。
タイムゾーンのない時間型
DATETIME
型は、2018-01-13 09:48:27 のようにローカルの日時を表します。タイムゾーンの情報は含まれません。このような列は、UTC を使用して列の精度に基づいてエポックミリ秒またはマイクロ秒に変換されます。TIMESTAMP
型は、タイムゾーン情報のないタイムスタンプを表します。これは、書き込み時に MySQL によってサーバー (またはセッション) の現在のタイムゾーンから UTC に変換され、値を読み戻すときに UTC からサーバー (またはセッション) の現在のタイムゾーンに変換されます。以下に例を示します。
-
値が
2018-06-20 06:37:03
のDATETIME
は、1529476623000
になります。 -
値が
2018-06-20 06:37:03
のTIMESTAMP
は2018-06-20T13:37:03Z
になります。
このような列は、サーバー (またはセッション) の現在のタイムゾーンに基づいて、UTC の同等の io.debezium.time.ZonedTimestamp
に変換されます。タイムゾーンは、デフォルトでサーバーからクエリーされます。これに失敗した場合は、データベース serverTimezone
MySQL 設定オプションで明示的に指定される必要があります。たとえば、データベースのタイムゾーン (グローバルなタイムゾーンまたは serverTimezone
オプションを使用してコネクターのために設定) が America/Los_Angeles である場合、値 2018-06-20T13:37:03Z を持つ ZonedTimestamp
によって TIMESTAMP 値の 2018-06-20 06:37:03 が表されます。
Kafka Connect および Debezium を実行している JVM のタイムゾーンは、これらの変換には影響しません。
時間値に関連するプロパティーの詳細は、MySQL コネクター設定プロパティー のドキュメントを参照してください。
- time.precision.mode=adaptive_time_microseconds(default)
MySQL コネクターは、イベントがデータベースの値を正確に表すようにするため、列のデータ型定義に基づいてリテラル型とセマンテック型を判断します。すべての時間フィールドはマイクロ秒単位です。正しくキャプチャーされる
TIME
フィールドの値は、範囲が00:00:00.000000
から23:59:59.999999
までの正の値です。表2.9 time.precision.mode=adaptive_time_microseconds の場合のマッピング MySQL 型 リテラル型 セマンティック型 DATE
INT32
io.debezium.time.Date
エポックからの日数を表します。TIME[(M)]
INT64
io.debezium.time.MicroTime
時間の値をマイクロ秒単位で表し、タイムゾーン情報は含まれません。MySQL では、M
を0-6
の範囲にすることができます。DATETIME, DATETIME(0), DATETIME(1), DATETIME(2), DATETIME(3)
INT64
io.debezium.time.Timestamp
エポックからの経過時間をミリ秒で表し、タイムゾーン情報は含まれません。DATETIME(4), DATETIME(5), DATETIME(6)
INT64
io.debezium.time.MicroTimestamp
エポックからの経過時間をマイクロ秒で表し、タイムゾーン情報は含まれません。- time.precision.mode=connect
MySQL コネクターは定義された Kafka Connect の論理型を使用します。この方法はデフォルトの方法よりも精度が低く、データベース列に
3
を超える 少数秒の精度値がある場合は、イベントの精度が低くなる可能性があります。00:00:00.000
から23:59:59.999
までの値のみを処理できます。テーブルのtime.precision.mode=connect
の値が、必ずサポートされる範囲内になるようにすることができる場合のみ、TIME
を設定します。connect
設定は、今後の Debezium バージョンで削除される予定です。表2.10 time.precision.mode=connect の場合のマッピング MySQL 型 リテラル型 セマンティック型 DATE
INT32
org.apache.kafka.connect.data.Date
エポックからの日数を表します。TIME[(M)]
INT64
org.apache.kafka.connect.data.Time
午前 0 時以降の時間値をマイクロ秒で表し、タイムゾーン情報は含まれません。DATETIME[(M)]
INT64
org.apache.kafka.connect.data.Timestamp
エポックからの経過時間をミリ秒で表し、タイムゾーン情報は含まれません。
10 進数型
Debezium コネクターは、decimal.handling.mode
コネクター設定プロパティー の設定にしたがって 10 進数を処理します。
- decimal.handling.mode=precise
表2.11 decimal.handing.mode=precise の場合のマッピング MySQL 型 リテラル型 セマンティック型 NUMERIC[(M[,D])]
BYTES
org.apache.kafka.connect.data.Decimal
scale
スキーマパラメーターには、小数点を移動した桁数を表す整数が含まれます。DECIMAL[(M[,D])]
BYTES
org.apache.kafka.connect.data.Decimal
scale
スキーマパラメーターには、小数点を移動した桁数を表す整数が含まれます。- decimal.handling.mode=double
表2.12 decimal.handing.mode=double の場合のマッピング MySQL 型 リテラル型 セマンティック型 NUMERIC[(M[,D])]
FLOAT64
該当なし
DECIMAL[(M[,D])]
FLOAT64
該当なし
- decimal.handling.mode=string
表2.13 decimal.handing.mode=string の場合のマッピング MySQL 型 リテラル型 セマンティック型 NUMERIC[(M[,D])]
STRING
該当なし
DECIMAL[(M[,D])]
STRING
該当なし
ブール値
MySQL は、特定の方法で BOOLEAN
の値を内部で処理します。BOOLEAN
列は、内部で TINYINT(1)
データ型にマッピングされます。ストリーミング中にテーブルが作成されると、Debezium は元の DDL を受信するため、適切な BOOLEAN
マッピングが使用されます。スナップショットの作成中、Debezium は SHOW CREATE TABLE
を実行して、BOOLEAN
と TINYINT(1)
の両方のカラムに TINYINT(1)
を返すテーブル定義を取得します。その後、Debezium は元の型のマッピングを取得する方法はないため、TINYINT(1)
にマッピングします。
以下は ConfigMap の例になります。
converters=boolean boolean.type=io.debezium.connector.mysql.converters.TinyIntOneToBooleanConverter boolean.selector=db1.table1.*, db1.table2.column1
空間型
現在、Debezium MySQL コネクターは以下の空間データ型をサポートしています。
MySQL 型 | リテラル型 | セマンティック型 |
---|---|---|
|
|
|
2.4. Debezium コネクターを実行するための MySQL の設定
Debezium をインストールおよび実行する前に、一部の MySQL 設定タスクが必要になります。
詳細は以下を参照してください。
2.4.1. Debezium コネクターの MySQL ユーザーの作成
Debezium MySQL コネクターには MySQL ユーザーアカウントが必要です。この MySQL ユーザーは、Debezium MySQL コネクターが変更をキャプチャーするすべてのデータベースに対して適切なパーミッションを持っている必要があります。
前提条件
- MySQL サーバー。
- SQL コマンドの基本知識。
手順
MySQL ユーザーを作成します。
mysql> CREATE USER 'user'@'localhost' IDENTIFIED BY 'password';
必要なパーミッションをユーザーに付与します。
mysql> GRANT SELECT, RELOAD, SHOW DATABASES, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'user' IDENTIFIED BY 'password';
以下の表はパーミッションについて説明しています。
重要グローバル読み取りロックを許可しない Amazon RDS や Amazon Aurora などのホストオプションを使用している場合、テーブルレベルのロックを使用して 整合性スナップショット を作成します。この場合、作成するユーザーに
LOCK TABLES
パーミッションも付与する必要があります。詳細は、スナップショット を参照してください。ユーザーのパーミッションの最終処理を行います。
mysql> FLUSH PRIVILEGES;
キーワード | 説明 |
---|---|
| コネクターがデータベースのテーブルから行を選択できるようにします。これは、スナップショットを実行する場合にのみ使用されます。 |
|
内部キャッシュのクリアまたはリロード、テーブルのフラッシュ、またはロックの取得を行う |
|
|
| コネクターが MySQL サーバーの binlog に接続し、読み取りできるようにします。 |
| コネクターが以下のステートメントを使用できるようにします。
これは必ずコネクターに必要です。 |
| パーミッションが適用されるデータベースを指定します。 |
| パーミッションを付与するユーザーを指定します。 |
| ユーザーの MySQL パスワードを指定します。 |
2.4.2. Debezium の MySQL binlog の有効化
MySQL レプリケーションのバイナリーロギングを有効にする必要があります。バイナリーログは、変更を伝播するためにレプリケーションツールのトランザクション更新を記録します。
前提条件
- MySQL サーバー。
- 適切な MySQL ユーザーの権限。
手順
log-bin
オプションがすでにオンになっているかどうかを確認します。mysql> SELECT variable_value as "BINARY LOGGING STATUS (log-bin) ::" FROM information_schema.global_variables WHERE variable_name='log_bin';
OFF
の場合は、以下に説明するプロパティーで MySQL サーバー設定ファイルを設定します。server-id = 223344 log_bin = mysql-bin binlog_format = ROW binlog_row_image = FULL expire_logs_days = 10
再度 binlog の状態をチェックして、変更を確認します。
mysql> SELECT variable_value as "BINARY LOGGING STATUS (log-bin) ::" FROM information_schema.global_variables WHERE variable_name='log_bin';
プロパティー | 説明 |
---|---|
|
|
|
|
|
|
|
|
|
これは、binlog ファイルが自動的に削除される日数です。デフォルトは |
2.4.3. Debezium の MySQL グローバルトランザクション識別子の有効化
グローバルトランザクション識別子 (GTID) は、クラスター内のサーバーで発生するトランザクションを一意に識別します。Debezium MySQL コネクターには必要ありませんが、GTID を使用すると、レプリケーションを単純化し、プライマリーサーバーとレプリカサーバーの一貫性が保たれるかどうかを簡単に確認することができます。
GTID は MySQL 5.6.5 以降で利用できます。詳細は MySQL のドキュメント を参照してください。
前提条件
- MySQL サーバー。
- SQL コマンドの基本知識。
- MySQL 設定ファイルへのアクセス。
手順
gtid_mode
を有効にします。mysql> gtid_mode=ON
enforce_gtid_consistency
を有効にします。mysql> enforce_gtid_consistency=ON
変更を確認します。
mysql> show global variables like '%GTID%';
結果
+--------------------------+-------+ | Variable_name | Value | +--------------------------+-------+ | enforce_gtid_consistency | ON | | gtid_mode | ON | +--------------------------+-------+
オプション | 説明 |
---|---|
| MySQL サーバーの GTID モードが有効かどうかを指定するブール値。
|
| トランザクションに安全な方法でログに記録できるステートメントの実行を許可することにより、サーバーが GTID の整合性を強制するかどうかを指定するブール値。GTID を使用する場合に必須です。
|
2.4.4. Debezium の MySQL セッションタイムアウトの設定
大規模なデータベースに対して最初の整合性スナップショットが作成されると、テーブルの読み込み時に、確立された接続がタイムアウトする可能性があります。MySQL 設定ファイルで interactive_timeout
と wait_timeout
を設定すると、この動作の発生を防ぐことができます。
前提条件
- MySQL サーバー。
- SQL コマンドの基本知識。
- MySQL 設定ファイルへのアクセス。
手順
interactive_timeout
を設定します。mysql> interactive_timeout=<duration-in-seconds>
wait_timeout
を設定します。mysql> wait_timeout=<duration-in-seconds>
オプション | 説明 |
---|---|
| サーバーが対話的な接続を閉じる前にアクティビティーの発生を待つ時間 (秒単位)。詳細は MySQL のドキュメント を参照してください。 |
| サーバーが非対話的な接続を閉じる前にアクティビティーの発生を待つ時間 (秒単位)。詳細は MySQL のドキュメント を参照してください。 |
2.4.5. Debezium MySQL コネクターのクエリーログイベントの有効化
各 binlog イベントの元の SQL
ステートメントを確認したい場合があります。MySQL 設定ファイルで binlog_rows_query_log_events
オプションを有効にすると、これを行うことができます。
このオプションは、MySQL 5.6 以降で利用できます。
前提条件
- MySQL サーバー。
- SQL コマンドの基本知識。
- MySQL 設定ファイルへのアクセス。
手順
binlog_rows_query_log_events
を有効にします。mysql> binlog_rows_query_log_events=ON
binlog_rows_query_log_events
は、binlog エントリーにSQL
ステートメントが含まれるようにするためのサポートを有効または無効にする値に設定されます。-
ON
= 有効化 -
OFF
= 無効化
-
2.5. Debezium MySQL コネクターのデプロイメント
Debezium MySQL コネクターをデプロイするには、コネクターファイルを Kafka Connect に追加し、コネクターを実行するカスタムコンテナーを作成してから、コネクター設定をコンテナーに追加します。Debezium MySQL コネクターのデプロイに関する詳細は、以下を参照してください。
2.5.1. Debezium MySQL コネクターのデプロイ
Debezium MySQL コネクターをデプロイするには、Debezium コネクターアーカイブが含まれるカスタム Kafka Connect コンテナーイメージをビルドし、このコンテナーイメージをコンテナーレジストリーにプッシュする必要があります。次に、以下のカスタムリソース (CR) を作成する必要があります。
-
Kafka Connect インスタンスを定義する
KafkaConnect
CR。image
は Debezium コネクターを実行するために作成したイメージの名前を指定します。この CR を、Red Hat AMQ Streams がデプロイされている OpenShift インスタンスに適用します。AMQ Streams は、Apache Kafka を OpenShift に取り入れる operator およびイメージを提供します。 -
Debezium MySQL コネクターを定義する
KafkaConnector
CR。この CR をKafkaConnect
CR を適用するのと同じ OpenShift インスタンスに適用します。
前提条件
- MySQL が稼働し、Debezium コネクターと連携するように MySQL を設定する手順 が完了済みである必要があります。
- AMQ Streams が OpenShift にデプロイされ、Apache Kafka および Kafka Connect を実行している。詳細は、Installing Debezium on OpenShift を参照してください。
- Podman または Docker がインストールされている。
-
Debezium コネクターを実行するコンテナーを追加する予定のコンテナーレジストリー (
quay.io
やdocker.io
など) でコンテナーを作成および管理するアカウントとパーミッションを持っている。
手順
Kafka Connect の Debezium MySQL コンテナーを作成します。
- Debezium MySQL コネクターアーカイブをダウンロードします。
Debezium MySQL コネクターアーカイブを展開して、コネクタープラグインのディレクトリー構造を作成します。以下に例を示します。
./my-plugins/ ├── debezium-connector-mysql │ ├── ...
registry.redhat.io/amq7/amq-streams-kafka-26-rhel7:1.6.0
をベースイメージとして使用する Docker ファイルを作成します。たとえば、ターミナルウィンドウから以下のコマンドを入力します。my-plugins
はプラグインディレクトリーの名前に置き換えます。cat <<EOF >debezium-container-for-mysql.yaml 1 FROM registry.redhat.io/amq7/amq-streams-kafka-26-rhel7:1.6.0 USER root:root COPY ./<my-plugins>/ /opt/kafka/plugins/ 2 USER 1001 EOF
このコマンドは、現在のディレクトリーに
debezium-container-for-mysql.yaml
という名前の Docker ファイルを作成します。前のステップで作成した
debezium-container-for-mysql.yaml
Docker ファイルからコンテナーイメージをビルドします。ファイルが含まれるディレクトリーから、ターミナルウィンドウを開き、以下のコマンドのいずれかを入力します。podman build -t debezium-container-for-mysql:latest .
docker build -t debezium-container-for-mysql:latest .
上記のコマンドは、
debezium-container-for-mysql
という名前のコンテナーイメージを構築します。カスタムイメージを
quay.io
などのコンテナーレジストリーまたは内部のコンテナーレジストリーにプッシュします。コンテナーレジストリーは、イメージをデプロイする OpenShift インスタンスで利用できる必要があります。以下のいずれかのコマンドを実行します。podman push <myregistry.io>/debezium-container-for-mysql:latest
docker push <myregistry.io>/debezium-container-for-mysql:latest
新しい Debezium MySQL
KafkaConnect
カスタムリソース (CR) を作成します。たとえば、以下の例のようにannotations
およびimage
プロパティーを指定するdbz-connect.yaml
という名前のKafkaConnect
CR を作成します。apiVersion: kafka.strimzi.io/v1beta1 kind: KafkaConnect metadata: name: my-connect-cluster annotations: strimzi.io/use-connector-resources: "true" 1 spec: #... image: debezium-container-for-mysql 2
以下のコマンドを入力して、
KafkaConnect
CR を OpenShift Kafka Connect 環境に適用します。oc create -f dbz-connect.yaml
このコマンドは、Debezium コネクターを実行するために作成したイメージの名前を指定する Kafka Connect インスタンスを追加します。
Debezium MySQL コネクターインスタンスを設定する
KafkaConnector
カスタムリソースを作成します。通常、コネクター設定プロパティーを設定する
.yaml
ファイルに Debezium MySQL コネクターを設定します。コネクター設定は、Debezium に対して、スキーマおよびテーブルのサブセットにイベントを生成するよう指示する可能性があり、または機密性の高い、大きすぎる、または不必要な指定のコラムで Debezium が値を無視、マスク、または切り捨てするようにプロパティーを設定する可能性もあります。以下の例では、ポート
3306
の MySQL ホスト (192.168.99.100
) に接続し、inventory
データベースへの変更をキャプチャーする Debezium コネクターを設定します。dbserver1
は、サーバーの論理名です。MySQL
inventory-connector.yaml
apiVersion: kafka.strimzi.io/v1beta1 kind: KafkaConnector metadata: name: inventory-connector 1 labels: strimzi.io/cluster: my-connect-cluster spec: class: io.debezium.connector.mysql.MySqlConnector tasksMax: 1 2 config: 3 database.hostname: mysql 4 database.port: 3306 database.user: debezium database.password: dbz database.server.id: 184054 5 database.server.name: dbserver1 6 database.include.list: inventory 7 database.history.kafka.bootstrap.servers: my-cluster-kafka-bootstrap:9092 8 database.history.kafka.topic: schema-changes.inventory 9
表2.19 コネクター設定の説明 項目 説明 1
コネクターの名前。
2
1 度に 1 つのタスクのみが動作する必要があります。MySQL コネクターは MySQL サーバーの
binlog
を読み取るため、単一のコネクタータスクを使用することで、順序とイベントの処理が適切に行われるようになります。Kafka Connect サービスはコネクターを使用して作業を行う 1 つ以上のタスクを開始し、実行中のタスクを自動的に Kafka Connect サービスのクラスター全体に分散します。いずれかのサービスが停止またはクラッシュすると、これらのタスクは稼働中のサービスに再分散されます。3
コネクターの設定。
4
データベースホスト。これは、MySQL サーバーを実行しているコンテナーの名前です (
mysql
)。5
connector の一意 ID。
6
MySQL サーバーまたはクラスターの論理名。この名前は、変更イベントレコードを受信するすべての Kafka トピックの接頭辞として使用されます。
7
inventory
データベースの変更のみがキャプチャーされます。8
DDL ステートメントをデータベース履歴トピックに書き込み、復元するためにコネクターによって使用される Kafka ブローカーのリスト。再起動時に、コネクターが読み取りを開始すべき時点で binlog に存在したデータベースのスキーマを復元します。
9
データベー履歴トピックの名前。このトピックは内部使用のみを目的としており、コンシューマーが使用しないようにしてください。
Kafka Connect でコネクターインスタンスを作成します。たとえば、
KafkaConnector
リソースをinventory-connector.yaml
ファイルに保存した場合は、以下のコマンドを実行します。oc apply -f inventory-connector.yaml
上記のコマンドは
inventory-connector
を登録し、コネクターはKafkaConnector
CR に定義されているinventory
データベースに対して実行を開始します。コネクターが作成され、起動されたことを確認します。
Kafka Connect ログ出力を表示して、コネクターが作成され、指定データベースの変更のキャプチャーが開始されたことを確認します。
oc logs $(oc get pods -o name -l strimzi.io/cluster=my-connect-cluster)
ログの出力を確認し、Debezium により初回のスナップショットが実行されたことを確認します。ログには、以下のメッセージと同様の出力が表示されます。
... INFO Starting snapshot for ... ... INFO Snapshot is using user 'debezium' ...
コネクターがエラーがなく正常に起動すると、コネクターが変更をキャプチャーする各テーブルのトピックが作成されます。CR のサンプルでは、
include.list
プロパティーに指定されたテーブルのトピックがあります。ダウンストリームアプリケーションは、これらのトピックをサブスクライブできます。以下のコマンドを実行して、コネクターによってトピックが作成されたことを検証します。
oc get kafkatopics
Debezium MySQL コネクターに設定できる設定プロパティーの完全リストは、MySQL コネクター設定プロパティーを参照してください。
結果
コネクターが起動すると、コネクターが設定された MySQL データベースの 整合性スナップショットが実行 されます。その後、コネクターは行レベルの操作のデータ変更イベントの生成を開始し、変更イベントレコードを Kafka トピックにストリーミングします。
2.5.2. Debezium MySQL コネクター設定プロパティーの説明
Debezium MySQL コネクターには、アプリケーションに適したコネクター動作を実現するために使用できる設定プロパティーが多数あります。多くのプロパティーにはデフォルト値があります。プロパティーに関する情報は、以下のように設定されています。
以下の設定プロパティーは、デフォルト値がない場合は必須です。
プロパティー | デフォルト | 説明 |
---|---|---|
コネクターの一意名。同じ名前で再登録を試みると失敗します。このプロパティーはすべての Kafka Connect コネクターに必要です。 | ||
コネクターの Java クラスの名前。MySQL コネクターに常に | ||
| このコネクターのために作成する必要のあるタスクの最大数。MySQL コネクターは常に単一のタスクを使用するため、この値を使用しません。そのため、デフォルト値は常に許容されます。 | |
MySQL データベースサーバーの IP アドレスまたはホスト名。 | ||
| MySQL データベースサーバーのポート番号 (整数)。 | |
MySQL データベースサーバーへの接続時に使用する MySQL ユーザーの名前。 | ||
MySQL データベースサーバーへの接続時に使用するパスワード。 | ||
Debezium が変更をキャプチャーする特定の MySQL データベースサーバー/クラスターの namespace を識別および提供する論理名。論理名は、他のコネクター全体で一意となる必要があります。これは、このコネクターによって生成されるイベントを受信するすべての Kafka トピック名の接頭辞として使用されるためです。この名前には英数字とアンダースコアのみを使用できます。 | ||
random | このデータベースクライアントの数値 ID。MySQL クラスターで現在稼働しているすべてのデータベースプロセスで一意である必要があります。このコネクターは、MySQL データベースクラスターを (この一意の ID を持つ) 別のサーバーとして結合するため、binlog を読み取ることができます。デフォルトでは、5400 から 6400 までの乱数が生成されますが、値を明示的に設定することが推奨されます。 | |
コネクターがデータベーススキーマの履歴を保存する Kafka トピックの完全名。 | ||
Kafka クラスターへの最初の接続を確立するために コネクターが使用するホストとポートのペアの一覧。このコネクションは、コネクターによって以前に保存されたデータベーススキーマ履歴の取得や、ソースデータベースから読み取られる各 DDL ステートメントの書き込みに使用されます。各ペアは、Kafka Connect プロセスによって使用される同じ Kafka クラスターを示す必要があります。 | ||
空の文字列 |
変更をキャプチャーするデータベースの名前と一致する正規表現のコンマ区切りリスト (任意)。コネクターは、名前が | |
空の文字列 |
変更をキャプチャーしないデータベースの名前と一致する正規表現のコンマ区切りリスト (任意)。コネクターは、名前が | |
空の文字列 |
変更をキャプチャーするテーブルの完全修飾テーブル識別子と一致する正規表現のコンマ区切りリスト (任意)。コネクターは | |
空の文字列 |
変更をキャプチャーしないテーブルの完全修飾テーブル識別子と一致する正規表現のコンマ区切りリスト (任意)。コネクターは | |
空の文字列 | 変更イベントレコード値から除外する列の完全修飾名と一致する正規表現のコンマ区切りリスト (任意)。列の完全修飾名の形式は databaseName.tableName.columnName です。 | |
空の文字列 | 変更イベントレコード値に含める列の完全修飾名と一致する正規表現のコンマ区切りリスト (任意)。列の完全修飾名の形式は databaseName.tableName.columnName です。 | |
該当なし | フィールド値が指定された文字数より長い場合に、変更イベントレコード値で値を省略する必要がある文字ベースの列の完全修飾名と一致する正規表現のコンマ区切りリスト (任意)。単一の設定で、異なる長さの複数のプロパティーを設定できます。長さは正の整数である必要があります。列の完全修飾名の形式は databaseName.tableName.columnName です。 | |
該当なし |
変更イベントメッセージで、指定された数のアスタリスク ( | |
該当なし |
変更イベントレコード値で値が仮名である必要がある文字ベースの列の完全修飾名と一致する、正規表現のコンマ区切りリスト (任意)。仮名は、アルゴリズム | |
該当なし | 出力された変更イベントレコードの該当するフィールドスキーマに元の型および長さをパラメーターとして追加する必要がある列の完全修飾名と一致する、正規表現のコンマ区切りリスト (任意)。以下のスキーマパラメーターは、それぞれ可変幅型の元の型名および長さを伝達するために使用されます。
それぞれ元の型名と長さ (可変幅型の場合) を伝達するために使用されます。これは、シンクデータベースの対応する列を適切にサイズ調整するのに便利です。列の完全修飾名の形式は以下のいずれかになります。 databaseName.tableName.columnName databaseName.schemaName.tableName.columnName | |
該当なし | 出力された変更イベントレコードの該当するフィールドスキーマに元の型および長さをパラメーターとして追加する必要がある列のデータベース固有のデータ型名と一致する、正規表現のコンマ区切りリスト (任意)。以下のスキーマパラメーターは、それぞれ可変幅型の元の型名および長さを伝達するために使用されます。
それぞれ元の型名と長さ (可変幅型の場合) を伝達するために使用されます。これは、シンクデータベースの対応する列を適切にサイズ調整するのに便利です。完全修飾データ型名の形式は以下のいずれかになります。 databaseName.tableName.typeName databaseName.schemaName.tableName.typeName MySQL 固有のデータ型名の一覧は、MySQL コネクターによるデータ型のマッピング方法 を参照してください。 | |
|
時間、日付、およびタイムスタンプは、以下を含む異なる精度の種類で表すことができます。 | |
|
コネクターによる | |
|
変更イベントで BIGINT UNSIGNED 列を表す方法を指定します。可能な設定: | |
| コネクターがデータベーススキーマの変更を、データベースサーバー ID と同じ名前の Kafka トピックに公開するかどうかを指定するブール値。各スキーマの変更はデータベース名が含まれるキーを使用して記録され、その値には DDL ステートメントが含まれます。これは、コネクターがデータベース履歴を内部で記録する方法には依存しません。 | |
|
変更イベントを生成した元の SQL クエリーがコネクターに含まれる必要があるかどうかを指定するブール値。 | |
|
binlog イベントのデシリアライズ中にコネクターがどのように例外に反応するかを指定します。 | |
|
内部スキーマ表現に存在しないテーブルに関連する binlog イベントに対してコネクターがどのように反応する必要があるかを指定します。つまり、内部表現はデータベースと一貫性がありません。 | |
|
データベースログから読み取られた変更イベントが Kafka に書き込まれる前に配置される、ブロッキングキューの最大サイズを指定する正の整数値。このキューは、Kafka への書き込みが遅い場合や Kafka が利用できない場合などに、binlog リーダーにバックプレシャーを提供できます。キューに発生するイベントは、このコネクターによって定期的に記録されるオフセットには含まれません。デフォルトは 8192 で、 | |
| このコネクターの反復処理中に処理される必要があるイベントの各バッチの最大サイズを指定する正の整数値。デフォルトは 2048 です。 | |
| ブロッキングキューの最大サイズ (バイト単位) の long 値。この機能はデフォルトで無効になっています。正の long 値が設定されると有効になります。 | |
| コネクターがイベントのバッチの処理を開始する前に、新しい変更イベントの発生を待つ期間をミリ秒単位で指定する正の整数値。デフォルトは 1000 ミリ秒 (1 秒) です。 | |
| コネクターが MySQL データベースサーバーへの接続を試行した後、タイムアウトするまでの最大の待機期間をミリ秒単位で指定する正の整数値。デフォルトは 30 秒です。 | |
MySQL サーバーで binlog の位置を見つけるために使用される GTID セットのソース UUID に一致する、正規表現のコンマ区切りリスト。これらの include パターンのいずれかに一致するソースを持つ GTID の範囲のみが使用されます。 | ||
MySQL サーバーで binlog の位置を見つけるために使用される GTID セットのソース UUID に一致する、正規表現のコンマ区切りリスト。これらすべての exclude パターンに一致しないソースを持つ GTID の範囲のみが使用されます。また、 | ||
|
削除 イベントの後に tombstone イベントが続くかどうかを制御します。 | |
該当なし |
テーブルの列名と一致する正規表現が含まれるテーブルのセミコロン区切りのリスト。コネクターは、一致する列の値を Kafka トピックに送信する変更イベントレコードのキーフィールドにマップします。これは、テーブルにプライマリーキーがない場合や、プライマリーキーではないフィールドに応じて Kafka トピックで変更イベントレコードを順序付けする場合に便利です。 | |
bytes |
バイナリー列 (例: |
高度な MySQL コネクター設定プロパティー
以下の表は、高度な MySQL コネクタープロパティー について説明しています。これらのプロパティーのデフォルト値を変更する必要はほとんどありません。そのため、コネクター設定にデフォルト値を指定する必要はありません。
プロパティー | デフォルト | 説明 |
---|---|---|
| MySQL サーバー/クラスターへの接続を確実に維持するために、別のスレッドを使用するかどうかを指定するブール値。 | |
| 組み込みシステムテーブルを無視するかどうかを指定するブール値。これは、テーブルの include および exclude リストに関係なく適用されます。デフォルトでは、システムテーブルは変更がキャプチャーされないように除外され、システムテーブルに変更が加えられてもイベントは生成されません。 | |
| 永続化されたデータのポーリングが行われている間にコネクターが起動/回復を待つ最大時間 (ミリ秒単位) を指定する整数値。デフォルトは 100 ミリ秒です。 | |
|
エラーでコネクターのリカバリーが失敗する前に、コネクターが永続化された履歴データの読み取りを試行する最大回数。データが受信されなかった場合に最大待機する時間は、 | |
|
コネクターが不正または不明なデータベースのステートメントを無視するかどうか、または人が問題を修正するために処理を停止するかどうかを指定するブール値。安全なデフォルトは | |
|
コネクターがすべての DDL ステートメントを記録するかどうかを指定するブール値 | |
|
暗号化された接続を使用するかどうかを指定します。可能な設定: | |
0 |
binlog リーダーによって使用される先読みバッファーのサイズ。デフォルト設定 | |
|
コネクターの起動時にスナップショットを実行するための基準を指定します。可能な設定: | |
|
コネクターがグローバル MySQL 読み込みロックを保持するかどうか、およびその期間を制御します。これにより、コネクターによるスナップショットの実行中にデータベースが更新されないようにします。可能な設定: | |
|
スナップショットを作成する | |
スナップショットに含まれるテーブル行を制御します。このプロパティーはスナップショットにのみ影響します。binlog からキャプチャーされたイベントには影響しません。databaseName.tableName の形式で完全修飾テーブル名のコンマ区切りリストを指定します。 | ||
|
スナップショットの実行中、コネクターは変更をキャプチャーするように設定されている各テーブルにクエリーを実行します。コネクターは各クエリーの結果を使用して、そのテーブルのすべての行のデータが含まれる読み取りイベントを生成します。このプロパティーは、MySQL コネクターがテーブルの結果をメモリーに格納するか、またはストリーミングを行うかを決定します。メモリーへの格納はすばやく処理できますが、大量のメモリーを必要とします。ストリーミングを行うと、処理は遅くなりますが、非常に大きなテーブルにも対応できます。このプロパティーの設定は、コネクターが結果のストリーミングを行う前にテーブルに含まれる必要がある行の最小数を指定します。 | |
|
コネクターがハートビートメッセージを Kafka トピックに送信する頻度を制御します。デフォルトの動作では、コネクターはハートビートメッセージを送信しません。 | |
|
コネクターがハートビートメッセージを送信するトピックの名前を制御します。トピック名のパターンは次のようになります。 | |
トランザクションログを読み取る接続ではなく、データベースへの JDBC 接続が確立されたときに実行される SQL ステートメントのセミコロン区切りのリスト。SQL ステートメントでセミコロンを区切り文字としてではなく、文字として指定する場合は、2 つのセミコロン ( | ||
コネクターの起動時にスナップショットを実行するまでコネクターが待つ必要がある間隔 (ミリ秒単位)。クラスターで複数のコネクターを起動する場合、このプロパティーは、コネクターのリバランスが行われる原因となるスナップショットの中断を防ぐのに役立ちます。 | ||
スナップショットの実行中、コネクターは行のバッチでテーブルの内容を読み取ります。このプロパティーは、バッチの行の最大数を指定します。 | ||
| スナップショットの実行時に、テーブルロックを取得するまで待つ最大時間 (ミリ秒単位) を指定する正の整数。コネクターがこの期間にテーブルロックを取得できないと、スナップショットは失敗します。Debezium MySQL コネクターによるデータベーススナップショットの実行方法 を参照してください。 | |
|
コネクターによって 2 桁の西暦が 4 桁の西暦に変換されるかどうかを示すブール値。変換が完全にデータベースに委譲されている場合は、 | |
コネクターが | Avro の命名要件 に準拠するためにフィールド名がサニタイズされるかどうかを示します。 | |
ストリーミング中にスキップする oplog 操作のコンマ区切りリスト。指定できる値は、 |
パススルー設定プロパティー
MySQL コネクターは、Kafka プロデューサーおよびコンシューマーの作成時に使用されるパススルー設定プロパティーもサポートします。具体的には、データベース履歴に書き込む Kafka プロデューサーの作成時に、database.history.producer.
接頭辞で始まるすべてのコネクター設定プロパティーが (接頭辞なしで) 使用されます。接頭辞 database.history.consumer.
で始まるすべてのプロパティーは、コネクターの起動時にデータベース履歴を読み取る Kafka コンシューマーを作成する際に(接頭辞なしで)使用されます。
たとえば、以下のコネクター設定プロパティーを使用すると、Kafka ブローカーへの接続をセキュアにすることができます。
database.history.producer.security.protocol=SSL database.history.producer.ssl.keystore.location=/var/private/ssl/kafka.server.keystore.jks database.history.producer.ssl.keystore.password=test1234 database.history.producer.ssl.truststore.location=/var/private/ssl/kafka.server.truststore.jks database.history.producer.ssl.truststore.password=test1234 database.history.producer.ssl.key.password=test1234 database.history.consumer.security.protocol=SSL database.history.consumer.ssl.keystore.location=/var/private/ssl/kafka.server.keystore.jks database.history.consumer.ssl.keystore.password=test1234 database.history.consumer.ssl.truststore.location=/var/private/ssl/kafka.server.truststore.jks database.history.consumer.ssl.truststore.password=test1234 database.history.consumer.ssl.key.password=test1234
パススループロパティーの詳細は、Kafka のドキュメント を参照してください。
データベースドライバーのパススループロパティー
Kafka プロデューサーおよびコンシューマーのパススループロパティーの他に、データベースドライバーのパススループロパティー があります。これらのプロパティーには database.
接頭辞があります。たとえば、database.tinyInt1isBit=false
は JDBC URL に渡されます。
2.6. Debezium MySQL コネクターのパフォーマンスの監視
Debezium MySQL コネクターは、Zookeeper、Kafka、および Kafka Connect によって提供される JMX メトリクスの組み込みサポートに加えて、3 種類のメトリクスを提供します。
- スナップショットメトリクス は、スナップショットの実行中にコネクター操作に関する情報を提供します。
- Binlog メトリクス は、コネクターが binlog を読み取る際のコネクター操作に関する情報を提供します。
- スキーマ履歴メトリクス は、コネクターのスキーマ履歴の状態に関する情報を提供します。
Debezium モニターリングのドキュメント では、JMX を使用してこれらのメトリクスを公開する方法の詳細を提供します。
2.6.1. MySQL データベースのスナップショット作成時の Debezium の監視
MBean は debezium.mysql:type=connector-metrics,context=snapshot,server=<database.server.name>
です。
属性 | タイプ | 説明 |
---|---|---|
| コネクターが読み取りした最後のスナップショットイベント。 | |
| コネクターが最新のイベントを読み取りおよび処理してからの経過時間 (ミリ秒単位)。 | |
| 前回の開始またはリセット以降にコネクターで確認されたイベントの合計数。 | |
| コネクターに設定された include/exclude リストのフィルターリングルールによってフィルターされたイベントの数。 | |
| コネクターによって監視されるテーブルの一覧。 | |
| snapshotter とメインの Kafka Connect ループの間でイベントを渡すために使用されるキューの長さ。 | |
| snapshotter とメインの Kafka Connect ループの間でイベントを渡すために使用されるキューの空き容量。 | |
| スナップショットに含まれているテーブルの合計数。 | |
| スナップショットによってまだコピーされていないテーブルの数。 | |
| スナップショットが起動されたかどうか。 | |
| スナップショットが中断されたかどうか。 | |
| スナップショットが完了したかどうか。 | |
| スナップショットが完了したかどうかに関わらず、これまでスナップショットにかかった時間 (秒単位)。 | |
| スナップショットの各テーブルに対してスキャンされる行数が含まれるマップ。テーブルは、処理中に増分がマップに追加されます。スキャンされた 10,000 行ごとに、テーブルの完成時に更新されます。 | |
|
キューの最大バッファー (バイト単位)。 | |
| キュー内のレコードの現在のデータ (バイト単位)。 |
Debezium MySQL コネクターは、HoldingGlobalLock
カスタムスナップショットメトリクスも提供します。このメトリクスは、コネクターが現在グローバルまたはテーブル書き込みロックを保持するかどうかを示すブール値に設定されます。
2.6.2. Debezium MySQL コネクターの binlog 読み取りの監視
MBean は debezium.mysql:type=connector-metrics,context=binlog,server=<database.server.name>
です。
トランザクション関連の属性は、binlog イベントのバッファーが有効になっている場合にのみ利用できます。詳細は、高度な MySQL コネクター設定プロパティーの binlog.buffer.size
を参照してください。
属性 | タイプ | 説明 |
---|---|---|
| コネクターが読み取られた最後のストリーミングイベント。 | |
| コネクターが最新のイベントを読み取りおよび処理してからの経過時間 (ミリ秒単位)。 | |
| 前回の開始またはリセット以降にコネクターで確認されたイベントの合計数。 | |
| コネクターに設定された include/exclude リストのフィルターリングルールによってフィルターされたイベントの数。 | |
| コネクターによって監視されるテーブルの一覧。 | |
| ストリーマーとメイン Kafka Connect ループの間でイベントを渡すために使用されるキューの長さ。 | |
| ストリーマーとメインの Kafka Connect ループの間でイベントを渡すために使用されるキューの空き容量。 | |
| コネクターが現在データベースサーバーに接続されているかどうかを示すフラグ。 | |
| 最後の変更イベントのタイムスタンプとそれを処理するコネクターとの間の期間 (ミリ秒単位)。この値は、データベースサーバーとコネクターが稼働しているマシンのクロック間の差異に対応します。 | |
| コミットされた処理済みトランザクションの数。 | |
| 最後に受信したイベントの位置。 | |
| 最後に処理されたトランザクションのトランザクション識別子。 | |
| キューの最大バッファー (バイト単位)。 | |
| キュー内のレコードの現在のデータ (バイト単位)。 |
Debezium MySQL コネクターは、以下のカスタム binlog メトリクスも提供します。
属性 | タイプ | 説明 |
---|---|---|
| コネクターによって最後に読み取られた binlog ファイルの名前。 | |
| コネクターによって読み取られた binlog 内の最新の位置 (バイト単位)。 | |
| コネクターが現在 MySQL サーバーから GTID を追跡しているかどうかを示すフラグ。 | |
| binlog の読み取り時にコネクターによって処理される最新の GTID セットの文字列表現。 | |
| MySQL コネクターによってスキップされたイベントの数。通常、MySQL の binlog からの不正形式のイベントまたは解析不可能なイベントが原因で、イベントがスキップされます。 | |
| MySQL コネクターによる切断の数。 | |
| ロールバックされ、ストリーミングされなかった処理済みトランザクションの数。 | |
|
想定された | |
|
先読みバッファーに適合しないトランザクションの数。最適なパフォーマンスを得るには、この値は |
2.6.3. Debezium MySQL コネクターのスキーマ履歴の監視
MBean は debezium.mysql:type=connector-metrics,context=schema-history,server=<database.server.name>
です。
属性 | タイプ | 説明 |
---|---|---|
|
データベース履歴の状態を示す | |
| リカバリーが開始された時点のエポック秒の時間。 | |
| リカバリーフェーズ中に読み取られた変更の数。 | |
| リカバリーおよびランタイム中に適用されるスキーマ変更の合計数。 | |
| 最後の変更が履歴ストアから復元された時点からの経過時間 (ミリ秒単位)。 | |
| 最後の変更が適用された時点からの経過時間 (ミリ秒単位)。 | |
| 履歴ストアから復元された最後の変更の文字列表現。 | |
| 最後に適用された変更の文字列表現。 |
2.7. Debezium MySQL コネクターによる障害および問題の処理方法
Debezium は、複数のアップストリームデータベースのすべての変更をキャプチャーする分散システムであり、イベントの見逃しや損失は発生しません。システムが正常に操作している場合や、慎重に管理されている場合は、Debezium は変更イベントレコードごとに 1 度だけ 配信します。
障害が発生しても、システムはイベントを失いません。ただし、障害から復旧している間は、変更イベントが繰り返えされる可能性があります。このような正常でない状態では、Debezium は Kafka と同様に、変更イベントを 少なくとも 1 回 配信します。
詳細は以下を参照してください。
設定および起動エラー
以下の状況では、起動時にコネクターが失敗し、エラーまたは例外がログに記録され、実行が停止されます。
- コネクターの設定が無効である。
- 指定の接続パラメーターを使用してコネクターを MySQL サーバーに接続できない。
- MySQL に履歴がない binlog の位置でコネクターが再起動を試行する。
このような場合、エラーメッセージには問題の詳細が含まれ、推奨される回避策も含まれることがあります。設定の修正したり、MySQL の問題に対処した後、コネクターを再起動します。
ただし、高可用性 MySQL クラスターで GTID が有効になっている場合は、コネクターをすぐに再起動できます。これはクラスターの別の MySQL サーバーに接続し、最後のトランザクションを表すサーバーの binlog の場所を特定し、その特定の場所から新しいサーバーの binlog の読み取りを開始します。
GTID が有効になっていない場合、コネクターは接続した MySQL サーバーのみの binlog の位置を記録します。正しい binlog の位置から再起動するには、その特定のサーバーに再接続する必要があります。
Kafka Connect が正常に停止する
Kafka Connect が正常に停止すると、Debezium MySQL コネクタータスクが停止され、新しい Kafka Connect プロセスで再起動される間に短い遅延が発生します。
Kafka Connect プロセスのクラッシュ
Kafka Connect がクラッシュすると、プロセスが停止し、最後に処理されたオフセットが記録されずに Debezium MySQL コネクタータスクが終了します。分散モードでは、Kafka Connect は他のプロセスでコネクタータスクを再起動します。ただし、MySQL コネクターは以前のプロセスで記録された最後のオフセットから再開します。つまり、代替のタスクによってクラッシュ前に処理された同じイベントの一部が生成され、重複したイベントが作成される可能性があります。
各変更イベントメッセージには、重複イベントの特定に使用できるソース固有の情報が含まれます。以下に例を示します。
- イベント元
- MySQL サーバーのイベント時間
- binlog ファイル名と位置
- GTID (使用されている場合)
MySQL が binlog ファイルをパージする
Debezium MySQL コネクターが長時間停止すると、MySQL サーバーは古い binlog ファイルをパージするため、コネクターの最後の位置が失われる可能性があります。コネクターが再起動すると、MySQL サーバーに開始点がなくなり、コネクターは別の最初のスナップショットを実行します。スナップショットが無効の場合、コネクターはエラーによって失敗します。
MySQL コネクターが最初のスナップショットを実行する方法に関する詳細は Debezium MySQL コネクターによるデータベーススナップショットの実行方法 を参照してください。
第3章 PostgreSQL の Debezium コネクター
Debezium の PostgreSQL コネクターは、PostgreSQL データベースのスキーマで行レベルの変更をキャプチャーします。PostgreSQL バージョン 10、11、および 12 がサポートされます。
PostgreSQL サーバーまたはクラスターに初めて接続すると、コネクターはすべてのスキーマの整合性スナップショットを作成します。スナップショットの完了後、コネクターはデータベースのコンテンツを挿入、更新、および削除する行レベルの変更を継続的にキャプチャーします。これらの行レベルの変更は、PostgreSQL データベースにコミットされています。コネクターはデータの変更イベントレコードを生成し、それらを Kafka トピックにストリーミングします。各テーブルのデフォルトの動作では、コネクターは生成されたすべてのイベントをそのテーブルの個別の Kafka トピックにストリーミングします。アプリケーションとサービスは、そのトピックからのデータ変更イベントレコードを使用します。
Debezium PostgreSQL コネクターを使用するための情報および手順は、以下のように設定されています。
- 「Debezium PostgreSQL コネクターの概要」
- 「Debezium PostgreSQL コネクターの仕組み」
- 「Debezium PostgreSQL コネクターのデータ変更イベントの説明」
- 「Debezium PostgreSQL コネクターによるデータ型のマッピング方法」
- 「Debezium コネクターを実行するための PostgreSQL の設定」
- 「Debezium PostgreSQL コネクターのデプロイメント」
- 「Debezium PostgreSQL コネクターのパフォーマンスの監視」
- 「Debezium PostgreSQL コネクターによる障害および問題の処理方法」
3.1. Debezium PostgreSQL コネクターの概要
PostgreSQL の 論理デコード 機能は、バージョン 9.4 で導入されました。これは、トランザクションログにコミットされた変更の抽出を可能にし、出力プラグイン を用いてユーザーフレンドリーな方法でこれらの変更の処理を可能にするメカニズムです。出力プラグインを使用すると、クライアントは変更を使用できます。
PostgreSQL コネクターには、連携してデータベースの変更を読み取りおよび処理する 2 つの主要部分が含まれています。
-
pgoutput
は、PostgreSQL 10+ の標準的な論理デコード出力プラグインです。これは、この Debezium リリースでサポートされている唯一の論理デコード出力プラグインです。このプラグインは PostgreSQL コミュニティーにより維持され、PostgreSQL 自体によって 論理レプリケーション に使用されます。このプラグインは常に存在するため、追加のライブラリーをインストールする必要はありません。Debezium コネクターは、raw レプリケーションイベントストリームを直接変更イベントに変換します。 - PostgreSQL の ストリーミングレプリケーションプロトコル および PostgreSQL JDBC ドライバー を使用して、論理デコード出力プラグインによって生成された変更を読み取る Java コード (実際の Kafka Connect コネクター)。
コネクターは、キャプチャーされた各行レベルの挿入、更新、および削除操作の 変更イベント を生成し、個別の Kafka トピックの各テーブルに対する変更イベントレコードを送信します。クライアントアプリケーションは、対象のデータベーステーブルに対応する Kafka トピックを読み取り、これらのトピックから受け取るすべての行レベルイベントに対応できます。
通常、PostgreSQL は一定期間後にログ先行書き込み (WAL、write-ahead log) をパージします。つまり、コネクターにはデータベースに加えられたすべての変更の完全な履歴はありません。そのため、PostgreSQL コネクターが最初に特定の PostgreSQL データベースに接続すると、データベーススキーマごとに 整合性スナップショット を実行して起動します。コネクターは、スナップショットの完成後に、スナップショットが作成された正確な時点から変更のストリーミングを続行します。これにより、コネクターはすべてのデータの整合性のあるビューで開始し、スナップショットの作成中に加えられた変更は省略されません。
コネクターはフォールトトラレントです。コネクターは変更を読み取り、イベントを生成するため、各イベントの WAL の位置を記録します。コネクターが何らかの理由で停止した場合 (通信障害、ネットワークの問題、クラッシュなど)、コネクターは再起動後に最後に停止した場所から WAL の読み取りを続行します。これにはスナップショットが含まれます。スナップショット中にコネクターが停止した場合、コネクターは再起動時に新しいスナップショットを開始します。
コネクターは PostgreSQL の論理デコード機能に依存および反映します。これには、以下の制限があります。
- 論理デコードは DDL の変更をサポートしません。よって、コネクターは DDL の変更イベントをコンシューマーに報告できません。
-
論理デコードのレプリケーションスロットは、
プライマリー
サーバーでのみサポートされます。PostgreSQL サーバーのクラスターがある場合、コネクターはアクティブなprimary
サーバーでのみ実行できます。hot
またはwarm
スタンバイのレプリカでは実行できません。primary
サーバーが失敗するか降格されると、コネクターは停止します。primary
サーバーの回復後に、コネクターを再起動できます。別の PostgreSQL サーバーがprimary
に昇格された場合は、コネクターの設定を調整してからコネクターを再起動します。
問題が発生した場合の動作 には、問題が発生した場合のコネクターの動作が説明されています。
Debezium は現在、UTF-8 文字エンコーディングのデータベースのみをサポートしています。1 バイト文字エンコーディングでは、拡張 ASCII コード文字が含まれる文字列を正しく処理できません。
3.2. Debezium PostgreSQL コネクターの仕組み
Debezium PostgreSQL コネクターを最適に設定および実行するには、コネクターによるスナップショットの実行方法、変更イベントのストリーム方法、Kafka トピック名の決定方法、およびメタデータの使用方法を理解すると便利です。
詳細は以下を参照してください。
3.2.1. PostgreSQL コネクターのセキュリティー
Debezium コネクターを使用して PostgreSQL データベースから変更をストリーミングするには、コネクターは特定の権限がデータベースで必要になります。必要な権限を付与する方法の 1 つとして、ユーザーに superuser
権限を付与する方法がありますが、これにより PostgreSQL データが不正アクセスによって公開される可能性ああります。Debezium ユーザーに過剰な権限を付与するのではなく、特定の特権を付与する専用の Debezium レプリケーションユーザーを作成することが推奨されます。
Debezium PostgreSQL ユーザーの権限設定の詳細は、パーミッションの設定 を参照してください。PostgreSQL の論理レプリケーションセキュリティーの詳細は、PostgreSQL のドキュメント を参照してください。
3.2.2. Debezium PostgreSQL コネクターによるデータベーススナップショットの実行方法
ほとんどの PostgreSQL サーバーは、WAL セグメントにデータベースの完全な履歴を保持しないように設定されています。つまり、PostgreSQL コネクターは WAL のみを読み取ってもデータベースの履歴全体を確認できません。そのため、コネクターが最初に起動すると、データベースの最初の 整合性スナップショット が実行されます。スナップショットを実行するためのデフォルト動作は、以下の手順で設定されます。この動作を変更するには、snapshot.mode
コネクター設定プロパティー を initial
以外の値に設定します。
-
SERIALIZABLE、READ ONLY、DEFERRABLE 分離レベルでトランザクションを開始し、このトランザクションでの後続の読み取りがデータの単一バージョンに対して行われるようにします。他のクライアントによる後続の
INSERT
、UPDATE
、およびDELETE
操作によるデータの変更は、このトランザクションでは確認できません。 追跡されている各テーブルで
ACCESS SHARE MODE
ロックを取得し、スナップショットの実行中にテーブルの構造的な変更が発生しないようにします。これらのロックは、スナップショット中にテーブルのINSERT
、UPDATE
、DELETE
操作が実行されないようにします。snapshot.mode
がexported
に設定されている場合、このステップは省略され、コネクターはロックフリーのスナップショットを実行することができます。- サーバーのトランザクションログの現在の位置を読み取ります。
-
データベーステーブルとスキーマをスキャンし、各行の
READ
イベントを生成し、そのイベントを適切なテーブル固有の Kafka トピックに書き込みます。 - トランザクションをコミットします。
- コネクターオフセットにスナップショットの正常な完了を記録します。
コネクターに障害が発生した場合、コネクターのリバランスが発生した場合、または 1. の後で 6. の完了前に停止した場合、コネクターは再起動後に新しいスナップショットを開始します。コネクターによって最初のスナップショットが完了した後、PostgreSQL コネクターは 3. で読み取りした位置からストリーミングを続行します。これにより、コネクターが更新を見逃さないようします。何らかの理由でコネクターが再び停止した場合、コネクターは再起動後に最後に停止した位置から変更のストリーミングを続行します。
snapshot.mode
を exported
に設定するように PostgreSQL コネクターを設定することを強く推奨します。initial
モード、initial only
モード、always
モードでは、データベースの負荷が高いときに、コネクターがスナップショットの実行から変更イベント記録のストリーミングに切り替わる間に、いくつかのイベントが失われることがあります。これは既知の問題であり、影響を受けるスナップショットモードは、内部で exported
モードを使用するように作用されます (DBZ-2337)。
設定 | 説明 |
---|---|
|
コネクターは起動時に常にスナップショットを実行します。スナップショットが完了した後、コネクターは上記の手順の 3. から変更のストリーミングを続行します。このモードは、以下のような状況で使用すると便利です。
|
|
コネクターはスナップショットを実行しません。このようにコネクターを設定したすると、起動時の動作は次のようになります。Kafka オフセットトピックに以前保存された LSN がある場合、コネクターはその位置から変更をストリーミングを続行します。保存された LSN がない場合、コネクターはサーバーで PostgreSQL の論理レプリケーションスロットが作成された時点で変更のストリーミングを開始します。 |
| コネクターはデータベースのスナップショットを実行し、変更イベントレコードをストリーミングする前に停止します。コネクターが起動していても、停止前にスナップショットを完了しなかった場合、コネクターはスナップショットプロセスを再起動し、スナップショットの完了時に停止します。 |
| コネクターは、レプリケーションスロットが作成された時点に基づいてデータベーススナップショットを実行します。このモードは、ロックのない方法でスナップショットを実行するのに最適です。 |
3.2.3. Debezium PostgreSQL コネクターによる変更イベントレコードのストリーミング方法
通常、PostgreSQL コネクターは、接続されている PostgreSQL サーバーから変更をストリーミングするのに大半の時間を費やします。このメカニズムは、PostgreSQL のレプリケーションプロトコル に依存します。このプロトコルにより、クライアントはログシーケンス番号 (LSN) と呼ばれる特定の場所で変更がサーバーのトランザクションログにコミットされる際に、サーバーから変更を受信することができます。
サーバーがトランザクションをコミットするたびに、別のサーバープロセスが 論理デコードプラグイン からコールバック関数を呼び出します。この関数はトランザクションからの変更を処理し、特定の形式 (Debezium プラグインの場合は Protobuf または JSON) に変換して、出力ストリームに書き込みます。その後、クライアントは変更を使用できます。
Debezium PostgreSQL コネクターは PostgreSQL クライアントとして動作します。コネクターが変更を受信すると、イベントを Debezium の create、update、または delete イベントに変換します。これには、イベントの LSN が含まれます。PostgreSQL コネクターは、同じプロセスで実行されている Kafka Connect フレームワークにレコードのこれらの変更イベントを転送します。Kafka Connect プロセスは、変更イベントレコードを適切な Kafka トピックに生成された順序で非同期に書き込みます。
Kafka Connect は定期的に最新の オフセット を別の Kafka トピックに記録します。オフセットは、各イベントに含まれるソース固有の位置情報を示します。PostgreSQL コネクターでは、各変更イベントに記録された LSN がオフセットです。
Kafka Connect が正常にシャットダウンすると、コネクターを停止し、すべてのイベントレコードを Kafka にフラッシュして、各コネクターから受け取った最後のオフセットを記録します。Kafka Connect の再起動時に、各コネクターの最後に記録されたオフセットを読み取り、最後に記録されたオフセットで各コネクターを起動します。コネクターを再起動すると、PostgreSQL サーバーにリクエストを送信し、その位置の直後に開始されるイベントを送信します。
PostgreSQL コネクターは、論理デコードプラグインによって送信されるイベントの一部としてスキーマ情報を取得します。ただし、コネクターはプライマリーキーが設定される列に関する情報を取得しません。コネクターは JDBC メタデータ (サイドチャネル) からこの情報を取得します。テーブルのプライマリーキー定義が変更される場合 (プライマリーキー列の追加、削除、または名前変更によって)、変更される場合、JDBC からのプライマリーキー情報が論理デコードプラグインが生成する変更イベントと同期されないごくわずかな期間が発生します。このごくわずかな期間に、キーの構造が不整合な状態でメッセージが作成される可能性があります。不整合にならないようにするには、以下のようにプライマリーキーの構造を更新します。
- データベースまたはアプリケーションを読み取り専用モードにします。
- Debezium に残りのイベントをすべて処理させます。
- Debezium を停止します。
- 関連するテーブルのプライマリーキー定義を更新します。
- データベースまたはアプリケーションを読み取り/書き込みモードにします。
- Debezium を再起動します。
PostgreSQL 10+ 論理デコードサポート (pgoutput
)
PostgreSQL 10+ の時点で、PostgreSQL でネイティブにサポートされる pgoutput
と呼ばれる論理レプリケーションストリームモードがあります。つまり、Debezium PostgreSQL コネクターは追加のプラグインを必要とせずにそのレプリケーションストリームを使用できます。これは、プラグインのインストールがサポートされないまたは許可されない環境で特に便利です。
詳細は、PostgreSQL の設定 を参照してください。
3.2.4. Debezium PostgreSQL の変更イベントレコードを受信する Kafka トピックのデフォルト名
PostgreSQL コネクターは、単一テーブルのすべての挿入、更新、および削除操作をのイベントを単一の Kafka トピックに書き込みます。デフォルトでは、serverName.schemaName.tableName です。
-
serverName は、
database.server.name
コネクター設定プロパティーで指定したコネクターの論理名です。 - SchemaName は、操作が発生したデータベーススキーマの名前です。
- tableName は、操作が発生したデータベーステーブルの名前です。
例えば、postgres
データベースと、products
、products_on_hand
、customers
、orders
の 4 つのテーブルを含む inventory
スキーマを持つ PostgreSQL インストレーションの変更をキャプチャするコネクターの設定において、fulfillment
が論理的なサーバー名であるとします。コネクターは以下の 4 つの Kafka トピックにレコードをストリーミングします。
-
fulfillment.inventory.products
-
fulfillment.inventory.products_on_hand
-
fulfillment.inventory.customers
-
fulfillment.inventory.orders
テーブルは特定のスキーマの一部ではなく、デフォルトの public
PostgreSQL スキーマで作成されたとします。Kafka トピックの名前は以下になります。
-
fulfillment.public.products
-
fulfillment.public.products_on_hand
-
fulfillment.public.customers
-
fulfillment.public.orders
3.2.5. Debezium PostgreSQL 変更イベントレコードのメタデータ
PostgreSQL コネクターによって生成された各レコードには、データベース変更イベント の他に、一部のメタデータも含まれています。メタデータには、サーバーでイベントが発生した場所、ソースパーティションの名前、イベントが置かれる Kafka トピックおよびパーティションの名前が含まれています。
"sourcePartition": { "server": "fulfillment" }, "sourceOffset": { "lsn": "24023128", "txId": "555", "ts_ms": "1482918357011" }, "kafkaPartition": null
-
source Partition
は、常にdatabase.server.name
コネクター設定プロパティーの設定をデフォルトとします。 sourceOffset
にはイベントが発生したサーバーの場所に関する情報が含まれています。-
lsn
はトランザクションログの PostgreSQL ログシーケンス番号 またはoffset
を表します。 -
txId
はイベント発生の原因となったサーバートランザクションの識別子を表します。 -
ts_ms
はトランザクションがコミットされたサーバー時間をエポックからの経過時間 (ミリ秒単位) で表します。
-
-
kafkaPartition
にnull
が設定されると、コネクターは特定の Kafka パーティションを使用しません。PostgreSQL コネクターは Kafka Connect パーティションを 1 つだけ使用し、生成されたイベントを 1 つの Kafka パーティションに配置します。
3.2.6. トランザクション境界を表す Debezium PostgreSQL コネクターによって生成されたイベント
Debezium は、トランザクション境界を表し、データ変更イベントメッセージをエンリッチするイベントを生成できます。Debezium はすべてのトランザクションの BEGIN
および END
に対して、以下のフィールドが含まれるイベントを生成します。
-
status
:BEGIN
またはEND
-
id
- 一意のトランザクション識別子の文字列表現。 -
event_count
(END
イベントの場合) -トランザクションによって出力されたイベントの合計数。 -
data_collections
(END
イベントの場合): 指定のデータコレクションからの変更によって出力されたイベントの数を提供するdata_collection
とevent_count
のペアの配列。
例
{ "status": "BEGIN", "id": "571", "event_count": null, "data_collections": null } { "status": "END", "id": "571", "event_count": 2, "data_collections": [ { "data_collection": "s1.a", "event_count": 1 }, { "data_collection": "s2.a", "event_count": 1 } ] }
トランザクションイベントは、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": "571", "total_order": "1", "data_collection_order": "1" } }
3.3. Debezium PostgreSQL コネクターのデータ変更イベントの説明
Debezium PostgreSQL コネクターは、行レベルの INSERT
、UPDATE
、および DELETE
操作ごとにデータ変更イベントを生成します。各イベントにはキーと値が含まれます。キーと値の構造は、変更されたテーブルによって異なります。
Debezium および Kafka Connect は、イベントメッセージの継続的なストリーム を中心として設計されています。ただし、これらのイベントの構造は時間の経過とともに変化する可能性があり、コンシューマーによる処理が困難になることがあります。これに対応するために、各イベントにはコンテンツのスキーマが含まれます。スキーマレジストリーを使用している場合は、コンシューマーがレジストリーからスキーマを取得するために使用できるスキーマ ID が含まれます。これにより、各イベントが自己完結型になります。
以下のスケルトン JSON は、変更イベントの基本となる 4 つの部分を示しています。ただし、アプリケーションで使用するために選択した Kafka Connect コンバーターの設定方法によって、変更イベントのこれら 4 部分の表現が決定されます。schema
フィールドは、変更イベントが生成されるようにコンバーターを設定した場合のみ変更イベントに含まれます。同様に、イベントキーおよびイベントペイロードは、変更イベントが生成されるようにコンバーターを設定した場合のみ変更イベントに含まれます。JSON コンバーターを使用し、変更イベントの基本となる 4 つの部分すべてを生成するように設定すると、変更イベントの構造は次のようになります。
{ "schema": { 1 ... }, "payload": { 2 ... }, "schema": { 3 ... }, "payload": { 4 ... }, }
項目 | フィールド名 | 説明 |
---|---|---|
1 |
|
最初の |
2 |
|
最初の |
3 |
|
2 つ目の |
4 |
|
2 つ目の |
デフォルトの動作では、コネクターによって、変更イベントレコードが イベントの元のテーブルと同じ名前を持つトピック にストリーミングされます。
Kafka 0.10 以降では、任意でイベントキーおよび値を タイムスタンプ とともに記録できます。このタイムスタンプはメッセージが作成された (プロデューサーによって記録) 時間または Kafka によってログに買い込まれた時間を示します。
PosgreSQL コネクターは、すべての Kafka Connect スキーマ名が Avro スキーマ名の形式 に準拠するようにします。つまり、論理サーバー名はアルファベットまたはアンダースコア (a-z、A-Z、または _) で始まる必要があります。論理サーバー名の残りの各文字と、スキーマ名とテーブル名の各文字は、アルファベット、数字、またはアンダースコア ( a-z、A-Z、0-9、または _) でなければなりません。無効な文字がある場合は、アンダースコアに置き換えられます。
論理サーバー名、スキーマ名、またはテーブル名に無効な文字が含まれ、名前を区別する唯一の文字が無効であると、無効な文字はすべてアンダースコアに置き換えられるため、予期せぬ競合が発生する可能性があります。
詳細は以下を参照してください。
3.3.1. Debezium PostgreSQL の変更イベントのキー
指定のテーブルでは、変更イベントのキーは、イベントが作成された時点でテーブルのプライマリーキーの各列のフィールドが含まれる構造を持ちます。また、テーブルの REPLICA IDENTITY
が FULL
または USING INDEX
に設定されている場合は、各ユニークキー制約のフィールドがあります。
public
データベーススキーマに定義されている customers
テーブルと、そのテーブルの変更イベントキーの例を見てみましょう。
テーブルの例
CREATE TABLE customers ( id SERIAL, first_name VARCHAR(255) NOT NULL, last_name VARCHAR(255) NOT NULL, email VARCHAR(255) NOT NULL, PRIMARY KEY(id) );
変更イベントキーの例
database.server.name
コネクター設定プロパティーに PostgreSQL_server
の値がある場合、この定義がある限り customers
テーブルの変更イベントはすべて同じキー構造を持ち、JSON では以下のようになります。
{ "schema": { 1 "type": "struct", "name": "PostgreSQL_server.public.customers.Key", 2 "optional": false, 3 "fields": [ 4 { "name": "id", "index": "0", "schema": { "type": "INT32", "optional": "false" } } ] }, "payload": { 5 "id": "1" }, }
項目 | フィールド名 | 説明 |
---|---|---|
1 |
|
キーのスキーマ部分は、キーの |
2 |
|
キーのペイロードの構造を定義するスキーマの名前。このスキーマは、変更されたテーブルのプライマリーキーの構造を記述します。キースキーマ名の形式は connector-name.database-name.table-name.
|
3 |
|
イベントキーの |
4 |
|
各フィールドの名前、インデックス、およびスキーマなど、 |
5 |
|
この変更イベントが生成された行のキーが含まれます。この例では、キーには値が |
column.exclude.list
および column.include.list
コネクター設定プロパティーは、テーブル列のサブセットのみをキャプチャーできるようにしますが、プライマリーキーまたは一意キーのすべての列は常にイベントのキーに含まれます。
テーブルにプライマリーキーまたは一意キーがない場合は、変更イベントのキーは null になります。プライマリーキーや一意キーの制約がないテーブルの行は一意に識別できません。
3.3.2. Debezium PostgreSQL 変更イベントの値
変更イベントの値はキーよりも若干複雑です。キーと同様に、値には schema
セクションと payload
セクションがあります。schema
セクションには、入れ子のフィールドを含む、 Envelope
セクションの payload
構造を記述するスキーマが含まれています。データを作成、更新、または削除する操作のすべての変更イベントには、Envelope 構造を持つ値 payload があります。
変更イベントキーの例を紹介するために使用した、同じサンプルテーブルについて考えてみましょう。
CREATE TABLE customers ( id SERIAL, first_name VARCHAR(255) NOT NULL, last_name VARCHAR(255) NOT NULL, email VARCHAR(255) NOT NULL, PRIMARY KEY(id) );
この表への変更に対する変更イベントの値は、REPLICA IDENTITY
設定およびイベントの目的である操作により異なります。
詳細は、以下を参照してください。
Replica identity
REPLICA IDENTITY は UPDATE
および DELETE
イベントの論理デコードプラグインで利用可能な情報量を決定する PostgreSQL 固有のテーブルレベルの設定です。具体的には、REPLICA IDENTITY
の設定は、UPDATE
または DELETE
イベントが発生するたびに、関係するテーブル列の以前の値に利用可能な情報 (ある場合) を制御します。
REPLICA IDENTITY
には 4 つの可能性があります。
DEFAULT
- テーブルにプライマリーキーがある場合に、UPDATE
およびDELETE
イベントにテーブルのプライマリーキー列の以前の値が含まれることがデフォルトの動作になります。UPDATE
イベントでは、値が変更されたプライマリーキー列のみが存在します。テーブルにプライマリーキーがない場合、コネクターはそのテーブルの
UPDATE
またはDELETE
イベントを出力しません。プライマリーキーのないテーブルの場合、コネクターは 作成 イベントのみを出力します。通常、プライマリーキーのないテーブルは、テーブルの最後にメッセージを追加するために使用されます。そのため、UPDATE
およびDELETE
イベントは便利ではありません。-
NOTHING
:UPDATE
およびDELETE
操作の出力されたイベントにはテーブル列の以前の値に関する情報は含まれません。 -
FULL
:UPDATE
およびDELETE
操作の出力されたイベントには、テーブルの列すべての以前の値が含まれます。 -
INDEX
index-name:UPDATE
およびDELETE
操作の発生したイベントには、指定されたインデックスに含まれる列の以前の値が含まれます。UPDATE
イベントには、更新された値を持つインデックス化された列も含まれます。
作成 イベント
以下の例は、customers
テーブルにデータを作成する操作に対して、コネクターによって生成される変更イベントの値の部分を示しています。
{ "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": "PostgreSQL_server.inventory.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": "PostgreSQL_server.inventory.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": "int64", "optional": true, "field": "txId" }, { "type": "int64", "optional": true, "field": "lsn" }, { "type": "int64", "optional": true, "field": "xmin" } ], "optional": false, "name": "io.debezium.connector.postgresql.Source", 3 "field": "source" }, { "type": "string", "optional": false, "field": "op" }, { "type": "int64", "optional": true, "field": "ts_ms" } ], "optional": false, "name": "PostgreSQL_server.inventory.customers.Envelope" 4 }, "payload": { 5 "before": null, 6 "after": { 7 "id": 1, "first_name": "Anne", "last_name": "Kretchmar", "email": "annek@noanswer.org" }, "source": { 8 "version": "1.4.2.Final", "connector": "postgresql", "name": "PostgreSQL_server", "ts_ms": 1559033904863, "snapshot": true, "db": "postgres", "schema": "public", "table": "customers", "txId": 555, "lsn": 24023128, "xmin": null }, "op": "c", 9 "ts_ms": 1559033904863 10 } }
項目 | フィールド名 | 説明 |
---|---|---|
1 |
| 値のペイロードの構造を記述する、値のスキーマ。変更イベントの値スキーマは、コネクターが特定のテーブルに生成するすべての変更イベントで同じになります。 |
2 |
|
|
3 |
|
|
4 |
|
|
5 |
|
値の実際のデータ。これは、変更イベントが提供する情報です。 |
6 |
|
イベント発生前の行の状態を指定する任意のフィールド。この例のように、 注記
このフィールドを利用できるかどうかは、各テーブルの |
7 |
|
イベント発生後の行の状態を指定する任意のフィールド。この例では、 |
8 |
| イベントのソースメタデータを記述する必須のフィールド。このフィールドには、イベントの発生元、イベントの発生順序、およびイベントが同じトランザクションの一部であるかどうかなど、このイベントと他のイベントを比較するために使用できる情報が含まれています。ソースメタデータには以下が含まれています。
|
9 |
|
コネクターによってイベントが生成される原因となった操作の型を記述する必須文字列。この例では、
|
10 |
|
コネクターがイベントを処理した時間を表示する任意のフィールド。この時間は、Kafka Connect タスクを実行している JVM のシステムクロックを基にします。 |
更新イベント
サンプル customers
テーブルにある更新の変更イベントの値には、そのテーブルの 作成 イベントと同じスキーマがあります。同様に、イベント値のペイロードは同じ構造を持ちます。ただし、イベント値ペイロードでは 更新 イベントに異なる値が含まれます。以下は、コネクターによって customers
テーブルでの更新に生成されるイベントの変更イベント値の例になります。
{ "schema": { ... }, "payload": { "before": { 1 "id": 1 }, "after": { 2 "id": 1, "first_name": "Anne Marie", "last_name": "Kretchmar", "email": "annek@noanswer.org" }, "source": { 3 "version": "1.4.2.Final", "connector": "postgresql", "name": "PostgreSQL_server", "ts_ms": 1559033904863, "snapshot": null, "db": "postgres", "schema": "public", "table": "customers", "txId": 556, "lsn": 24023128, "xmin": null }, "op": "u", 4 "ts_ms": 1465584025523 5 } }
項目 | フィールド名 | 説明 |
---|---|---|
1 |
|
データベースをコミットする前に行にあった値が含まれる任意のフィールド。この例では、テーブルの |
2 |
|
イベント発生後の行の状態を指定する任意のフィールド。この例では、 |
3 |
|
イベントのソースメタデータを記述する必須のフィールド。
|
4 |
|
操作の型を記述する必須の文字列。更新 イベントの値では、 |
5 |
|
コネクターがイベントを処理した時間を表示する任意のフィールド。この時間は、Kafka Connect タスクを実行している JVM のシステムクロックを基にします。 |
行のプライマリーキー/一意キーの列を更新すると、行のキーの値が変更されます。キーが変更されると、3 つのイベントが Debezium によって出力されます。3 つのイベントとは、DELETE
イベント、行の古いキーを持つ 廃棄 (tombstone)、およびそれに続く行の新しいキーを持つイベントです。詳細は次のセクションで説明します。
プライマリーキーの更新
行のプライマリーキーフィールドを変更する UPDATE
操作は、プライマリーキーの変更と呼ばれます。プライマリーキーの変更では、UPDATE
イベントレコードの代わりにコネクターが古いキーの DELETE
イベントレコードと、新しい (更新された) キーの CREATE
イベントレコードを出力します。これらのイベントには通常の構造と内容があり、イベントごとにプライマリーキーの変更に関連するメッセージヘッダーがあります。
-
DELETE
イベントレコードには、メッセージヘッダーとして__debezium.newkey
が含まれます。このヘッダーの値は、更新された行の新しいプライマリーキーです。 -
CREATE
イベントレコードには、メッセージヘッダーとして__debezium.oldkey
が含まれます。このヘッダーの値は、更新された行にあった以前の (古い) プライマリーキーです。
削除 イベント
削除 変更イベントの値は、同じテーブルの 作成 および 更新 イベントと同じ schema
の部分になります。サンプル customers
テーブルの 削除 イベントの payload
部分は以下のようになります。
{ "schema": { ... }, "payload": { "before": { 1 "id": 1 }, "after": null, 2 "source": { 3 "version": "1.4.2.Final", "connector": "postgresql", "name": "PostgreSQL_server", "ts_ms": 1559033904863, "snapshot": null, "db": "postgres", "schema": "public", "table": "customers", "txId": 556, "lsn": 46523128, "xmin": null }, "op": "d", 4 "ts_ms": 1465581902461 5 } }
項目 | フィールド名 | 説明 |
---|---|---|
1 |
|
イベント発生前の行の状態を指定する任意のフィールド。削除 イベント値の |
2 |
|
イベント発生後の行の状態を指定する任意のフィールド。削除 イベント値の |
3 |
|
イベントのソースメタデータを記述する必須のフィールド。削除 イベント値の
|
4 |
|
操作の型を記述する必須の文字列。 |
5 |
|
コネクターがイベントを処理した時間を表示する任意のフィールド。この時間は、Kafka Connect タスクを実行している JVM のシステムクロックを基にします。 |
削除 変更イベントレコードは、この行の削除を処理するために必要な情報を持つコンシューマーを提供します。
プライマリーキーを持たないテーブルに対して生成された 削除 イベントをコンシューマーが処理できるようにするには、テーブルの REPLICA IDENTITY
を FULL
に設定します。テーブルに主キーがなく、テーブルの REPLICA IDENTITY
が DEFAULT
または NOTHING
に設定されている場合、削除 イベントの before
フィールドはありません。
PostgreSQL コネクターイベントは、Kafka のログコンパクション と動作するように設計されています。ログコンパクションにより、少なくとも各キーの最新のメッセージが保持される限り、一部の古いメッセージを削除できます。これにより、トピックに完全なデータセットが含まれ、キーベースの状態のリロードに使用できるようにするとともに、Kafka がストレージ領域を確保できるようにします。
廃棄 (tombstone) イベント
行が削除された場合でも、Kafka は同じキーを持つ以前のメッセージをすべて削除できるため、削除 イベントの値はログコンパクションで動作します。ただし、Kafka が同じキーを持つすべてのメッセージを削除するには、メッセージの値が null
である必要があります。これを可能にするには、PostgreSQL コネクターは、値が null
値以外の同じキーを持つ特別な 廃棄 イベントが含まれる 削除 イベントに従います。
切り捨て (truncate) イベント
切り捨て (truncate) 変更イベントは、テーブルが切り捨てられていることを伝えます。この場合のメッセージキーは null
で、メッセージの値は以下のようになります。
{ "schema": { ... }, "payload": { "source": { 1 "version": "1.4.2.Final", "connector": "postgresql", "name": "PostgreSQL_server", "ts_ms": 1559033904863, "snapshot": false, "db": "postgres", "schema": "public", "table": "customers", "txId": 556, "lsn": 46523128, "xmin": null }, "op": "t", 2 "ts_ms": 1559033904961 3 } }
項目 | フィールド名 | 説明 |
---|---|---|
1 |
|
イベントのソースメタデータを記述する必須のフィールド。切り捨て (truncate) イベント値の
|
2 |
|
操作の型を記述する必須の文字列。 |
3 |
|
コネクターがイベントを処理した時間を表示する任意のフィールド。この時間は、Kafka Connect タスクを実行している JVM のシステムクロックを基にします。 |
1 つの TRUNCATE
ステートメントが複数のテーブルに適用された場合、切り捨てられたテーブルごとに 1 つの切り捨て (truncate) 変更イベントレコードが出力されます。
切り捨て (truncate) イベントは、テーブル全体に加えた変更を表し、メッセージキーを持たないので、単一のパーティションを持つトピックを使用しない限り、テーブルに関する変更イベント (作成、更新 など) とそのテーブルの 切り捨て (truncate) イベントの順番は保証されません。たとえば、これらのイベントが異なるパーティションから読み取られる場合、コンシューマーは 更新 イベントを 切り捨て (truncate) イベントの後でのみ受け取る可能性があります。
3.4. Debezium PostgreSQL コネクターによるデータ型のマッピング方法
PostgreSQL コネクターは、行が存在するテーブルのように構造化されたイベントで行への変更を表します。イベントには、各列の値のフィールドが含まれます。その値がどのようにイベントで示されるかは、列の PostgreSQL のデータ型によって異なります。以下のセクションでは、PostgreSQL データ型をイベントフィールドの リテラル型 および セマンティック型にマッピングする方法を説明します。
-
literal type は、Kafka Connect スキーマタイプ (
INT8
、INT16
、INT32
、INT64
、FLOAT32
、FLOAT64
、BOOLEAN
、STRING
、BYTES
、ARRAY
、MAP
、STRUCT
) を使用して、値がどのように表現されるかを記述します。 - セマンティック型 は、フィールドの Kafka Connect スキーマの名前を使用して、Kafka Connect スキーマがフィールドの 意味 をキャプチャーする方法を記述します。
詳細は以下を参照してください。
基本型
以下の表は、コネクターによる基本型へのマッピング方法を説明しています。
PostgreSQL のデータ型 | リテラル型 (スキーマ型) | セマンティック型 (スキーマ名) および注記 |
---|---|---|
|
| 該当なし |
|
| 該当なし |
|
|
|
|
|
|
|
| 該当なし |
|
| 該当なし |
|
| 該当なし |
|
| 該当なし |
|
| 該当なし |
|
| 該当なし |
|
| 該当なし |
|
| 該当なし |
|
| 該当なし |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
該当なし |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| 該当なし |
|
| 該当なし |
|
|
該当なし |
|
|
n/a |
|
|
n/a |
|
|
該当なし |
|
|
該当なし |
|
|
該当なし |
|
|
|
Temporal (一時) 型
タイムゾーン情報が含まれる PostgreSQL の TIMESTAMPTZ
and TIMETZ
データ型以外に、時間型がマッピングされる仕組みは time.precision.mode
コネクター設定プロパティーの値によって異なります。ここでは、以下のマッピングについて説明します。
time.precision.mode=adaptive
time.precision.mode
プロパティーがデフォルトの adaptive
に設定された場合、コネクターは列のデータ型定義に基づいてリテラル型とセマンティック型を決定します。これにより、イベントがデータベースの値を 正確 に表すようになります。
PostgreSQL のデータ型 | リテラル型 (スキーマ型) | セマンティック型 (スキーマ名) および注記 |
---|---|---|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
time.precision.mode=adaptive_time_microseconds
time.precision.mode
設定プロパティーが adaptive_time_microseconds
に設定されている場合には、コネクターは列のデータ型定義に基づいて一時的な型のリテラル型とセマンティック型を決定します。これにより、マイクロ秒としてキャプチャーされた TIME
フィールド以外は、イベントがデータベースの値を 正確 に表すようになります。
PostgreSQL のデータ型 | リテラル型 (スキーマ型) | セマンティック型 (スキーマ名) および注記 |
---|---|---|
|
|
|
|
|
|
|
|
|
|
|
|
time.precision.mode=connect
time.precision.mode
設定プロパティーが connect
に設定された場合、コネクターは Kafka Connect の論理型を使用します。これは、コンシューマーが組み込みの Kafka Connect の論理型のみを処理でき、可変精度の時間値を処理できない場合に便利です。ただし、PostgreSQL はマイクロ秒の精度をサポートするため、 connect
時間精度を指定してコネクターによって生成されたイベントは、データベース列の少数秒の精度値が 3 よりも大きい場合に、精度が失われます。
PostgreSQL のデータ型 | リテラル型 (スキーマ型) | セマンティック型 (スキーマ名) および注記 |
---|---|---|
|
|
|
|
|
|
|
|
|
TIMESTAMP 型
TIMESTAMP
型は、タイムゾーン情報のないタイムスタンプを表します。このような列は、UTC を基にして同等の Kafka Connect 値に変換されます。例えば、time.precision.mode
がconnect
に設定されていない場合、TIMESTAMP
値 2018-06-20 15:13:16.945104 は、io.debezium.time.Micro Timestamp
の値 1529507596945104 で表されます。
Kafka Connect および Debezium を実行している JVM のタイムゾーンは、この変換には影響しません。
10 進数型
PostgreSQL コネクター設定プロパティーの設定 decimal.handling.mode
は、コネクターが 10 進数型をマッピングする方法を決定します。
decimal.handling.mode
プロパティーが precise
に設定されている場合には コネクターはDECIMAL
と NUMERIC
列すべてに Kafka Connect org.apache.kafka.connect.data.Decimal
論理型を使用します。これはデフォルトのモードです。
PostgreSQL のデータ型 | リテラル型 (スキーマ型) | セマンティック型 (スキーマ名) および注記 |
---|---|---|
|
|
|
|
|
|
このルールには例外があります。スケーリング制約なしで NUMERIC
または DECIMAL
型が使用されると、データベースから取得される値のスケールは値ごとに異なります (可変)。この場合、コネクターは io.debezium.data.Variable Scale Decimal
を使用し、これには転送された値とスケールの両方が含まれます。
PostgreSQL のデータ型 | リテラル型 (スキーマ型) | セマンティック型 (スキーマ名) および注記 |
---|---|---|
|
|
|
|
|
|
decimal.handling.mode
プロパティーが double
に設定されている場合、コネクターはすべての DECIMAL
および NUMERIC
値を Java の double 値として表し、次の表のようにエンコードします。
PostgreSQL のデータ型 | リテラル型 (スキーマ型) | セマンティック型 (スキーマ名) |
---|---|---|
|
| |
|
|
decimal.handling.mode
設定プロパティーの最後の設定は string
です。この場合、コネクターは DECIMAL
および NUMERIC
値をフォーマットされた文字列表現として表し、それらを以下の表のようにエンコードします。
PostgreSQL のデータ型 | リテラル型 (スキーマ型) | セマンティック型 (スキーマ名) |
---|---|---|
|
| |
|
|
Postgre SQL は、decimal.handling.mode
の設定が string
または double
の場合、DECIMAL
/NUMERIC
値に格納される特別な値として Na N
(not a number) をサポートしています。この場合、コネクターは NaN
をDouble.NaN
または文字列定数 NAN
のいずれかとしてエンコードします。
HSTORE 型
dhstore.handling.mode
コネクター設定プロパティーが json
(デフォルト) に設定されている場合、コネクターは HSTORE
値を JSON 値の文字列表現として表し、以下の表で示すようにエンコードします。hstore.handling.mode
プロパティーが map
に設定されている場合、コネクターは HSTORE
値に MAP
スキーマタイプを使用します。
PostgreSQL のデータ型 | リテラル型 (スキーマ型) | セマンティック型 (スキーマ名) および注記 |
---|---|---|
|
|
|
|
|
該当なし |
ドメイン型
PostgreSQL は、他の基礎となるタイプに基づいたユーザー定義の型をサポートします。このような列型を使用すると、Debezium は完全な型階層に基づいて列の表現を公開します。
PostgreSQL ドメイン型を使用する列で変更をキャプチャーするには、特別に考慮する必要があります。デフォルトデータベース型の 1 つを拡張するドメインタイプと、カスタムの長さまたはスケールを定義するドメインタイプが含まれるように列が定義されると、生成されたスキーマは定義されたその長さとスケールを継承します。
カスタムの長さまたはスケールを定義するドメインタイプを拡張する別のドメインタイプが含まれるように列が定義されていると、その情報は PostgreSQL ドライバーの列メタデータにはないため、生成されたスキーマは定義された長さやスケールを継承 しません。
ネットワークアドレス型
PostgreSQL には、IPv4、IPv6、および MAC アドレスを保存できるデータ型があります。ネットワークアドレスの格納には、プレーンテキスト型ではなくこの型を使用することが推奨されます。ネットワークアドレス型は、入力エラーチェックと特化した演算子および関数を提供します。
PostgreSQL のデータ型 | リテラル型 (スキーマ型) | セマンティック型 (スキーマ名) および注記 |
---|---|---|
|
|
該当なし |
|
|
該当なし |
|
|
該当なし |
|
|
該当なし |
PostGIS タイプ
PostgreSQL コネクターは、すべての PostGIS データ型 をサポートします。
PostGIS データ型 | リテラル型 (スキーマ型) | セマンティック型 (スキーマ名) および注記 |
---|---|---|
|
|
詳細は、Open Geospatial Consortium Simple Features Access を参照してください。 |
|
|
詳細は、Open Geospatial Consortium Simple Features Access を参照してください。 |
TOAST 化された値
PostgreSQL ではページサイズにハード制限があります。つまり、約 8 KB を超える値は link::https://www.postgresql.org/docs/current/storage-toast.html[TOAST ストレージ] を使用して保存する必要があります。これは、データベースからのレプリケーションメッセージに影響します。TOAST メカニズムを使用して保存され、変更されていない値は、テーブルのレプリカ ID の一部でない限り、メッセージに含まれません。競合が発生する可能性があるため、Debezium が不足している値を直接データベースから読み取る安全な方法はありません。そのため、Debezium は以下のルールに従って、TOAST 化された値を処理します。
-
REPLICA IDENTITY FULL
- TOAST 列の値を持つテーブルは、他の列と同様に変更イベントのbefore
およびafter
フィールドの一部となります。 -
REPLICA IDENTITY DEFAULT
のあるテーブル - データベースからUPDATE
イベントを受信すると、レプリカ ID の一部ではない変更されていない TOAST 列値はイベントに含まれません。同様に、DELETE
イベントを受信するときに TOAST 列 (ある場合) はbefore
フィールドにありません。この場合、Debezium は列値を安全に提供できないため、コネクターはコネクター設定プロパティーで定義されたプレースホルダー値 (toasted.value.placeholder
) を返します。
3.5. Debezium コネクターを実行するための PostgreSQL の設定
本リリースの Debezium では、ネイティブの pgoutput
論理レプリケーションストリームのみがサポートされます。pgoutput
プラグインを使用するように PostgreSQL を設定するには、レプリケーションスロットを有効にし、レプリケーションの実行に必要な権限を持つユーザーを設定します。
詳細は以下を参照してください。
3.5.1. Debezium pgoutput
プラグインのレプリケーションスロットの設定
PostgreSQL の論理デコード機能はレプリケーションスロットを使用します。レプリケーションスロットを設定するには、postgresql.conf
ファイルに以下を指定します。
wal_level=logical max_wal_senders=1 max_replication_slots=1
これらの設定は、PostgreSQL サーバーを以下のように指示します。
-
wal_level
- 先行書き込みログで論理デコードを使用します。 -
max_wal_senders
- WAL 変更の処理に、1 つの個別プロセスの最大を使用します。 -
max_replication_slots
- WAL の変更をストリーミングするために作成される 1 つのレプリケーションスロットの最大を許可します。
レプリケーションスロットは、Debezium の停止中でも Debezium に必要なすべての WAL エントリーを保持することが保証されいます。したがって、以下の点を避けるために、レプリケーションスロットを注意して監視することが重要になります。
- 過剰なディスク消費量。
- レプリケーションスロットが長期間使用されないと発生する可能性がある、あらゆる状態 (カタログの肥大化など)。
詳細は、レプリケーションスロットに関する PostgreSQL のドキュメント を参照してください。
PostgreSQL ログ先行書き込みの設定 や仕組みを理解していると、Debezium PostgreSQL コネクターを使用する場合に役立ちます。
3.5.2. Debezium コネクターの PostgreSQL パーミッションの設定
PostgreSQL サーバーを設定して Debezium コネクターを実行するには、レプリケーションを実行できるデータベースユーザーが必要です。レプリケーションは、適切なパーミッションを持つデータベースユーザーのみが実行でき、設定された数のホストに対してのみ実行できます。
セキュリティーで説明されているように、スーパーユーザーはデフォルトで必要な REPLICATION
および LOGIN
ロールを持っていますが、Debezium レプリケーションユーザーの権限を昇格しないことが推奨されます。代わりに、必要最低限の特権を持つ Debezium ユーザーを作成します。
前提条件
- PostgreSQL の管理者権限。
手順
ユーザーにレプリケーションの権限を付与するには、少なくとも
REPLICATION
およびLOGIN
権限を持つ PostgreSQL ロールを定義し、そのロールをユーザーに付与します。以下に例を示します。CREATE ROLE <name> REPLICATION LOGIN;
3.5.3. Debezium が PostgreSQL パブリケーションを作成できるように権限を設定
Debezium は、PostgreSQL ソーステーブルの変更イベントを、テーブル用に作成された パブリケーション からストリーミングします。パブリケーションには、1 つ以上のテーブルから生成される変更イベントのフィルターされたセットが含まれます。各パブリケーションのデータは、パブリケーションの仕様に基づいてフィルターされます。この仕様は、PostgreSQL データベース管理者または Debezium コネクターが作成できます。Debezium PostgreSQL コネクターに、パブリケーションの作成やレプリケートするデータの指定を許可するには、コネクターはデータベースで特定の権限で操作する必要があります。
パブリケーションの作成方法を決定するオプションは複数あります。通常、コネクターを設定する前に、キャプチャーするテーブルのパブリケーションを手動で作成することが推奨されます。しかし、Debezium がパブリケーションを自動的に作成し、それに追加するデータを指定できるように、ご使用の環境を設定できます。
Debezium は include list および exclude list プロパティーを使用して、データがパブリケーションに挿入される方法を指定します。Debezium がパブリケーションを作成できるようにするオプションの詳細は、publication.autocreate.mode
を参照してください。
Debezium が PostgreSQL パブリケーションを作成するには、以下の権限を持つユーザーとして実行する必要があります。
- パブリケーションにテーブルを追加するためのデータベースのレプリケーション権限。
-
パブリケーションを追加するためのデータベースの
CREATE
権限。 -
最初のテーブルデータをコピーするためのテーブルの
SELECT
権限。テーブルの所有者には、テーブルに対するSELECT
権限が自動的に付与されます。
テーブルをパブリケーションに追加する場合は、ユーザーはテーブルの所有者になります。ただし、ソーステーブルはすでに存在するため、元の所有者と所有権を共有する仕組みが必要です。共有所有権を有効にするには、PostgreSQL レプリケーショングループを作成した後、既存のテーブルの所有者とレプリケーションユーザーをそのグループに追加します。
手順
レプリケーショングループを作成します。
CREATE ROLE <replication_group>;
テーブルの元の所有者をグループに追加します。
GRANT REPLICATION_GROUP TO <original_owner>;
Debezium レプリケーションユーザーをグループに追加します。
GRANT REPLICATION_GROUP TO <replication_user>;
テーブルの所有権を
<replication_group>
に移します。ALTER TABLE <table_name> OWNER TO REPLICATION_GROUP;
Debezium がキャプチャ設定を指定するためには、の値が publication.autocreate.mode
を filtered
に設定する必要があります。
3.5.4. Debezium コネクターホストでのレプリケーションを許可するように PostgreSQL を設定
Debezium による PostgreSQL データのレプリケーションを可能にするには、データベースを設定し、PostgreSQL コネクターを実行するホストでのレプリケーションを許可する必要があります。データベースとのレプリケーションが許可されるクライアントを指定するには、エントリーを PostgreSQL ホストベースの認証ファイル pg_hba.conf
に追加します。pg_hba.conf
ファイルの詳細は、the PostgreSQL のドキュメントを参照してください。
手順
pg_hba.conf
ファイルにエントリーを追加して、データベースホストでレプリケートできる Debezium コネクターホストを指定します。以下に例を示します。pg_hba.conf
ファイルの例です。local replication <youruser> trust 1 host replication <youruser> 127.0.0.1/32 trust 2 host replication <youruser> ::1/128 trust 3
ネットワークマスクの詳細は、PostgreSQL のドキュメント を参照してください。
3.5.5. Debezium WAL ディスク領域の消費を管理するための PostgreSQL の設定
場合によっては、WAL ファイルによって使用される PostgreSQL ディスク領域が、異常に急上昇したり増加することがあります。このような場合、いくつかの理由が考えられます。
コネクターがデータを受信した最大の LSN は、サーバーの
pg_replication_slots
ビューのconfirmed_flush_lsn
列で確認できます。この LSN よりも古いデータは利用できず、データベースがディスク領域を解放します。また、
pg_replication_slots
ビューのrestart_lsn
列には、コネクターが必要とする可能性のある最も古い WAL の LSN が含まれています。confirmed_flush_lsn
の値が定期的に増加し、restart_lsn
の値に遅延が発生する場合は、データベースは領域を解放する必要があります。データベースは、通常バッチブロックでディスク領域を解放します。これは想定内の動作であり、ユーザーによるアクションは必要ありません。
-
追跡されるデータベースには多くの更新がありますが、一部の更新のみがコネクターの変更をキャプチャーするテーブルおよびスキーマに関連します。この状況は、定期的なハートビートイベントで簡単に解決できます。コネクターの
heartbeat.interval.ms
コネクター設定プロパティーを設定します。 PostgreSQL インスタンスには複数のデータベースが含まれ、その 1 つがトラフィックが多いデータベースです。Debezium は、他のデータベースと比較して、トラフィックが少ない別のデータベースで変更をキャプチャーします。レプリケーションスロットがデータベースごとに機能し、Debezium が呼び出しされないため、Debezium は LSN を確認できません。WAL はすべてのデータベースで共有されているため、Debezium が変更をキャプチャーするデータベースによってイベントが出力されるまで、使用量が増加する傾向にあります。これに対応するには、以下を行う必要があります。
-
heartbeat.interval.ms
コネクター設定プロパティーを使用して、定期的なハートビートレコードの生成を有効にします。 - Debezium が変更をキャプチャーするデータベースから変更イベントを定期的に送信します。
新しい行を挿入したり、同じ行を定期的に更新することで、別のプロセスがテーブルを定期的に更新します。次に PostgreSQL は Debezium を呼び出して、最新の LSN を確認し、データベースが WAL 領域を解放できるようにします。このタスクは、
heartbeat.action.query
コネクター設定プロパティーを使用して自動化できます。-
3.6. Debezium PostgreSQL コネクターのデプロイメント
Debezium PostgreSQL コネクターをデプロイするには、コネクターファイルを Kafka Connect に追加し、コネクターを実行するカスタムコンテナーを作成して、コネクター設定をコンテナーに追加します。詳細は以下を参照してください。
- https://access.redhat.com/documentation/ja-jp/red_hat_integration/2021.Q1/html-single/debezium_user_guide/index#deploying-debezium-postgresql-connectors
- https://access.redhat.com/documentation/ja-jp/red_hat_integration/2021.Q1/html-single/debezium_user_guide/index#descriptions-of-debezium-postgresql-connector-configuration-properties
3.6.1. Debezium PostgreSQL コネクターのデプロイ
Debezium PostgreSQL コネクターをデプロイするには、Debezium コネクターアーカイブが含まれるカスタム Kafka Connect コンテナーイメージをビルドし、このコンテナーイメージをコンテナーレジストリーにプッシュする必要があります。次に、2 つのカスタムリソース (CR) を作成する必要があります。
-
Kafka Connector を設定し、Debezium コネクターを実行するために作成したイメージの名前を指定する
KafkaConnect
CR。この CR を OpenShift Kafka インスタンスに適用します。 -
Debezium PostgreSQL コネクターを設定する
KafkaConnector
CR。この CR を、Red Hat AMQ Streams がデプロイされている OpenShift インスタンスに適用します。
前提条件
- PostgreSQL が実行され、PostgreSQL を設定して Debezium コネクターを実行 する手順が実行済みである。
- Red Hat AMQ Streams を使用して、OpenShift で Apache Kafka および Kafka Connect を設定し、実行済みである。AMQ Streams は、Kafka を OpenShift に取り入れる operator およびイメージを提供します。
- Podman または Docker がインストールされている。
-
Debezium コネクターを実行するコンテナーを追加する予定のコンテナーレジストリー (
quay.io
やdocker.io
など) でコンテナーを作成および管理するアカウントとパーミッションを持っている。
手順
Kafka Connect の Debezium PostgreSQL コンテナーを作成します。
- Debezium PostgreSQL コネクターアーカイブ をダウンロードします。
Debezium PostgreSQL コネクターアーカイブを展開して、コネクタープラグインのディレクトリー構造を作成します。以下に例を示します。
./my-plugins/ ├── debezium-connector-postgresql │ ├── ...
registry.redhat.io/amq7/amq-streams-kafka-26-rhel7:1.6.0
をベースイメージとして使用する Docker ファイルを作成します。たとえば、ターミナルウィンドウに以下を入力します。cat <<EOF >debezium-container-for-postgresql.yaml 1 FROM {DockerKafkaConnect} USER root:root COPY ./my-plugins/ /opt/kafka/plugins/ 2 USER 1001 EOF
このコマンドは、現在のディレクトリーに
debezium-container-for-postgresql.yaml
という名前の Docker ファイルを作成します。前のステップで作成した
debezium-container-for-postgresql.yaml
Docker ファイルからコンテナーイメージをビルドします。ファイルが含まれるディレクトリーから、以下のコマンドを実行します。podman build -t debezium-container-for-postgresql:latest .
docker build -t debezium-container-for-postgresql:latest .
build
コマンドは、debezium-container-for-postgresql
という名前のコンテナーイメージを構築します。カスタムイメージを
quay.io
などのコンテナーレジストリーまたは内部のコンテナーレジストリーにプッシュします。このレジストリーが OpenShift インスタンスからアクセス可能であることを確認します。以下に例を示します。podman push debezium-container-for-postgresql:latest
新しい Debezium PostgreSQL
KafkaConnect
カスタムリソース (CR) を作成します。たとえば、以下の例のようにannotations
およびimage
プロパティーを指定するdbz-connect.yaml
という名前のKafkaConnect
CR を作成します。apiVersion: kafka.strimzi.io/v1beta1 kind: KafkaConnect metadata: name: my-connect-cluster annotations: strimzi.io/use-connector-resources: "true" 1 spec: image: debezium-container-for-postgresql 2
以下のコマンドを実行して、
KafkaConnect
CR を OpenShift Kafka インスタンスに適用します。oc create -f dbz-connect.yaml
これにより、OpenShift の Kafka Connect 環境が更新され、Debezium コネクターを実行するために作成したイメージの名前を指定する Kafka Connector インスタンスが追加されます。
Debezium PostgreSQL コネクターインスタンスを設定する
KafkaConnector
カスタムリソースを作成します。コネクター設定プロパティーを設定する
.yaml
ファイルに Debezium PostgreSQL コネクターを設定します。コネクター設定は、Debezium に対して、スキーマおよびテーブルのサブセットにイベントを生成するよう指示する可能性があり、または機密性の高い、大きすぎる、または不必要な指定のコラムで Debezium が値を無視、マスク、または切り捨てするようにプロパティーを設定する可能性もあります。これらの設定で指定できる PostgreSQL コネクタープロパティーの完全リスト を参照してください。以下の例では、ポート
5432
で PostgreSQL サーバーホスト192.168.99.100
に接続する Debezium コネクターを設定します。このホストには、sampledb
という名前のデータベース、public
という名前のスキーマがあり、fulfillment
はサーバーの論理名です。fulfillment-connector.yaml
apiVersion: kafka.strimzi.io/v1beta1 kind: KafkaConnector metadata: name: fulfillment-connector 1 labels: strimzi.io/cluster: my-connect-cluster spec: class: io.debezium.connector.postgresql.PostgresConnector tasksMax: 1 2 config: 3 database.hostname: 192.168.99.100 4 database.port: 5432 database.user: debezium database.password: dbz database.dbname: sampledb database.server.name: fulfillment 5 schema.include.list: public 6 plugin.name: pgoutput 7
- 1
- コネクターの名前。
- 2
- 1 度に 1 つのタスクのみが動作する必要があります。PostgreSQL コネクターは PostgreSQL サーバーの
192.168.99.100
を読み取るため、単一のコネクタータスクを使用することで、順序とイベントの処理が適切に行われるようになります。Kafka Connect サービスはコネクターを使用して作業を行う 1 つ以上のタスクを開始し、実行中のタスクを自動的に Kafka Connect サービスのクラスター全体に分散します。いずれかのサービスが停止またはクラッシュすると、これらのタスクは稼働中のサービスに再分散されます。 - 3
- コネクターの設定。
- 4
- PostgreSQL サーバーを実行しているデータベースホストの名前。この例では、データベースのホスト名は
192.168.99.100
です。 - 5
- 一意のサーバー名。サーバー名は、PostgreSQL サーバーまたはサーバーのクラスターの論理識別子です。この名前は、変更イベントレコードを受信するすべての Kafka トピックの接頭辞として使用されます。
- 6
- コネクターは
public
スキーマでのみ変更をキャプチャーします。選択したテーブルでのみ変更をキャプチャーするようにコネクターを設定できます。table.include.list
コネクター設定プロパティーを参照してください。 - 7
- PostgreSQL サーバーにインストールされている PostgreSQL 論理デコードプラグイン の名前。Postgre SQL 10 以降でサポートされている値は
pgoutput
のみですが、明示的にplugin.name
をpgoutput
に設定する必要があります。
Kafka Connect でコネクターインスタンスを作成します。たとえば、
KafkaConnector
リソースをfulfillment-connector.yaml
ファイルに保存した場合は、以下のコマンドを実行します。oc apply -f fulfillment-connector.yaml
このコマンドは
meetment-connector
を登録して、コネクターがKafkaConnector
CR に定義されているsampledb
データベースに対して実行を開始します。コネクターが作成され、起動されたことを確認します。
Kafka Connect ログ出力を表示して、コネクターが作成され、指定データベースの変更のキャプチャーが開始されたことを確認します。
oc logs $(oc get pods -o name -l strimzi.io/cluster=my-connect-cluster)
ログの出力を確認し、初回のスナップショットが実行されたことを確認します。以下のような出力が表示されるはずです。
... INFO Starting snapshot for ... ... INFO Snapshot is using user 'debezium' ...
コネクターがエラーがなく正常に起動すると、コネクターが変更をキャプチャーする各テーブルのトピックが作成されます。CR のサンプルでは、
public
スキーマの各テーブルにトピックがあります。ダウンストリームアプリケーションは、これらのトピックをサブスクライブできます。以下のコマンドを実行して、コネクターによってトピックが作成されたことを検証します。
oc get kafkatopics
結果
コネクターが起動すると、コネクターが設定された PostgreSQL サーバーデータベースの 整合性スナップショットが実行 されます。その後、コネクターは行レベルの操作のデータ変更イベントの生成を開始し、変更イベントレコードを Kafka トピックにストリーミングします。
3.6.2. Debezium PostgreSQL コネクター設定プロパティーの説明
Debezium PostgreSQL コネクターには、アプリケーションに適したコネクター動作を実現するために使用できる設定プロパティーが多数あります。多くのプロパティーにはデフォルト値があります。プロパティーに関する情報は、以下のように設定されています。
以下の設定プロパティーは、デフォルト値がない場合は必須です。
プロパティー | デフォルト | 説明 |
---|---|---|
コネクターの一意名。同じ名前で再登録を試みると失敗します。このプロパティーはすべての Kafka Connect コネクターに必要です。 | ||
コネクターの Java クラスの名前。Postgre SQL コネクターには、常に | ||
| このコネクターのために作成する必要のあるタスクの最大数。PostgreSQL コネクターは常に単一のタスクを使用するため、この値を使用しません。そのため、デフォルト値は常に許容されます。 | |
| PostgreSQL サーバーにインストールされている PostgreSQL 論理デコードプラグイン の名前。
サポートされている値は | |
| 特定のデータベース/スキーマの特定のプラグインから変更をストリーミングするために作成された PostgreSQL 論理デコードスロットの名前。サーバーはこのスロットを使用して、設定する Debezium コネクターにイベントをストリーミングします。 スロット名は PostgreSQL レプリケーションスロットの命名ルール に準拠する必要があり、命名ルールには各レプリケーションスロットには名前が付けられ、名前にはアルファベットの小文字、数字、およびアンダースコアを使用できます。と記載されています。 | |
| コネクターが正常に想定されるように停止した場合に論理レプリケーションスロットを削除するかどうか。デフォルトの動作では、コネクターが停止したときにレプリケーションスロットはコネクターに設定された状態を保持します。コネクターが再起動すると、同じレプリケーションスロットがあるため、コネクターは停止した場所から処理を開始できます。
テストまたは開発環境でのみ | |
|
このパブリケーションが存在しない場合は起動時に作成され、すべてのテーブルが含まれます。Debezium は、設定されている場合は、独自の include/exclude リストフィルターを適用し、対象となる特定のテーブルのイベントのみをパブリケーションが変更するように制限します。コネクターユーザーがこのパブリケーションを作成するには、スーパーユーザーの権限が必要であるため、通常はコネクターを初めて開始する前にパブリケーションを作成することをお勧めします。 パブリケーションがすでに存在し、すべてのテーブルが含まれてているか、テーブルのサブセットで設定されている場合、Debezium は定義されているようにパブリケーションを使用します。 | |
PostgreSQL データベースサーバーの IP アドレスまたはホスト名。 | ||
| PostgreSQL データベースサーバーのポート番号 (整数)。 | |
PostgreSQL データベースサーバーに接続するための PostgreSQL データベースユーザーの名前。 | ||
PostgreSQL データベースサーバーへの接続時に使用するパスワード。 | ||
変更をストリーミングする PostgreSQL データベースの名前。 | ||
Debezium が変更をキャプチャーする特定の PostgreSQL データベースサーバーまたはクラスターの namespace を識別および提供する論理名。データベースサーバーの論理名には英数字とアンダースコアのみを使用する必要があります。論理名は、他のコネクター全体で一意となる必要があります。これは、このコネクターからレコードを受信するすべての Kafka トピックのトピック名接頭辞として使用されるためです。 | ||
変更をキャプチャーする対象とするスキーマの名前と一致する正規表現のコンマ区切りリスト (任意)。 | ||
変更をキャプチャーする対象としないスキーマの名前と一致する正規表現のコンマ区切りリスト (任意)。システムスキーマ以外で、 | ||
変更をキャプチャーするテーブルの完全修飾テーブル識別子と一致する正規表現のコンマ区切りリスト (任意)。 | ||
変更をキャプチャーしないテーブルの完全修飾テーブル識別子と一致する正規表現のコンマ区切りリスト (任意)。 | ||
変更イベントレコード値に含まれる必要がある列の完全修飾名と一致する正規表現のコンマ区切りリスト (任意)。列の完全修飾名の形式は schemaName.tableName.columnName です。また、 | ||
変更イベントレコード値から除外される必要がある列の完全修飾名と一致する正規表現のコンマ区切りリスト (任意)。列の完全修飾名の形式は schemaName.tableName.columnName です。また、 | ||
|
時間、日付、およびタイムスタンプは、異なる精度の種類で表すことができます。 | |
|
コネクターによる | |
|
コネクターによる | |
|
| |
|
PostgreSQL サーバーへの暗号化された接続を使用するかどうか。オプションには以下が含まれます。 | |
クライアントの SSL 証明書が含まれるファイルへのパス。詳細は PostgreSQL のドキュメント を参照してください。 | ||
クライアントの SSL 秘密鍵が含まれるファイルへのパス。詳細は PostgreSQL のドキュメント を参照してください。 | ||
| ||
サーバーが検証されるルート証明書が含まれるファイルへのパス。詳細は PostgreSQL のドキュメント を参照してください。 | ||
| TCP keep-alive プローブを有効にして、データベース接続がまだ有効であることを確認します。詳細は PostgreSQL のドキュメント を参照してください。 | |
|
削除 イベント後に廃棄 (tombstone) イベントを生成するかどうかを制御します。 | |
該当なし |
文字ベースの列の完全修飾名と一致する正規表現のコンマ区切りリスト (任意)。列の完全修飾名の形式は schemaName.tableName.columnName です。変更イベントレコードでは、これらの列の値がプロパティー名の 長さ によって指定される文字数よりも長い場合は切り捨てられます。単一の設定で、異なる長さを持つ複数のプロパティーを指定できます。長さは正の整数である必要があります (例: | |
該当なし |
文字ベースの列の完全修飾名と一致する正規表現のコンマ区切りリスト (任意)。列の完全修飾名の形式は schemaName.tableName.columnName です。変更イベント値では、指定のテーブルコラムの値はアスタリスク ( | |
該当なし |
文字ベースの列の完全修飾名と一致する正規表現のコンマ区切りリスト (任意)。列の完全修飾名の形式は schemaName.tableName.columnName です。変更イベント値では、指定された列の値は仮名に置き換えられます。 | |
該当なし |
列の完全修飾名と一致する正規表現のコンマ区切りリスト (任意)。列の完全修飾名の形式は、databaseName.tableName.columnName または databaseName.schemaName.tableName.columnName です。 | |
該当なし |
一部の列のデータベース固有のデータ型名と一致する正規表現のコンマ区切りリスト (任意)。完全修飾データ型名の形式は、databaseName.tableName.typeName または databaseName.schemaName.tableName.typeName です。 | |
空の文字列 |
テーブルの列名と一致する正規表現が含まれるテーブルのセミコロン区切りのリスト。コネクターは、一致する列の値を Kafka トピックに送信する変更イベントレコードのキーフィールドにマップします。これは、テーブルにプライマリーキーがない場合や、プライマリーキーではないフィールドに応じて Kafka トピックで変更イベントレコードを順序付けする場合に便利です。 | |
all_tables |
| |
bytes |
バイナリー ( | |
bytes |
+ 切り捨て (truncate) イベントの構造とそれらの順序付けセマンティクスについては、切り捨て (truncate) イベント を参照してください。 |
以下の 高度な 設定プロパティーには、ほとんどの状況で機能するデフォルト設定があるため、コネクターの設定で指定する必要はほとんどありません。
プロパティー | デフォルト | 説明 |
---|---|---|
|
コネクターの起動時にスナップショットを実行する基準を指定します。 | |
|
| |
| スナップショットの実行時に、テーブルロックを取得するまで待つ最大時間 (ミリ秒単位) を指定する正の整数値。コネクターがこの期間にテーブルロックを取得できないと、スナップショットは失敗します。詳細は コネクターによるスナップショットの実行方法 を参照してください。 | |
スナップショットに含まれるテーブル行を制御します。このプロパティーはスナップショットにのみ影響します。これは、論理デコードプラグインによって生成されるイベントには影響を与えません。databaseName.tableName の形式で完全修飾テーブル名のコンマ区切りリストを指定します。 | ||
|
イベントの処理中にコネクターが例外に反応する方法を指定します。 | |
| ブロッキングキューの最大サイズの正の整数値。コネクターは、Kafka に書き込む前にストリーミングレプリケーションから受信される変更イベントをブロッキングキューに配置します。このキューは、たとえば Kafka へのレコードの書き込みが遅い場合や Kafka が利用できない場合などにバックプレシャーを提供できます。 | |
| コネクターが処理するイベントの各バッチの最大サイズを指定する正の整数値。 | |
| ブロッキングキューの最大サイズ (バイト単位) の long 値。この機能はデフォルトで無効になっています。正の long 値が設定されると有効になります。 | |
| コネクターがイベントのバッチの処理を開始する前に、新しい変更イベントの発生を待つ期間をミリ秒単位で指定する正の整数値。デフォルトは 1000 ミリ秒 (1 秒) です。 | |
|
コネクターがデータタイプが不明なフィールドを見つけたときのコネクターの動作を指定します。コネクターが変更イベントからフィールドを省略し、警告をログに記録するのがデフォルトの動作です。 注記
| |
データベースへの JDBC 接続を確立するときにコネクターが実行する SQL ステートメントのセミコロン区切りリスト。セミコロンを区切り文字としてではなく、文字として使用する場合は、2 つの連続したセミコロン | ||
|
コネクターがハートビートメッセージを Kafka トピックに送信する頻度を制御します。デフォルトの動作では、コネクターはハートビートメッセージを送信しません。 | |
|
コネクターがハートビートメッセージを送信するトピックの名前を制御します。トピック名のパターンは次のようになります。 | |
コネクターがハートビートメッセージを送信するときにコネクターがソースデータベースで実行するクエリーを指定します。 | ||
|
テーブルのインメモリースキーマの更新をトリガーする条件を指定します。 | |
コネクターの起動時にスナップショットを実行するまでコネクターが待つ必要がある間隔 (ミリ秒単位)。クラスターで複数のコネクターを起動する場合、このプロパティーは、コネクターのリバランスが行われる原因となるスナップショットの中断を防ぐのに役立ちます。 | ||
| スナップショットの実行中、コネクターは行のバッチでテーブルの内容を読み取ります。このプロパティーは、バッチの行の最大数を指定します。 | |
設定された論理デコードプラグインに渡すパラメーターのセミコロン区切りリスト。例えば、 | ||
コネクターが
そうでない場合は | Avro の命名要件 に準拠するためにフィールド名がサニタイズされるかどうかを示します。 | |
| レプリケーションスロットへの接続に失敗した場合に、連続して接続を試行する最大回数です。 | |
| コネクターがレプリケーションスロットへの接続に失敗した場合に再試行を行う間隔 (ミリ秒単位)。 | |
|
コネクターが提供する定数を指定して、元の値がデータベースによって提供されていない Toast 化された値であることを示します。 | |
|
コネクターがトランザクション境界でイベントを生成し、トランザクションメタデータで変更イベントエンベロープを強化するかどうかを決定します。コネクターにこれを実行させる場合は | |
10000 (10 秒) | 再試行可能なエラーが発生した後にコネクターを再起動するまで待機する時間 (ミリ秒単位)。 |
パススルーコネクター設定プロパティー
コネクターは、Kafka プロデューサーおよびコンシューマーの作成時に使用される パススルー 設定プロパティーもサポートします。
Kafka プロデューサーおよびコンシューマーのすべての設定プロパティーについては、必ず Kafka ドキュメント を参照してください。PostgreSQL コネクターは 新しいコンシューマー設定プロパティー を使用します。
3.7. Debezium PostgreSQL コネクターのパフォーマンスの監視
Debezium PostgreSQL コネクターは、Zookeeper、Kafka、および Kafka Connect によって提供される JMX メトリクスの組み込みサポートに加えて、2 種類のメトリクスを提供します。
- スナップショットメトリクス は、スナップショットの実行中にコネクター操作に関する情報を提供します。
- メトリクスのストリーミング は、コネクターが変更をキャプチャーし、変更イベントレコードをストリーミングする際のコネクター操作に関する情報を提供します。
Debezium の監視に関するドキュメント は、JMX を使用してこれらのメトリクスを公開する方法の詳細を提供します。
3.7.1. PostgreSQL データベースのスナップショット作成時の Debezium の監視
MBean は debezium.postgres:type=connector-metrics,context=snapshot,server=<database.server.name>
です。
属性 | タイプ | 説明 |
---|---|---|
| コネクターが読み取りした最後のスナップショットイベント。 | |
| コネクターが最新のイベントを読み取りおよび処理してからの経過時間 (ミリ秒単位)。 | |
| 前回の開始またはリセット以降にコネクターで確認されたイベントの合計数。 | |
| コネクターに設定された include/exclude リストのフィルターリングルールによってフィルターされたイベントの数。 | |
| コネクターによって監視されるテーブルの一覧。 | |
| snapshotter とメインの Kafka Connect ループの間でイベントを渡すために使用されるキューの長さ。 | |
| snapshotter とメインの Kafka Connect ループの間でイベントを渡すために使用されるキューの空き容量。 | |
| スナップショットに含まれているテーブルの合計数。 | |
| スナップショットによってまだコピーされていないテーブルの数。 | |
| スナップショットが起動されたかどうか。 | |
| スナップショットが中断されたかどうか。 | |
| スナップショットが完了したかどうか。 | |
| スナップショットが完了したかどうかに関わらず、これまでスナップショットにかかった時間 (秒単位)。 | |
| スナップショットの各テーブルに対してスキャンされる行数が含まれるマップ。テーブルは、処理中に増分がマップに追加されます。スキャンされた 10,000 行ごとに、テーブルの完成時に更新されます。 | |
|
キューの最大バッファー (バイト単位)。 | |
| キュー内のレコードの現在のデータ (バイト単位)。 |
3.7.2. Debezium PostgreSQL コネクターレコードストリーミングの監視
MBean は debezium.postgres:type=connector-metrics,context=streaming,server=<database.server.name>
です。
属性 | タイプ | 説明 |
---|---|---|
| コネクターが読み取られた最後のストリーミングイベント。 | |
| コネクターが最新のイベントを読み取りおよび処理してからの経過時間 (ミリ秒単位)。 | |
| 前回の開始またはリセット以降にコネクターで確認されたイベントの合計数。 | |
| コネクターに設定された include/exclude リストのフィルターリングルールによってフィルターされたイベントの数。 | |
| コネクターによって監視されるテーブルの一覧。 | |
| ストリーマーとメイン Kafka Connect ループの間でイベントを渡すために使用されるキューの長さ。 | |
| ストリーマーとメインの Kafka Connect ループの間でイベントを渡すために使用されるキューの空き容量。 | |
| コネクターが現在データベースサーバーに接続されているかどうかを示すフラグ。 | |
| 最後の変更イベントのタイムスタンプとそれを処理するコネクターとの間の期間 (ミリ秒単位)。この値は、データベースサーバーとコネクターが稼働しているマシンのクロック間の差異に対応します。 | |
| コミットされた処理済みトランザクションの数。 | |
| 最後に受信したイベントの位置。 | |
| 最後に処理されたトランザクションのトランザクション識別子。 | |
| キューの最大バッファー (バイト単位)。 | |
| キュー内のレコードの現在のデータ (バイト単位)。 |
3.8. Debezium PostgreSQL コネクターによる障害および問題の処理方法
Debezium は、複数のアップストリームデータベースのすべての変更をキャプチャーする分散システムであり、イベントの見逃しや損失は発生しません。システムが正常に操作している場合や、慎重に管理されている場合は、Debezium は変更イベントレコードごとに 1 度だけ 配信します。
障害が発生しても、システムはイベントを失いません。ただし、障害から復旧している間は、変更イベントが繰り返えされる可能性があります。このような正常でない状態では、Debezium は Kafka と同様に、変更イベントを 少なくとも 1 回 配信します。
詳細は以下を参照してください。
設定および起動エラー
以下の状況では、起動時にコネクターが失敗し、エラーまたは例外がログに記録され、実行が停止されます。
- コネクターの設定が無効である。
- 指定の接続パラメーターを使用してコネクターを PostgreSQL に接続できない。
- コネクターは (LSN を使用して) PostgreSQL WAL の以前に記録された位置から再起動され、PostgreSQL ではその履歴が利用できなくなります。
このような場合、エラーメッセージには問題の詳細が含まれ、推奨される回避策も含まれることがあります。設定の修正したり、PostgreSQL の問題に対処した後、コネクターを再起動します。
PostgreSQL コネクターは、最後に処理されたオフセットを PostgreSQL LSN の形式で外部に保存します。コネクターが再起動し、サーバーインスタンスに接続すると、コネクターはサーバーと通信し、その特定のオフセットからストリーミングを続行します。このオフセットは、Debezium レプリケーションスロットがそのままの状態である限り利用できます。プライマリーサーバーでレプリケーションスロットを削除しないでください。削除するとデータが失われます。スロットが削除された場合の障害例は、次のセクションを参照してください。
クラスターの障害
PostgreSQL はリリース 12 より、プライマリーサーバー上でのみ論理レプリケーションスロットを許可するようになりました。つまり、Debezium PostgreSQL コネクターをデータベースクラスターのアクティブなプライマリーサーバーのみにポイントできます。また、レプリケーションスロット自体はレプリカに伝播されません。プライマリーサーバーがダウンした場合は、新しいプライマリーを昇格する必要があります。
新しいプライマリーには、pgoutput
プラグインが使用するよう設定されたレプリケーションスロットと、変更をキャプチャーするデータベースが必要です。その後でのみ、コネクターが新しいサーバーを示すようにし、コネクターを再起動することができます。
フェイルオーバーが発生した場合は重要な注意点があります。レプリケーションスロットがそのままの状態で、データを損失していないことを確認するまで Debezium を一時停止する必要があります。フェイルオーバー後に以下を行います。
- アプリケーションが新しいプライマリーに書き込みする前に、Debezium のレプリケーションスロットを再作成するプロセスが必要です。これは重要です。このプロセスがないと、アプリケーションが変更イベントを見逃す可能性があります。
- 古いプライマリーが失敗する前に、Debezium がスロットのすべての変更を読み取りできることを確認する必要があることがあります。
失われた変更があるかどうかを確認し、取り戻すための信頼できる方法の 1 つは、障害が発生したプライマリーのバックアップを、障害が発生する直前まで復旧することです。これは管理が難しい場合がありますが、レプリケーションスロットで未使用の変更があるかどうかを確認することができます。
Kafka Connect のプロセスは正常に停止する
Kafka Connect が分散モードで実行され、Kafka Connect プロセスが正常に停止した場合を想定します。Kafka Connect はそのプロセスをシャットダウンする前に、プロセスのコネクタータスクをそのグループの別の Kafka Connect プロセスに移行します。新しいコネクタータスクは、以前のタスクが停止した場所でプロセスを開始します。コネクタータスクが正常に停止され、新しいプロセスで再起動されるまでの間、プロセスに短い遅延が発生します。
Kafka Connect プロセスのクラッシュ
Kafka Connector プロセスが予期せず停止した場合、最後に処理されたオフセットを記録せずに、実行中のコネクタータスクが終了します。Kafka Connect が分散モードで実行されている場合は、Kafka Connect は他のプロセスでこれらのコネクタータスクを再起動します。ただし、PostgreSQL コネクターは、以前のプロセスで最後に記録されたオフセットから再開します。つまり、新しい代替タスクによって、クラッシュの直前に処理された同じ変更イベントが生成される可能性があります。重複するイベントの数は、オフセットのフラッシュ期間とクラッシュの直前のデータ変更の量によって異なります。
障害からの復旧中に一部のイベントが重複された可能性があるため、コンシューマーは常に重複されたイベントがある可能性を想定する必要があります。Debezium の変更はべき等であるため、一連のイベントは常に同じ状態になります。
各変更イベントレコードでは Debezium コネクターは、イベント発生時の PostgreSQL サーバー時間、サーバートランザクションの ID、トランザクションの変更が書き込まれたログ先行書き込みの位置など、イベント発生元に関するソース固有の情報を挿入します。コンシューマーは、LSN を重点としてこの情報を追跡し、イベントが重複しているかどうかを判断します。
コネクターの一定期間の停止
コネクターが正常に停止された場合、データベースを引き続き使用できます。変更はすべて PostgreSQL WAL に記録されます。コネクターが再起動すると、停止した場所で変更のストリーミングが再開されます。つまり、コネクターが停止した間に発生したデータベースのすべての変更に対して変更イベントレコードが生成されます。
適切に設定された Kafka クラスターは大量のスループットを処理できます。Kafka Connect は Kafka のベストプラクティスに従って作成され、十分なリソースがあれば Kafka Connect コネクターも非常に多くのデータベース変更イベントを処理できます。このため、Debezium コネクターがしばらく停止した後に再起動すると、停止中に発生したデータベースの変更に対して処理の遅れを取り戻す可能性が非常に高くなります。遅れを取り戻すのに掛かる時間は、Kafka の機能やパフォーマンス、および PostgreSQL のデータに加えられた変更の量によって異なります。
第4章 MongoDB の Debezium コネクター
Debezium の MongoDB コネクターは、データベースおよびコレクションにおけるドキュメントの変更に対して、MongoDB レプリカセットまたは MongoDB シャードクラスターを追跡し、これらの変更を Kafka トピックのイベントとして記録します。コネクターは、シャードクラスターにおけるシャードの追加または削除、各レプリカセットのメンバーシップの変更、各レプリカセット内の選出、および通信問題の解決待ちを自動的に処理します。
4.1. 概要
MongoDB のレプリケーションメカニズムは冗長性と高可用性を提供し、実稼働環境における MongoDB の実行に推奨される方法です。MongoDB コネクターは、レプリカセットまたはシャードクラスターの変更をキャプチャーします。
MongoDB レプリカセット は、すべてが同じデータのコピーを持つサーバーのセットで設定され、レプリケーションによって、クライアントがレプリカセットの プライマリー のドキュメントに追加したすべての変更が、セカンダリーと呼ばれる別のレプリカセットのサーバーに適用されるようにします。MongoDB のレプリケーションでは、プライマリーが oplog (または操作ログ) に変更を記録した後、各セカンダリーがプライマリーの oplog を読み取って、すべての操作を順番に独自のドキュメントに適用します。新規サーバーをレプリカセットに追加すると、そのサーバーは最初にプライマリーのすべてのデータベースおよびコレクションの スナップショット を実行し、次にプライマリーの oplog を読み取り、スナップショットの開始後に加えられたすべての変更を適用します。この新しいサーバーは、プライマリーの oplog の最後に到達するとセカンダリーになり、クエリーを処理できます。
MongoDB コネクターはこのレプリケーションメカニズムを使用しますが、実際にはレプリカセットのメンバーにはなりません。ただし、MongoDB のセカンダリーと同様に、コネクターはレプリカセットのプライマリーの oplog を常に読み取ります。また、コネクターが初めてレプリカセットを表示するとき、oplog を確認して最後に記録されたトランザクションを取得した後、プライマリーのデータベースおよびコレクションのスナップショットを実行します。すべてのデータがコピーされると、コネクターは oplog から読み取った位置から変更をストリーミングします。MongoDB oplog における操作は べき等 であるため、操作の適用回数に関係なく、同じ最終状態になります。
MongoDB コネクターが変更を処理すると、イベントの発生元となる oplog の位置を定期的に記録します。MongoDB コネクターが停止したときに、最後に処理した oplog の位置を記録するため、再起動時にはその位置からストリーミングが開始されます。つまり、コネクターを停止、アップグレード、または維持でき、後で再起動できます。イベントを何も失うことなく、停止した場所を正確に特定します。当然ながら、MongoDB の oplogs は通常は最大サイズに制限されているため、コネクターを長時間停止しないようにしてください。長時間停止すると、oplog の操作によってはコネクターによって読み取られる前にパージされる可能性があります。この場合、コネクターを再起動すると、不足している oplog 操作が検出され、スナップショットが実行されます。その後、変更のストリーミングが続行されます。
MongoDB コネクターは、レプリカセットのメンバーシップとリーダーシップの変更、シャードクラスター内でのシャードの追加と削除、および通信障害の原因となる可能性のあるネットワーク問題にも非常に寛容です。コネクターは常にレプリカセットのプライマリーノードを使用して変更をストリーミングします。そのため、レプリカセットの選出が行われ、他のノードがプライマリーになると、コネクターはすぐ変更のストリーミングを停止し、新しいプライマリーに接続し、新しいプライマリーを使用して変更のストリーミングを開始します。同様に、コネクターがレプリカセットのプライマリーと通信する際に問題が発生した場合は、再接続を試み (ネットワークまたはレプリカセットを圧倒しないように指数バックオフを使用)、最後に停止した位置から変更のストリーミングを続行します。これにより、コネクターはレプリカセットメンバーシップの変更を動的に調整でき、通信の失敗を自動的に処理できます。
4.2. MongoDB の設定
MongoDB コネクターは MongoDB の oplog を使用して変更をキャプチャーするため、コネクターは MongoDB レプリカセットと、各シャードが個別のレプリカセットであるシャードクラスターとのみ動作します。レプリカセット または シャードクラスター の設定については、MongoDB ドキュメントを参照してください。また、レプリカセットで アクセス制御と認証 を有効にする方法についても理解するようにしてください。
oplog が読み取られる admin
データベースを読み取るために適切なロールを持つ MongoDB ユーザーも必要です。さらに、ユーザーはシャードクラスターの設定サーバーで config
データベースを読み取りできる必要もあり、listDatabases
権限も必要です。
4.3. サポートされる MongoDB トポロジー
MongoDB コネクターはさまざまな MongoDB トポロジーで使用できます。
4.3.1. MongoDB レプリカセット
MongoDB コネクターは単一の MongoDB レプリカセット から変更をキャプチャーできます。実稼働のレプリカセットには、少なくとも 3 つのメンバー が必要です。
レプリカセットで MongoDB コネクターを使用するには、コネクターの mongodb.hosts
プロパティーを使用して、1 つ以上のレプリカセットサーバーのアドレスを シードアドレス として提供します。コネクターはこれらのシードを使用してレプリカセットに接続した後、レプリカセットからメンバーの完全セットを取得し、どのメンバーがプライマリーであるかを認識します。コネクターは、プライマリーに接続するタスクを開始し、プライマリーの oplog から変更をキャプチャーします。レプリカセットが新しいプライマリーを選出すると、タスクは自動的に新しいプライマリーに切り替えます。
MongoDB がプロキシーと面する場合 (Docker on OS X や Windows などのように)、クライアントがレプリカセットに接続し、メンバーを検出すると、MongoDB クライアントはプロキシーを有効なメンバーから除外し、プロキシーを経由せずに直接メンバーに接続しようとし、失敗します。
このような場合、コネクターのオプションの mongodb.members.auto.discover
設定プロパティーを false
に設定して、コネクターにメンバーシップの検出を見送るように指示し、代わりに最初のシードアドレス (mongodb.hosts
プロパティーによって指定) をプライマリーノードとして使用するよう指示します。これは機能する可能性がありますが、選出が行われるときに問題が発生します。
4.3.2. MongoDB のシャードクラスター
MongoDB のシャードクラスター は以下で設定されます。
- レプリカセットとしてデプロイされる 1 つ以上のシャード。
- クラスターの設定サーバーとして動作する個別のレプリカセット。
-
クライアントが接続し、要求を適切なシャードにルーティングする 1 つ以上の ルーター (
mongos
とも呼ばれます)。
シャードクラスターで MongoDB コネクターを使用するには、コネクターを設定サーバーレプリカセットのホストアドレスで設定します。コネクターがこのレプリカセットに接続すると、シャードクラスターの設定サーバーとして動作していることを検出し、クラスターでシャードとして使用される各レプリカセットに関する情報を検出した後、各レプリカセットから変更をキャプチャーするために別のタスクを起動します。新しいシャードがクラスターに追加される場合または既存のシャードが削除される場合、コネクターはそのタスクを自動的に調整します。
4.3.3. MongoDB スタンドアロンサーバー
スタンドアロンサーバーには oplog がないため、MongoDB コネクターはスタンドアロン MongoDB サーバーの変更を監視できません。スタンドアロンサーバーが 1 つのメンバーを持つレプリカセットに変換されると、コネクターが動作します。
MongoDB は、実稼働でのスタンドアロンサーバーの実行を 推奨しません。
4.4. MongoDB コネクターの仕組み
MongoDB コネクターが設定およびデプロイされると、シードアドレスの MongoDB サーバーに接続して起動し、利用可能な各レプリカセットの詳細を判断します。各レプリカセットには独立した独自の oplog があるため、コネクターはレプリカセットごとに個別のタスクの使用を試みます。コネクターは、使用するタスクの最大数を制限でき、十分なタスクが利用できない場合は、コネクターは各タスクに複数のレプリカセットを割り当てます。ただし、タスクはレプリカセットごとに個別のスレッドを使用します。
シャードクラスターに対してコネクターを実行する場合は、レプリカセットの数よりも大きい tasks.max
の値を使用します。これにより、コネクターはレプリカセットごとに 1 つのタスクを作成でき、Kafka Connect が利用可能なワーカープロセス全体でタスクを調整、配布、および管理できるようにします。
4.4.1. 論理コネクター名
コネクター設定プロパティー mongodb.name
は、MongoDB レプリカセットまたはシャードされたクラスターの 論理名 として提供されます。コネクターは、論理名をさまざまな方法で使用します。すべてトピック名のプレフィックとして使用したり、各レプリカセットの oplog の位置を記録するときに一意の識別子として使用したりします。
各 MongoDB コネクターに、ソース MongoDB システムを意味する一意の論理名を命名する必要があります。論理名は、アルファベットまたはアンダースコアで始まり、残りの文字を英数字またはアンダースコアとすることが推奨されます。
4.4.2. スナップショットの実行
タスクがレプリカセットを使用して起動すると、コネクターの論理名とレプリカセット名を使用して、コネクターが変更の読み取りを停止した位置を示す オフセット を検出します。オフセットが検出され、oplog に存在する場合、タスクは記録されたオフセットの位置から即座に 変更のストリーミング を続行します。
ただし、オフセットが見つからない場合や、oplog にその位置が含まれなくなった場合、タスクは スナップショット を実行してレプリカセットの内容の現在の状態を取得する必要があります。このプロセスは、oplog の現在の位置を記録して開始され、オフセット (スナップショットが開始されたことを示すフラグとともに) として記録します。その後、タスクは各コレクションをコピーし、できるだけ多くのスレッドを生成し (snapshot.max.threads
設定プロパティーの値まで)、この作業を並行して行います。コネクターは、確認した各ドキュメントの個別の 読み取りイベント を記録します。読み取りイベントにはオブジェクトの識別子、オブジェクトの完全な状態、およびオブジェクトが見つかった MongoDB レプリカセットの ソース 情報が含まれます。ソース情報には、スナップショット中にイベントが生成されたことを示すフラグも含まれます。
このスナップショットは、コネクターのフィルターと一致するすべてのコレクションがコピーされるまで継続されます。タスクのスナップショットが完了する前にコネクターが停止した場合は、コネクターを再起動すると、再びスナップショットを開始します。
コネクターがレプリカセットのスナップショットを実行している間は、タスクの再割り当てと再設定をしないようにします。コネクターはスナップショットの進捗とともにメッセージをログに記録します。最大限の制御を行う場合は、各コネクターに対して Kafka Connect の個別のクラスターを実行します。
4.4.3. 変更のストリーミング
レプリカセットのコネクタータスクがオフセットを取得すると、オフセットを使用して変更のストリーミングを開始する oplog の位置を判断します。その後、タスクはレプリカセットのプライマリーノードに接続し、その位置から変更のストリーミングを開始し、すべての作成、挿入、および削除操作を処理して、それらを Debezium の 変更イベント に変換します。各変更イベントには操作が検出された oplog の位置が含まれ、コネクターはこれを最新のオフセットとして定期的に記録します。オフセットが記録される間隔は、Kafka Connect ワーカー設定プロパティーである offset.flush.interval.ms
によって制御されます。
コネクターが正常に停止されると、処理された最後のオフセットが記録され、再起動時にコネクターは停止した場所から続行されます。しかし、コネクターのタスクが予期せず終了した場合、最後にオフセットが記録された後、最後のオフセットが記録される前に、タスクによってイベントが処理および生成されることがあります。再起動時に、コネクターは最後に 記録された オフセットから開始し、クラッシュの前に生成された同じイベントを生成する可能性があります。
すべてが通常どおり動作している場合、Kafka コンシューマーは実際にすべてのメッセージを 1 度だけ 確認します。ただし、問題が発生した場合は、Kafka はコンシューマーが 少なくとも 1 度 各メッセージを確認することのみを保証します。したがって、コンシューマーが複数回メッセージを確認することを想定する必要があります。
前述のように、コネクタータスクは常にレプリカセットのプライマリーノードを使用して oplog からの変更をストリーミングし、コネクターが可能な限り最新の操作を確認できるようにし、代わりにセカンダリーが使用された場合よりも短いレイテンシーで変更をキャプチャーできるようにします。レプリカセットが新しいプライマリーを選出すると、コネクターは即座に変更のストリーミングを停止し、新しいプライマリーに接続して、同じ場所にある新しいプライマリーノードから変更のストリーミングを開始します。同様に、コネクターとレプリカセットメンバーとの通信で問題が発生した場合は、レプリカセットが過剰にならないように指数バックオフを使用して再接続を試みます。接続の確立後、停止した場所から変更のストリーミングを続行します。これにより、コネクターはレプリカセットメンバーシップの変更を動的に調整でき、通信障害を自動的に処理できます。
要約すると、MongoDB コネクターはほとんどの状況で実行を継続します。通信の問題により、問題が解決されるまでコネクターが待機する可能性があります。
4.4.4. トピック名
MongoDB コネクターは、各コレクションのドキュメントに対するすべての挿入、更新、および削除操作のイベントを 1 つの Kafka トピックに書き込みます。Kafka トピックの名前は常に logicalName.databaseName.collectionName の形式を取ります。logicalName は、mongodb.name
設定プロパティーで指定されるコネクターの 論理名、databaseName は操作が発生したデータベースの名前、collectionName は影響を受けるドキュメントが存在する MongoDB コレクションの名前です。
たとえば、products
, products_on_hand
, customers
, and orders
の 4 つのコレクションで設定される inventory
データベースを含む MongoDB レプリカセットについて考えてみましょう。コネクターが監視するこのデータベースの論理名が fulfillment
である場合、コネクターは以下の 4 つの Kafka トピックでイベントを生成します。
-
fulfillment.inventory.products
-
fulfillment.inventory.products_on_hand
-
fulfillment.inventory.customers
-
fulfillment.inventory.orders
トピック名には、レプリカセット名やシャード名が含まれないことに注意してください。その結果、シャード化コレクションへの変更 (各シャードにコレクションのドキュメントのサブセットが含まれる) はすべて同じ Kafka トピックに移動します。
Kafka を設定して、必要に応じてトピックを 自動作成 できます。そうでない場合は、Kafka 管理ツールを使用してコネクターを起動する前にトピックを作成する必要があります。
4.4.5. パーティション
MongoDB コネクターは、イベントのトピックパーティションを明示的に決定しません。代わりに、Kafka がキーに基づいてパーティションを判断できるようにします。Kafka Connect ワーカー設定に Partitioner
実装の名前を定義することで、Kafka のパーティショニングロジックを変更できます。
Kafka は、1 つのトピックパーティションに書き込まれたイベントのみ、合計順序を維持します。キーでイベントのパーティションを行うと、同じキーを持つすべてのイベントは常に同じパーティションに移動します。これにより、特定のドキュメントのすべてのイベントが常に完全に順序付けされます。
4.4.6. データ変更イベント
Debezium MongoDB コネクターは、データを挿入、更新、または削除する各ドキュメントレベルの操作に対してデータ変更イベントを生成します。各イベントにはキーと値が含まれます。キーと値の構造は、変更されたコレクションによって異なります。
Debezium および Kafka Connect は、イベントメッセージの継続的なストリーム を中心として設計されています。ただし、これらのイベントの構造は時間の経過とともに変化する可能性があり、コンシューマーによる処理が困難になることがあります。これに対応するために、各イベントにはコンテンツのスキーマが含まれます。スキーマレジストリーを使用している場合は、コンシューマーがレジストリーからスキーマを取得するために使用できるスキーマ ID が含まれます。これにより、各イベントが自己完結型になります。
以下のスケルトン JSON は、変更イベントの基本となる 4 つの部分を示しています。ただし、アプリケーションで使用するために選択した Kafka Connect コンバーターの設定方法によって、変更イベントのこれら 4 部分の表現が決定されます。schema
フィールドは、変更イベントが生成されるようにコンバーターを設定した場合のみ変更イベントに含まれます。同様に、イベントキーおよびイベントペイロードは、変更イベントが生成されるようにコンバーターを設定した場合のみ変更イベントに含まれます。JSON コンバーターを使用し、変更イベントの基本となる 4 つの部分すべてを生成するように設定すると、変更イベントの構造は次のようになります。
{ "schema": { 1 ... }, "payload": { 2 ... }, "schema": { 3 ... }, "payload": { 4 ... }, }
項目 | フィールド名 | 説明 |
---|---|---|
1 |
|
最初の |
2 |
|
最初の |
3 |
|
2 つ目の |
4 |
|
2 つ目の |
デフォルトでは、コネクターによって、変更イベントレコードがイベントの元のコレクションと同じ名前を持つトピックにストリーミングされます。トピック名 を参照してください。
MongoDB コネクターは、すべての Kafka Connect スキーマ名が Avro スキーマ名の形式 に準拠するようにします。つまり、論理サーバー名はアルファベットまたはアンダースコア (a-z、A-Z、または _) で始まる必要があります。論理サーバー名の残りの各文字と、データベース名とコレクション名の各文字は、アルファベット、数字、またはアンダースコア ( a-z、A-Z、0-9、または _) でなければなりません。無効な文字がある場合は、アンダースコアに置き換えられます。
論理サーバー名、データベース名、またはコレクション名に無効な文字が含まれ、名前を区別する唯一の文字が無効であると、無効な文字はすべてアンダースコアに置き換えられるため、予期せぬ競合が発生する可能性があります。
4.4.6.1. 変更イベントキー
変更イベントのキーには、変更されたドキュメントのキーのスキーマと、変更されたドキュメントの実際のキーのスキーマが含まれます。特定のコレクションでは、スキーマとそれに対応するペイロードの両方に単一の id
フィールドが含まれます。このフィールドの値は、MongoDB Extended JSON のシリアライゼーションの厳格モード から派生する文字列として表されるドキュメントの識別子です。
論理名が fulfillment
のコネクター、inventory
データベースが含まれるレプリカセット、および以下のようなドキュメントが含まれる customers
コレクションについて考えてみましょう。
ドキュメントの例
{ "_id": 1004, "first_name": "Anne", "last_name": "Kretchmar", "email": "annek@noanswer.org" }
変更イベントキーの例
customers
コレクションへの変更をキャプチャーする変更イベントのすべてに、イベントキースキーマがあります。customers
コレクションに前述の定義がある限り、customers
コレクションへの変更をキャプチャーする変更イベントのキー構造はすべて以下のようになります。JSON では、以下のようになります。
{ "schema": { 1 "type": "struct", "name": "fulfillment.inventory.customers.Key", 2 "optional": false, 3 "fields": [ 4 { "field": "id", "type": "string", "optional": false } ] }, "payload": { 5 "id": "1004" } }
項目 | フィールド名 | 説明 |
---|---|---|
1 |
|
キーのスキーマ部分は、キーの |
2 |
|
キーのペイロードの構造を定義するスキーマの名前。このスキーマは、変更したドキュメントのキーの構造を説明します。キースキーマ名の形式は connector-name.database-name.collection-name.
|
3 |
|
イベントキーの |
4 |
|
各フィールドの名前、型、および必要かどうかなど、 |
5 |
|
この変更イベントが生成されたドキュメントのキーが含まれます。この例では、キーには型 |
この例では、整数の識別子を持つドキュメントを使用しますが、有効な MongoDB ドキュメント識別子は、ドキュメント識別子を含め、同じように動作します。ドキュメント識別子の場合、イベントキーの payload.id
値は、厳格モードを使用する MongoDB Extended JSON シリアライゼーションとして更新されたドキュメントの元の _id
フィールドを表す文字列です。以下の表では、さまざまな型の _id
フィールドを表す例を示します。
タイプ | MongoDB _id の値 | キーのペイロード |
---|---|---|
Integer | 1234 |
|
Float | 12.34 |
|
String | "1234" |
|
Document |
|
|
ObjectId |
|
|
バイナリー |
|
|
4.4.6.2. 変更イベントの値
変更イベントの値はキーよりも若干複雑です。キーと同様に、値には schema
セクションと payload
セクションがあります。schema
セクションには、入れ子のフィールドを含む、 Envelope
セクションの payload
構造を記述するスキーマが含まれています。データを作成、更新、または削除する操作のすべての変更イベントには、Envelope 構造を持つ値 payload があります。
変更イベントキーの例を紹介するために使用した、同じサンプルドキュメントについて考えてみましょう。
ドキュメントの例
{ "_id": 1004, "first_name": "Anne", "last_name": "Kretchmar", "email": "annek@noanswer.org" }
このドキュメントへの変更に対する変更イベントの値部分には、以下の各イベントタイプについて記述されています。
4.4.6.3. 作成 イベント
以下の例は、customers
コレクションにデータを作成する操作に対して、コネクターによって生成される変更イベントの値の部分を示しています。
{ "schema": { 1 "type": "struct", "fields": [ { "type": "string", "optional": true, "name": "io.debezium.data.Json", 2 "version": 1, "field": "after" }, { "type": "string", "optional": true, "name": "io.debezium.data.Json", "version": 1, "field": "patch" }, { "type": "string", "optional": true, "name": "io.debezium.data.Json", "version": 1, "field": "filter" }, { "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": "rs" }, { "type": "string", "optional": false, "field": "collection" }, { "type": "int32", "optional": false, "field": "ord" }, { "type": "int64", "optional": true, "field": "h" } ], "optional": false, "name": "io.debezium.connector.mongo.Source", 3 "field": "source" }, { "type": "string", "optional": true, "field": "op" }, { "type": "int64", "optional": true, "field": "ts_ms" } ], "optional": false, "name": "dbserver1.inventory.customers.Envelope" 4 }, "payload": { 5 "after": "{\"_id\" : {\"$numberLong\" : \"1004\"},\"first_name\" : \"Anne\",\"last_name\" : \"Kretchmar\",\"email\" : \"annek@noanswer.org\"}", 6 "patch": null, "source": { 7 "version": "1.4.2.Final", "connector": "mongodb", "name": "fulfillment", "ts_ms": 1558965508000, "snapshot": false, "db": "inventory", "rs": "rs0", "collection": "customers", "ord": 31, "h": 1546547425148721999 }, "op": "c", 8 "ts_ms": 1558965515240 9 } }
項目 | フィールド名 | 説明 |
---|---|---|
1 |
| 値のペイロードの構造を記述する、値のスキーマ。変更イベントの値スキーマは、コネクターが特定のコレクションに生成するすべての変更イベントで同じになります。 |
2 |
|
|
3 |
|
|
4 |
|
|
5 |
|
値の実際のデータ。これは、変更イベントが提供する情報です。 |
6 |
|
イベント発生後のドキュメントの状態を指定する任意のフィールド。この例では、 |
7 |
| イベントのソースメタデータを記述する必須のフィールド。このフィールドには、イベントの発生元、イベントの発生順序、およびイベントが同じトランザクションの一部であるかどうかなど、このイベントと他のイベントを比較するために使用できる情報が含まれています。ソースメタデータには以下が含まれています。
|
8 |
|
コネクターによってイベントが生成される原因となった操作の型を記述する必須文字列。この例では、
|
9 |
|
コネクターがイベントを処理した時間を表示する任意のフィールド。この時間は、Kafka Connect タスクを実行している JVM のシステムクロックを基にします。 |
4.4.6.4. 更新イベント
サンプル customers
コレクションにある更新の変更イベントの値には、そのコレクションの 作成 イベントと同じスキーマがあります。同様に、イベント値のペイロードは同じ構造を持ちます。ただし、イベント値ペイロードでは 更新 イベントに異なる値が含まれます。更新 イベントには after
の値はありません。その代わりに、以下の 2 つのフィールドがあります。
-
patch
は、べき等更新操作の JSON 表現が含まれる文字列フィールドです。 -
filter
は、更新の選択基準の JSON 表現が含まれる文字列フィールドです。filter
文字列には、シャード化コレクションの複数のシャードキーフィールドを含めることができます。
以下は、コネクターによって customers
コレクションでの更新に生成されるイベントの変更イベント値の例になります。
{ "schema": { ... }, "payload": { "op": "u", 1 "ts_ms": 1465491461815, 2 "patch": "{\"$set\":{\"first_name\":\"Anne Marie\"}}", 3 "filter": "{\"_id\" : {\"$numberLong\" : \"1004\"}}", 4 "source": { 5 "version": "1.4.2.Final", "connector": "mongodb", "name": "fulfillment", "ts_ms": 1558965508000, "snapshot": true, "db": "inventory", "rs": "rs0", "collection": "customers", "ord": 6, "h": 1546547425148721999 } } }
項目 | フィールド名 | 説明 |
---|---|---|
1 |
|
コネクターによってイベントが生成される原因となった操作の型を記述する必須文字列。この例では、 |
2 |
|
コネクターがイベントを処理した時間を表示する任意のフィールド。この時間は、Kafka Connect タスクを実行している JVM のシステムクロックを基にします。 |
3 |
|
ドキュメントへの実際の MongoDB のべき等変更の JSON 文字列表現が含まれます。この例では、更新で |
4 |
| 更新するドキュメントの特定に使用された MongoDB 選択基準の JSON 文字列表現が含まれます。 |
5 |
| イベントのソースメタデータを記述する必須のフィールド。このフィールドには、同じコレクションの 作成 イベントと同じ情報が含まれますが、oplog の異なる位置からのイベントであるため、値は異なります。ソースメタデータには以下が含まれています。
|
Debezium 変更イベントでは、MongoDB は patch
フィールドの内容を提供します。このフィールドの形式は、MongoDB データベースのバージョンによって異なります。したがって、新しい MongoDB データベースバージョンにアップグレードする場合は、形式が変更された可能性があるため注意してください。本書のサンプルは、MongoDB 3.4 から取得したため、ご使用のアプリケーションではイベントの形式が異なる場合があります。
MongoDB の oplog では、更新 イベントには変更されたドキュメントの前 または 後 の状態は含まれません。そのため、Debezium コネクターがこの情報を提供することはできません。ただし、Debezium コネクターは 作成 および 読み取り イベントでドキュメントの開始状態を提供します。ストリームのダウンストリームのコンシューマーは、ドキュメントごとに最新状態を維持し、新しいイベントの状態を保存された状態に比較することで、ドキュメント状態を再構築できます。Debezium コネクターはこの状態を維持できません。
4.4.6.5. 削除 イベント
delete change イベントの値は、create や update と同じ schema
部分を持ちます。delete イベントの payload
部分には、同じコレクションの 作成 と 更新 イベントとは異なる値が含まれます。特に、 削除 イベントには after
の値や patch
の値は含まれません。以下は、customers
コレクションのドキュメントの 削除 イベントの例になります。
{ "schema": { ... }, "payload": { "op": "d", 1 "ts_ms": 1465495462115, 2 "filter": "{\"_id\" : {\"$numberLong\" : \"1004\"}}", 3 "source": { 4 "version": "1.4.2.Final", "connector": "mongodb", "name": "fulfillment", "ts_ms": 1558965508000, "snapshot": true, "db": "inventory", "rs": "rs0", "collection": "customers", "ord": 6, "h": 1546547425148721999 } } }
項目 | フィールド名 | 説明 |
---|---|---|
1 |
|
操作の型を記述する必須の文字列。 |
2 |
|
コネクターがイベントを処理した時間を表示する任意のフィールド。この時間は、Kafka Connect タスクを実行している JVM のシステムクロックを基にします。 |
3 |
| 削除するドキュメントの特定に使用された MongoDB 選択基準の JSON 文字列表現が含まれます。 |
4 |
| イベントのソースメタデータを記述する必須のフィールド。このフィールドには、同じコレクションの 作成 または 更新 イベントと同じ情報が含まれますが、oplog の異なる位置からのイベントであるため、値は異なります。ソースメタデータには以下が含まれています。
|
MongoDB コネクターイベントは、Kafka ログコンパクション と動作するように設計されています。ログコンパクションにより、少なくとも各キーの最新のメッセージが保持される限り、一部の古いメッセージを削除できます。これにより、トピックに完全なデータセットが含まれ、キーベースの状態のリロードに使用できるようにするとともに、Kafka がストレージ領域を確保できるようにします。
廃棄 (tombstone) イベント
一意に識別ドキュメントの MongoDB コネクターイベントはすべて同じキーを持ちます。ドキュメントが削除された場合でも、Kafka は同じキーを持つ以前のメッセージをすべて削除できるため、削除 イベントの値はログコンパクションで動作します。ただし、Kafka がそのキーを持つすべてのメッセージを削除するには、メッセージの値が null
である必要があります。これを可能にするために、Debezium の MongoDB コネクターは 削除 イベントを出力した後に、null
値以外で同じキーを持つ特別な廃棄 (tombstone) イベントを出力します。tombstone イベントは、同じキーを持つすべてのメッセージを削除できることを Kafka に通知します。
4.4.7. トランザクションメタデータ
Debezium は、トランザクションメタデータ境界を表すイベントを生成でき、データメッセージをエンリッチできます。
4.4.7.1. トランザクション境界
Debezium は、すべてのトランザクションの BEGIN
および END
のイベントを生成します。各イベントに以下が含まれます。
-
status
:BEGIN
またはEND
-
id
- 一意のトランザクション識別子の文字列表現。 -
event_count
(END
イベント用) - トランザクションによって出力されるイベントの合計数 -
data_collections
(END
イベントの場合): 指定のデータコレクションからの変更によって出力されたイベントの数を提供するdata_collection
とevent_count
のペアの配列。
以下は、メッセージの内容の例です。
{ "status": "BEGIN", "id": "1462833718356672513", "event_count": null, "data_collections": null } { "status": "END", "id": "1462833718356672513", "event_count": 2, "data_collections": [ { "data_collection": "rs0.testDB.tablea", "event_count": 1 }, { "data_collection": "rs0.testDB.tableb", "event_count": 1 } ] }
トランザクションイベントは、<database.server.name>.transaction
という名前のトピックに書き込まれます。
4.4.7.2. データイベントのエンリッチメント
トランザクションメタデータを有効にすると、データメッセージ Envelope
は新しい transaction
フィールドでエンリッチされます。このフィールドは、複合フィールドの形式ですべてのイベントに関する情報を提供します。
-
id
- 一意のトランザクション識別子の文字列表現。 -
total_order
- トランザクションによって生成されたすべてのイベントを対象とするイベントの絶対位置 -
data_collection_order
- トランザクションによって出力されたすべてのイベントを対象とするイベントのデータコレクションごとの位置。
以下は、メッセージの内容の例です。
{ "before": null, "after": { "pk": "2", "aa": "1" }, "source": { ... }, "op": "c", "ts_ms": "1580390884335", "transaction": { "id": "1462833718356672513", "total_order": "1", "data_collection_order": "1" } }
4.5. Deploying the MongoDB connector
Debezium MongoDB コネクターをデプロイするには、Debezium MongoDB コネクターアーカイブをインストールして、コネクターを設定し、その設定を Kafka Connect に追加してコネクターを起動します。
MongoDB コネクターをインストールするには、Debezium の OpenShift へのインストール の手順に従います。主なステップは以下のとおりです。
- Red Hat AMQ Streams を使用して OpenShift で Apache Kafka および Kafka Connect を設定します。AMQ Streams は、Kafka を OpenShift に取り入れる operator およびイメージを提供します。
- Debezium MongoDB コネクター をダウンロードします。
- コネクターのファイルを Kafka Connect 環境に展開します。
コネクタープラグインの親ディレクトリーを Kafka Connect
plugin.path
に追加します。以下に例を示します。plugin.path=/kafka/connect
上記の例では、Debezium MongoDB コネクターを
/kafka/connect/Debezium-connector-mongodb
パスに展開したことを前提としています。- Kafka Connect プロセスを再起動して、新しい JAR ファイルが確実に取得されるようにします。
また、MongoDB を設定 して Debezium コネクターを実行する必要もあります。
その他のリソース
デプロイメントプロセスや AMQ Streams でのコネクターのデプロイに関する詳細は、Debezium のインストールガイドを参照してください。
4.5.1. 設定例
コネクターを使用して特定の MongoDB レプリカセットまたはシャードクラスターの変更イベントを生成するには、JSON で設定ファイルを作成します。コネクターが起動すると、MongoDB レプリカセットでコレクションのスナップショットを実行し、レプリカセットの oplogs の読み取りを開始して、挿入、更新、削除されたすべてのドキュメントのイベントを生成します。任意で、不必要なコレクションを除外します。
以下は、MongoDB レプリカセット rs0
を 192.168.99.100 のポート 27017 でモニターする MongoDB コネクターの設定例です (論理名は fullfillment
)。通常、コネクターに使用できる設定プロパティーを使用して、.yaml
ファイルに Debezium MongoDB コネクターを設定します。
apiVersion: kafka.strimzi.io/v1beta1 kind: KafkaConnector metadata: name: inventory-connector 1 labels: strimzi.io/cluster: my-connect-cluster spec: class: io.debezium.connector.mongodb.MongoDbConnector 2 config: mongodb.hosts: rs0/192.168.99.100:27017 3 mongodb.name: fulfillment 4 collection.include.list: inventory[.]* 5
- 1 1 1 1 1 1
- Kafka Connect サービスに登録する場合のコネクターの名前。
- 2 2 2 2 2 2
- MongoDB コネクタークラスの名前。
- 3 3 3 3 3 3
- MongoDB レプリカセットへの接続に使用するホストアドレス。
- 4 4 4 4 4 4
- 生成されたイベントの namespace を形成する MongoDB レプリカセットの論理名。コネクターが書き込む Kafka トピックの名前、Kafka Connect スキーマ名、および Arvo コンバーターが使用される場合に対応する Avro スキーマの namespace のすべてに使用されます。
- 5 5 5 5
- 監視するすべてのコレクションのコレクション namespace (例: <dbName>.<collectionName>) と一致する正規表現のリスト。これは任意です。
これらの設定で指定できる コネクタープロパティーの完全リスト を参照してください。
この設定は、POST 経由で稼働中の Kafka Connect サービスに送信できます。その後、設定を記録し、MongoDB レプリカセットまたはシャードクラスターに接続するコネクタータスクを 1 つ起動して、各レプリカセットにタスクを割り当てます。必要に応じてスナップショットを実行し、oplog を読み取り、Kafka トピックへのイベントを記録します。
4.5.2. コネクター設定の追加
提供される Debezium コンテナーを使用して、Debezium MongoDB コネクターをデプロイできます。この手順では、Debezium のカスタム Kafka Connect コンテナーイメージをビルドし、必要に応じて Debezium コネクターを設定して、コネクター設定を Kafka Connect 環境に追加します。
前提条件
- Podman または Docker がインストールされ、コンテナーを作成および管理するのに十分な権限がある。
- Debezium MongoDB コネクターアーカイブがインストールされている。
手順
Debezium MongoDB コネクターアーカイブを展開して、コネクタープラグインのディレクトリー構造を作成します。以下に例を示します。
tree ./my-plugins/ ./my-plugins/ ├── debezium-connector-mongodb │ ├── ...
Debezium コネクターを実行するためのカスタムイメージを作成し、パブリッシュします。
registry.redhat.io/amq7/amq-streams-kafka-26-rhel7:1.6.0
をベースイメージとして使用して、新規のDockerfile
を作成します。以下の例の my-plugins は、プラグインディレクトリーの名前に置き換えます。FROM registry.redhat.io/amq7/amq-streams-kafka-26-rhel7:1.6.0 USER root:root COPY ./my-plugins/ /opt/kafka/plugins/ USER 1001
Kafka Connect は、コネクターの実行を開始する前に、
/opt/kafka/plugins
ディレクトリーにあるサードパーティープラグインをロードします。コンテナーイメージをビルドします。たとえば、前のステップで作成した
Dockerfile
をdebezium-container-for-mongodb
として保存し、Dockerfile
が現在のディレクトリーにある場合は、以下のコマンドを実行します。podman build -t debezium-container-for-mongodb:latest .
カスタムイメージをコンテナーレジストリーにプッシュします。例を以下に示します。
podman push debezium-container-for-mongodb:latest
新しいコンテナーイメージを示します。次のいずれかを行います。
KafkaConnector
カスタムリソースのspec.image
プロパティーを編集します。このプロパティーが設定されていると、クラスターオペレータのSTRIMZI_DEFAULT_KAFKA_CONNECT_IMAGE
変数がオーバーライドされます。以下に例を示します。apiVersion: kafka.strimzi.io/v1beta1 kind: KafkaConnector metadata: name: my-connect-cluster spec: #... image: debezium-container-for-mongodb
-
install/cluster-operator/050-Deployment-strimzi-cluster-operator.yaml
ファイルのSTRIMZI_DEFAULT_KAFKA_CONNECT_IMAGE
変数を編集し、新しいコンテナーイメージを示すようにした後、Cluster Operator を再インストールします。このファイルを編集する場合は、OpenShift クラスターに適用する必要があります。
-
Debezium MongoDB コネクターインスタンスを定義する
KafkaConnector
カスタムリソースを作成します。コネクター設定の例 を参照してください。 コネクターインスタンスを適用します。以下に例を示します。
oc apply -f inventory-connector.yaml
これにより
inventory-connector
が登録され、コネクターがinventory
データベースに対して実行されるようになります。コネクターが作成され、指定されたデータベース内の変更のキャプチャーを開始したことを確認します。例えば
inventory-connector
が起動したときの Kafka Connect のログ出力を見ることで、コネクターのインスタンスを確認することができます。Kafka Connect のログ出力を表示します。
oc logs $(oc get pods -o name -l strimzi.io/name=my-connect-cluster-connect)
ログの出力を確認し、初回のスナップショットが実行されたことを確認します。以下のような行が表示されるはずです。
... INFO Starting snapshot for ... ... INFO Snapshot is using user 'debezium' ...
結果
コネクターが起動すると、コネクターが設定された MongoDB データベースの 整合性スナップショットが実行 されます。その後、コネクターはドキュメントレベルの操作のデータ変更イベントの生成を開始し、変更イベントレコードを Kafka トピックにストリーミングします。
4.5.3. 監視
Debezium MongoDB コネクターには、Zookeeper、Kafka、および Kafka Connect にある JMX メトリクスの組み込みサポートに加えて、2 種類のメトリクスがあります。
- スナップショットの実行時にコネクターを監視するための、スナップショットメトリクス。
- oplog イベントの処理時にコネクターを監視するための、ストリーミングメトリクス。
JMX 経由でこれらのメトリクスを公開する方法の詳細は、監視に関するドキュメント を参照してください。
4.5.3.1. スナップショットメトリクス
MBean は debezium.mongodb:type=connector-metrics,context=snapshot,server=<mongodb.name>
です。
属性 | タイプ | 説明 |
---|---|---|
| コネクターが読み取りした最後のスナップショットイベント。 | |
| コネクターが最新のイベントを読み取りおよび処理してからの経過時間 (ミリ秒単位)。 | |
| 前回の開始またはリセット以降にコネクターで確認されたイベントの合計数。 | |
| コネクターに設定された include/exclude リストのフィルターリングルールによってフィルターされたイベントの数。 | |
| コネクターによって監視されるテーブルの一覧。 | |
| snapshotter とメインの Kafka Connect ループの間でイベントを渡すために使用されるキューの長さ。 | |
| snapshotter とメインの Kafka Connect ループの間でイベントを渡すために使用されるキューの空き容量。 | |
| スナップショットに含まれているテーブルの合計数。 | |
| スナップショットによってまだコピーされていないテーブルの数。 | |
| スナップショットが起動された |