Debezium ユーザーガイド
Debezium 1.7 での使用
概要
前書き
Debezium は、データベースの行レベルの変更をキャプチャーする分散サービスのセットで、アプリケーションがそれらの変更を認識し、応答できるようにします。Debezium は、各データベーステーブルにコミットされたすべての行レベルの変更を記録します。各アプリケーションは、対象のトランザクションログを読み取り、発生した順序ですべての操作を確認します。
本ガイドでは、以下の Debezium トピックの使用方法について説明します。
- 1章Debezium の概要
- 2章必要となるカスタムリソースのアップグレード
- 3章Db2 の Debezium コネクター
- 4章MongoDB の Debezium コネクター
- 5章MySQL の Debezium コネクター
- 6章Oracle 向けの Debezium コネクター (テクノロジープレビュー)
- 7章PostgreSQL の Debezium コネクター
- 8章SQL Server の Debezium コネクター
- 9章Debezium の監視
- 10章Debezium のログ機能
- 11章アプリケーション用 Debezium コネクターの設定
- 12章Apache Kafka で交換されたメッセージを修正するためのトランスフォームの適用
多様性を受け入れるオープンソースの強化
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 を使用して監視できます。
- メッセージルーティング、フィルターリング、イベントフラット化などに、SMT(すぐに使用できる単一のメッセージ変換) などを使用できます。Debezium が提供する SMT の詳細は、Applying transformations to modify messagesd with Apache Kafkaを参照してください。
各コネクターのドキュメントには、コネクター機能と設定オプションの詳細が記載されています。
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 の new record state extraction の変換を設定する必要がある場合があります。この Kafka Connect SMT は、Debezium の変更イベントからシンクコネクターに after
構造を伝播します。これは、デフォルトで伝播される詳細な変更イベントレコードの代わりになります。
第2章 必要となるカスタムリソースのアップグレード
Debezium は、AMQ Streams on OpenShift で実行する Apache Kafka クラスターにデプロイされた Kafka コネクタープラグインです。OpenShift CRD v1
を準備するため、現行バージョンの AMQ Streams で、カスタムリソース定義 (CRD)API の必要なバージョンが v1beta2
に設定されます。API の v1beta2
バージョンは、以前にサポートされている v1beta1
および v1 alpha1
API バージョンを置き換えます。AMQ Streams では、v1alpha1
および v1beta1
API バージョンのサポートが非推奨になりました。以前のバージョンは、Debezium コネクターの設定に使用する KafkaConnect および KafkaConnector リソースを含む、AMQ Streams カスタムリソースから削除されています。
v1beta2
API バージョンをベースとする CRD は OpenAPI 構造スキーマを使用します。後続の v1alpha1 または v1beta1 API に基づくカスタムリソースは、構造的なスキーマをサポートしないため、現在のバージョンの AMQ Streams と互換性がありません。AMQ Streams2.0 にアップグレードする前に、API バージョン kafka.strimzi.io/v1beta2
を使用するように既存のカスタムリソースをアップグレードする必要があります。AMQ Streams 1.7 にアップグレードした後も、カスタムリソースをいつでもアップグレードできます。AMQ Streams2.0 以降にアップグレードする前に、v1beta2 API へのアップグレードを完了する必要があります。
CRD およびカスタムリソースのアップグレードを容易にするため、AMQ Streams は v1beta2
と互換性のある形式に自動的にアップグレードする API 変換ツールを提供します。ツールおよび AMQ Streams のアップグレード方法は、 OpenShift での AMQ Streams のデプロイおよびアップグレード を参照してください。
カスタムリソースを更新する要件は、AMQ Streams on OpenShift で実行される Debezium デプロイメントにのみ適用されます。要件は、Red Hat Enterprise Linux の Debezium には適用されません。
第3章 Db2 の Debezium コネクター
Debezium の Db2 コネクターは、Db2 データベースのテーブルで行レベルの変更をキャプチャーできます。このコネクターと互換性のある Db2 データベースのバージョンについては、Debezium でサポートされる設定ページを参照してください。
このコネクターは、テーブルをキャプチャーモードにする SQL ベースのポーリングモデルを使用する、SQL Server の Debezium 実装から大きく影響を受けます。テーブルがキャプチャーモードの場合、Debezium Db2 コネクターは、そのテーブルへの行レベルの更新ごとに変更イベントを生成し、ストリーミングします。
キャプチャーモードのテーブルには、関連する変更テーブルがあり、このテーブルは Db2 によって作成されます。キャプチャーモードのテーブルに対する変更ごとに、Db2 はその変更に関するデータをテーブルの関連する変更データテーブルに追加します。変更データテーブルには、行の各状態のエントリーが含まれます。また、削除に関する特別なエントリーもあります。Debezium Db2 コネクターは変更イベントを変更データテーブルから読み取り、イベントを Kafka トピックに出力します。
Debezium Db2 コネクターが Db2 データベースに初めて接続すると、コネクターが変更をキャプチャーするように設定されたテーブルの整合性スナップショットを読み取ります。デフォルトでは、システム以外のテーブルがすべて対象になります。キャプチャーモードにするテーブルまたはキャプチャーモードから除外するテーブルを指定できるコネクター設定プロパティーがあります。
スナップショットが完了すると、コネクターはコミットされた更新の変更イベントをキャプチャーモードのテーブルに出力し始めます。デフォルトでは、特定のテーブルの変更イベントは、テーブルと同じ名前を持つ Kafka トピックに移動します。アプリケーションとサービスはこれらのトピックから変更イベントを使用します。
コネクターには、Linux 用の Db2 の標準部分として利用できる抽象構文表記 (ASN) ライブラリーを使用する必要があります。ASN ライブラリーを使用するには、IBM InfoSphere Data Replication (IIDR) のライセンスが必要です。ASN ライブラリーを使用するには、IIDR をインストールする必要はありません。
Debezium Db2 コネクターを使用するための情報および手順は、以下のように設定されています。
3.1. Debezium Db2 コネクターの概要
Debezium Db2 コネクターは、Db2 で SQL レプリケーションを有効にする ASN Capture/Apply エージェント をベースにしています。キャプチャーエージェントは以下を行います。
- キャプチャーモードであるテーブルの変更データテーブルを生成します。
- キャプチャーモードのテーブルを監視し、更新の変更イベントを対応する変更データテーブルのテーブルに格納します。
Debezium コネクターは SQL インターフェイスを使用して変更イベントの変更データテーブルに対してクエリーを実行します。
データベース管理者は、変更をキャプチャーするテーブルをキャプチャーモードにする必要があります。便宜上およびテストを自動化するために、以下の管理タスクをコンパイルし、実行できる Debezium ユーザー定義機能 (UDF) が C にあります。
- ASN エージェントの開始、停止、および再初期化。
- テーブルをキャプチャーモードにする。
- レプリケーション (ASN) スキーマと変更データテーブルの作成。
- キャプチャーモードからのテーブルの削除。
また、Db2 制御コマンドを使用してこれらのタスクを実行することもできます。
対象のテーブルがキャプチャーモードになった後、コネクターは対応する変更データテーブルを読み取り、テーブル更新の変更イベントを取得します。コネクターは、変更されたテーブルと同じ名前を持つ Kafka トピックに対して、行レベルの挿入、更新、および削除操作ごとに変更イベントを出力します。これは、変更可能なデフォルトの動作です。クライアントアプリケーションは、対象のデータベーステーブルに対応する Kafka トピックを読み取り、行レベルの各変更イベントに対応できます。
通常、データベース管理者はテーブルのライフサイクルの途中でテーブルをキャプチャーモードにします。つまり、コネクターにはテーブルに加えられたすべての変更の完全な履歴はありません。そのため、Db2 コネクターが最初に特定の Db2 データベースに接続すると、キャプチャーモードである各テーブルで 整合性スナップショット を実行して起動します。コネクターは、スナップショットの完成後に、スナップショットが作成された時点から変更イベントをストリーミングします。これにより、コネクターはキャプチャーモードのテーブルの整合性のあるビューで開始し、スナップショットの実行中に加えられた変更は破棄されません。
Debezium コネクターはフォールトトラレントです。コネクターは変更イベントを読み取りおよび生成すると、変更データテーブルエントリーのログシーケンス番号 (LSN) を記録します。LSN はデータベースログの変更イベントの位置になります。コネクターが何らかの理由で停止した場合 (通信障害、ネットワークの問題、クラッシュなど)、コネクターは再起動後に最後に停止した場所の変更データテーブルの読み取りを続行します。これにはスナップショットが含まれます。つまり、コネクターの停止時にスナップショットが完了しなかった場合、コネクターの再起動時に新しいスナップショットが開始されます。
3.2. Debezium Db2 コネクターの仕組み
Debezium Db2 コネクターを最適に設定および実行するには、コネクターによるスナップショットの実行方法、変更イベントのストリーム方法、Kafka トピック名の決定方法、およびスキーマ変更の処理方法を理解すると便利です。
詳細は以下を参照してください。
3.2.1. Debezium Db2 コネクターによるデータベーススナップショットの実行方法
Db2 のレプリケーション機能は、データベース変更の完全な履歴を保存するようには設計されていません。そのため、Debezium Db2 コネクターが初めてデータベースに接続すると、キャプチャーモードのテーブルの整合性スナップショットを作成し、この状態を Kafka にストリーミングします。これにより、テーブルの内容のベースラインが確立されます。
デフォルトでは、Db2 コネクターがスナップショットを実行すると、以下が実行されます。
-
キャプチャーモードになっているテーブルを判断するため、スナップショットに含まれなければならないテーブルも判断されます。デフォルトでは、システム以外のテーブルはすべてキャプチャーモードになっています。
table.exclude.list
やtable.include.list
などのコネクター設定プロパティーを使用すると、キャプチャーモードである必要があるテーブルを指定できます。 -
キャプチャーモードの各テーブルでロックを取得します。これにより、スナップショットの実行中にこれらのテーブルでスキーマの変更が発生しないようにします。ロックのレベルは、
snapshot.isolation.mode
コネクター設定プロパティーによって決定されます。 - サーバーのトランザクションログで、最大 (最新) の LSN の位置を読み取ります。
- キャプチャーモードになっているすべてのテーブルのスキーマをキャプチャーします。コネクターは、内部データベース履歴トピックでこの情報を永続化します。
- 必要に応じて、ステップ 2 で取得したロックを解放します。通常、これらのロックは短期間のみ保持されます。
ステップ 3 で読み取られた LSN の位置で、コネクターはキャプチャーモードテーブルとそれらのスキーマをスキャンします。スキャン中、コネクターは以下を行います。
- スナップショットの開始前に、テーブルが作成されたことを確認します。そうでない場合は、スナップショットはそのテーブルをスキップします。スナップショットが完了し、コネクターが変更イベントの出力を開始した後、コネクターはスナップショットの実行中に作成されたテーブルの変更イベントを生成します。
- キャプチャーモードになっている各テーブルで、各行の 読み取り イベントを生成します。すべての 読み取り イベントには、LSN の位置が含まれ、これはステップ 3 で取得した LSN の位置と同じです。
- テーブルと同じ名前を持つ Kafka トピックに各 読み取り イベントを出力します。
- コネクターオフセットにスナップショットの正常な完了を記録します。
3.2.1.1. アドホックスナップショット
アドホックスナップショットの使用はテクノロジープレビュー機能です。テクノロジープレビュー機能は、Red Hat の実稼働環境のサービスレベルアグリーメント (SLA) ではサポートされません。また、機能的に完全ではない可能性があるため、Red Hat はテクノロジープレビュー機能を実稼働環境に実装することは推奨しません。テクノロジープレビュー機能は、最新の技術をいち早く提供し、開発段階で機能のテストやフィードバックの収集を可能にするために提供されます。サポート範囲の詳細は、テクノロジープレビュー機能のサポート範囲 を参照してください。
デフォルトでは、コネクターは初回スナップショット操作の開始後にのみ実行されます。通常の状況では、この最初のスナップショットが作成されると、コネクターではスナップショットプロセスは繰り返し処理されません。コネクターがキャプチャーする今後の変更イベントデータはストリーミングプロセス経由でのみ行われます。
ただし、場合によっては、最初のスナップショット中にコネクターを取得したデータが古くなったり、失われたり、または不完全となったり可能性があります。テーブルデータを再キャプチャーするメカニズムを提供するため、Debezium にはアドホックスナップショットを実行するオプションがあります。データベースで以下が変更されたことで、アドホックスナップショットが実行される場合があります。
- コネクター設定は、異なるテーブルセットをキャプチャーするように変更されます。
- Kafka トピックを削除して、再構築する必要があります。
- 設定エラーや他の問題が原因で、データの破損が発生します。
アドホックと呼ばれるスナップショット を開始することで、以前にスナップショットをキャプチャーしたテーブルのスナップショットを再実行できます。アドホックスナップショットには、シグナルテーブル を使用する必要があります。シグナルリクエストを Debezium シグナルテーブルに送信して、アドホックスナップショットを開始します。
既存のテーブルのアドホックスナップショットを開始すると、コネクターはテーブルにすでに存在するトピックにコンテンツを追加します。既存のトピックが削除された場合には、トピックの自動作成 が有効になっているのであれば、Debezium は自動的にトピックを作成できます。
アドホックのスナップショットシグナルは、スナップショットに追加するテーブルを指定します。スナップショットは、データベースの内容全体をキャプチャーしたり、データベース内のテーブルのサブセットのみをキャプチャーしたりできます。
execute-snapshot
メッセージをシグナルテーブルに送信してキャプチャーするテーブルを指定します。以下の表で説明されているように、run-snapshot
シグナルのタイプを incremental
に設定し、スナップショットに追加するテーブルの名前を指定します。
フィールド | デフォルト | 値 |
---|---|---|
|
|
実行するスナップショットのタイプを指定します。 |
| 該当なし |
スナップショットを作成するテーブルの完全修飾名が含まれる配列。 |
アドホックスナップショットのトリガー
execute-snapshot
シグナルタイプのエントリーをシグナルテーブルに追加して、アドホックスナップショットを開始します。コネクターがメッセージを処理した後に、スナップショット操作を開始します。スナップショットプロセスは、最初と最後のプライマリーキーの値を読み取り、これらの値を各テーブルの開始ポイントおよびエンドポイントとして使用します。テーブルのエントリー数と設定されたチャンクサイズに基づいて、Debezium はテーブルをチャンクに分割し、チャンクごとに 1 度に 1 つずつスナップショットを順番に作成していきます。
現在、execute-snapshot
アクションタイプは 増分スナップショット のみをトリガーします。詳細は、スナップショットの増分を参照してください。
3.2.1.2. 増分スナップショット
増分スナップショットの使用はテクノロジープレビュー機能です。テクノロジープレビュー機能は、Red Hat の実稼働環境のサービスレベルアグリーメント (SLA) ではサポートされません。また、機能的に完全ではない可能性があるため、Red Hat はテクノロジープレビュー機能を実稼働環境に実装することは推奨しません。テクノロジープレビュー機能は、最新の技術をいち早く提供し、開発段階で機能のテストやフィードバックの収集を可能にするために提供されます。サポート範囲の詳細は、テクノロジープレビュー機能のサポート範囲 を参照してください。
スナップショットを柔軟に管理するため、Debezium には 増分スナップショット と呼ばれる補助スナップショットメカニズムが含まれています。増分スナップショットは、Debezium コネクターにシグナルを送信するための Debezium メカニズムに依存します。
増分スナップショットでは、最初のスナップショットのように、データベースの完全な状態を一度にすべてキャプチャーする代わりに、一連の設定可能なチャンクで各テーブルを段階的にキャプチャーします。スナップショットがキャプチャーするテーブルと、各チャンクのサイズ を指定できます。チャンクのサイズにより、データベース上の各フェッチ操作中にスナップショットで収集される行数が決まります。増分スナップショットのデフォルトのチャンクサイズは 1 KB です。
増分スナップショットが進むと、Debezium はウォーターマークを使用して進捗を追跡し、キャプチャーする各テーブル行のレコードを管理します。この段階的なアプローチでは、標準の初期スナップショットプロセスと比較して、以下の利点があります。
- スナップショットが完了するまで、ストリーミングストリーミングを延期する代わりに、ストリームしたデータキャプチャーと並行して増分スナップショットを実行できます。コネクターはスナップショットプロセス全体で変更ログからのほぼリアルタイムイベントをキャプチャーし続け、他の操作はブロックしません。
- 増分スナップショットの進捗が中断された場合は、データを失うことなく再開できます。プロセスが再開すると、スナップショットは最初からテーブルをキャプチャーするのではなく、停止した時点から開始します。
-
いつでも増分スナップショットを実行し、必要に応じてプロセスを繰り返してデータベースの更新に適合できます。たとえば、コネクター設定を変更してテーブルを
table.include.list
プロパティーに追加した後にスナップショットを再実行します。
増分スナップショットプロセス
増分スナップショットを実行する場合には、Debezium は各テーブルをプライマリーキー別に分類して、設定されたチャンクサイズ に基づいてテーブルをチャンクに分割します。チャンクごとに作業し、テーブルの行ごとにチャンクでキャプチャーします。キャプチャーする行ごとに、スナップショットは READ
イベントを出力します。そのイベントは、対象となるチャンクのスナップショットを開始する時の行の値を表します。
スナップショットの作成が進むにつれ、他のプロセスがデータベースへのアクセスを継続し、テーブルレコードが変更される可能性があります。このような変更を反映させるように、通常通りに INSERT
、UPDATE
、DELETE
操作がトランザクションログにコミットされます。同様に、継続中の Debezium ストリーミングプロセスは、これらの変更イベントを検出し、対応する変更イベントレコードを Kafka に出力します。
Debezium を使用してプライマリーキーが同じレコード間での競合を解決する方法
場合によっては、ストリーミングプロセスが出力する UPDATE
または DELETE
イベントを順番に受信できます。つまり、ストリーミングプロセスは、スナップショットがその行の READ
イベントが含まれるチャンクをキャプチャーする前に、テーブルの行を変更するイベントを生成する可能性があります。スナップショットが最終的に対象の行にあった READ
イベントを出力すると、その値はすでに置き換えられています。Debezium は、シーケンスが到達する増分スナップショットイベントが正しい論理順序で処理されるように、競合を解決するためにバッファースキームを使用します。スナップショットのイベント間で競合が発生し、ストリームされたイベントが解決されてからでないと、Debezium はイベントのレコードを Kafka に送信しません。
スナップショットウィンドウ
遅れて入ってきた READ
イベントと、同じテーブルの行を変更するストリーミングイベント間の競合の解決を容易にするために、Debezium は スナップショットウィンドウ と呼ばれるものを使用します。スナップショットウィンドウは、増分スナップショットが指定のテーブルチャンクのデータをキャプチャーしている途中に、間隔を決定します。チャンクのスナップショットウィンドウを開く前に、Debezium は通常の動作に従い、トランザクションログから直接ターゲットの Kafka トピックにイベントをダウンストリームに出力します。ただし、特定のチャンクのスナップショットが開放された瞬間から終了するまで、Debezium は重複除去のステップを実行して、プライマリーキーが同じイベント間での競合を解決します。
データコレクションごとに、Debezium は 2 種類のイベントを出力し、それらの両方のレコードを単一の宛先 Kafka トピックに保存します。テーブルから直接キャプチャーするスナップショットレコードは、READ
操作として出力されます。その間、ユーザーはデータコレクションのレコードの更新を続け、各コミットを反映するようにトランザクションログが更新されるので、Debezium は変更ごとに UPDATE
または DELETE
操作を出力します。
スナップショットウィンドウが開放され、Debezium がスナップショットチャンクの処理を開始すると、スナップショットレコードをメモリーバッファーに提供します。スナップショットウィンドウ中に、バッファー内の READ
イベントのプライマリーキーは、受信ストリームイベントのプライマリーキーと比較されます。一致するものが見つからない場合、ストリーミングされたイベントレコードが Kafka に直接送信されます。Debezium が一致を検出すると、バッファーされた READ
イベントを破棄し、ストリーミングされたレコードを宛先トピックに書き込みます。これは、ストリーミングされたイベントが静的スナップショットイベントよりも論理的に優先されるためです。チャンクのスナップショットウィンドウが終了すると、バッファーに含まれるのは、関連するトランザクションログイベントが存在しない READ
イベントのみです。Debezium は、これらの残りの READ
イベントをテーブルの Kafka トピックに出力します。
コネクターは各スナップショットチャンクにプロセスを繰り返します。
増分スナップショットのトリガー
現在、増分スナップショットを開始する唯一の方法は、アドホックスナップショットシグナル をソースデータベースのシグナルテーブルに送信することです。SQL INSERT
クエリーとしてテーブルにシグナルを送信します。Debezium がシグナルテーブルの変更を検出すると、シグナルを読み取り、要求されたスナップショット操作を実行します。
送信するクエリーはスナップショットに追加するテーブルを指定し、必要に応じてスナップショット操作の種類を指定します。現在、スナップショット操作で唯一の有効なオプションはデフォルト値の incremental
だけです。
スナップショットに追加するテーブルを指定するには、テーブルを一覧表示する data-collections
アレイを指定します (例:
{"data-collections": ["public.MyFirstTable", "public.MySecondTable"]}
)。
増分スナップショットシグナルの data-collections
アレイにはデフォルト値がありません。data-collections
アレイが空である場合には、アクションが不要であり、スナップショットを実行しないことが、Debezium で検出されます。
前提条件
- シグナルデータコレクションがソースのデータベースに存在し、コネクターはこれをキャプチャーするように設定されています。
-
シグナルデータコレクションは
signal.data.collection
プロパティーで指定されます。
手順
SQL クエリーを送信し、アドホック増分スナップショット要求をシグナルテーブルに追加します。
INSERT INTO _<signalTable>_ (id, type, data) VALUES (_'<id>'_, _'<snapshotType>'_, '{"data-collections": ["_<tableName>_","_<tableName>_"],"type":"_<snapshotType>_"}');
以下に例を示します。
INSERT INTO myschema.debezium_signal (id, type, data) VALUES('ad-hoc-1', 'execute-snapshot', '{"data-collections": ["schema1.table1", "schema2.table2"],"type":"incremental"}');
コマンドの
id
、type
、およびdata
パラメーターの値は、シグナルテーブルのフィールド に対応します。以下の表では、これらのパラメーターについて説明しています。
表3.2 シグナルテーブルに増分スナップショットシグナルを送信する SQL コマンドのフィールドの説明 値 説明 myschema.debezium_signal
ソースデータベースにあるシグナルテーブルの完全修飾名を指定します。
ad-hoc-1
id
パラメーターは、シグナルリクエストのID
識別子として割り当てられる任意の文字列を指定します。
この文字列を使用して、シグナルテーブルのエントリーへのログメッセージを特定します。Debezium はこの文字列を使用しません。代わりに、スナップショット作成中に、Debezium は独自のID
文字列をウォーターマークシグナルとして生成します。execute-snapshot
type
パラメーターを指定し、シグナルがトリガーする操作を指定します。
data-collections
スナップショットに含めるテーブル名の配列を指定するシグナルの
data
フィールドの必須コンポーネント。
配列は、signal.data.collection
設定プロパティーにコネクターのシグナルテーブルの名前を指定するときに使用する形式で、完全修飾名別にテーブルを一覧表示します。incremental
実行するスナップショット操作の種類指定するシグナルの
data
フィールドの任意のtype
コンポーネント。
現在、唯一の有効なオプションはデフォルト値incremental
だけです。
シグナルテーブルに送信する SQL クエリーでのtype
値の指定は任意です。
値を指定しない場合には、コネクターは増分スナップショットを実行します。
以下の例は、コネクターによってキャプチャーされる増分スナップショットイベントの JSON を示しています。
以下に例を示します。増分スナップショットイベントメッセージ
{ "before":null, "after": { "pk":"1", "value":"New data" }, "source": { ... "snapshot":"incremental" 1 }, "op":"r", 2 "ts_ms":"1620393591654", "transaction":null }
項目 | フィールド名 | 説明 |
---|---|---|
1 |
|
実行するスナップショット操作タイプを指定します。 |
2 |
|
イベントタイプを指定します。 |
Db2 の Debezium コネクターでは、増分スナップショットの実行中のスキーマの変更はサポートしません。
3.2.2. Debezium Db2 コネクターによる変更データテーブルの読み取り方法
スナップショットの完了後、Debezium Db2 コネクターが初めて起動すると、キャプチャーモードである各ソーステーブルの変更データテーブルを識別します。コネクターは各変更データテーブルに対して以下を行います。
- 最後に保存された最大 LSN から現在の最大 LSN の間に作成された変更イベントを読み取ります。
- 各イベントのコミット LSN および変更 LSN に従って、変更イベントを順序付けます。これにより、コネクターはテーブルが変更された順序で変更イベントを出力します。
- コミット LSN および変更 LSN をオフセットとして Kafka Connect に渡します。
- コネクターが Kafka Connect に渡した最大 LSN を保存します。
再起動後、コネクターは停止した場所でオフセット (コミット LSN および変更 LSN) から変更イベントの出力を再開します。コネクターが稼働し、変更イベントを出力している間、キャプチャーモードからテーブルを削除したり、テーブルをキャプチャーモードに追加したりすると、コネクターは変更を検出して、それに合わせて動作を変更します。
3.2.3. Debezium Db2 変更イベントレコードを受信する Kafka トピックのデフォルト名
デフォルトでは、Db2 コネクターは、テーブルで発生するすべての INSERT
、UPDATE
、DELETE
操作の変更イベントを、そのテーブルに固有の単一の Apache Kafka トピックに書き込みます。コネクターは以下の規則を使用して変更イベントトピックに名前を付けます。
databaseName.schemaName.tableName
以下のリストは、デフォルト名のコンポーネントの定義を示しています。
- databaseName
-
database.server.name
コネクター設定プロパティーで指定したコネクターの論理名です。 - schemaName
- 操作が発生したスキーマの名前。
- tableName
- 操作が発生したテーブルの名前。
たとえば、4 つのテーブルが含まれる mydatabase
データベースを使用する Db2 インストールについて考えてみましょう。MYSCHEMA
スキーマにある PRODUCTS
、PRODUCTS_ON_HAND
、CUSTOMERS
、および ORDERS
コネクターはイベントを以下の 4 つの Kafka トピックに出力します。
-
mydatabase.MYSCHEMA.PRODUCTS
-
mydatabase.MYSCHEMA.PRODUCTS_ON_HAND
-
mydatabase.MYSCHEMA.CUSTOMERS
-
mydatabase.MYSCHEMA.ORDERS
コネクターは同様の命名規則を適用して、内部データベース履歴トピック (スキーマ変更トピック と トランザクションメタデータトピック) にラベルを付けます。
デフォルトのトピック名が要件を満たさない場合は、カスタムトピック名を設定できます。カスタムトピック名を設定するには、論理トピックルーティング SMT に正規表現を指定します。論理トピックルーティング SMT を使用してトピックの命名をカスタマイズする方法は、トピックルーティング を参照してください。
3.2.4. Debezium Db2 コネクターのスキーマ変更トピック
Debezium Db2 コネクターを設定すると、データベースのキャプチャーされたテーブルに適用されるスキーマの変更を記述するスキーマ変更イベントを生成できます。
Debezium は、以下の場合にスキーマ変更トピックにメッセージを出力します。
- 新しいテーブルがキャプチャーモードになる。
- テーブルがキャプチャーモードから削除される。
- データベーススキーマの更新 中に、キャプチャーモードであるテーブルのスキーマが変更される。
コネクターは、スキーマ変更イベントをすべて <serverName>
という名前の Kafka トピックに書き込みます。<serverName>
は database.server.name
設定プロパティーに指定されたコネクターの名前に置き換えます。コネクターがスキーマ変更トピックに送信するメッセージには以下の要素などのペイロードが含まれます。
databaseName
-
ステートメントが適用されるデータベースの名前。
databaseName
の値は、メッセージキーとして機能します。 pos
- ステートメントが表示される binlog の位置。
tableChanges
-
スキーマの変更後のテーブルスキーマ全体の構造化表現。
tableChanges
フィールドには、テーブルの各列のエントリーなどのアレイが含まれます。構造化された表現は JSON または Avro 形式でデータを表示するため、コンシューマーは DDL パーサーを介して最初にメッセージを処理しなくてもメッセージを簡単に読み取りできます。
キャプチャーモードであるテーブルでは、コネクターはスキーマ変更トピックにスキーマ変更の履歴だけでなく、内部データベース履歴トピックにも格納します。内部データベース履歴トピックはコネクターのみの使用を対象としており、使用するアプリケーションによる直接使用を目的としていません。スキーマ変更に関する通知が必要なアプリケーションが、スキーマ変更トピックからの情報のみを使用するようにしてください。
データベース履歴トピックをパーティションに分割しないでください。データベース履歴トピックが正しく機能するには、コネクターが出力するイベントレコードの一貫したグローバル順序を維持する必要があります。
トピックがパーティション間で分割されないようにするには、以下のいずれかの方法を使用してトピックのパーティション数を設定します。
-
データベース履歴トピックを手動で作成する場合は、パーティション数を
1
に指定します。 -
Apache Kafka ブローカーを使用してデータベース履歴トピックを自動的に作成する場合に、トピックが作成されるので、Kafka
num.partitions
設定オプションの値を1
に設定します。
コネクターがスキーマ変更トピックに出力するメッセージの形式は、初期の状態であり、通知なしに変更される可能性があります。
以下に例を示します。Db2 コネクターのスキーマ変更トピックに出力されるメッセージ
以下の例は、スキーマ変更トピックのメッセージを示しています。メッセージには、テーブルスキーマの論理表現が含まれます。
{ "schema": { ... }, "payload": { "source": { "version": "1.7.2.Final", "connector": "db2", "name": "db2", "ts_ms": 1588252618953, "snapshot": "true", "db": "testdb", "schema": "DB2INST1", "table": "CUSTOMERS", "change_lsn": null, "commit_lsn": "00000025:00000d98:00a2", "event_serial_no": null }, "databaseName": "TESTDB", 1 "schemaName": "DB2INST1", "ddl": null, 2 "tableChanges": [ 3 { "type": "CREATE", 4 "id": "\"DB2INST1\".\"CUSTOMERS\"", 5 "table": { 6 "defaultCharsetName": null, "primaryKeyColumnNames": [ 7 "ID" ], "columns": [ 8 { "name": "ID", "jdbcType": 4, "nativeType": null, "typeName": "int identity", "typeExpression": "int identity", "charsetName": null, "length": 10, "scale": 0, "position": 1, "optional": false, "autoIncremented": false, "generated": false }, { "name": "FIRST_NAME", "jdbcType": 12, "nativeType": null, "typeName": "varchar", "typeExpression": "varchar", "charsetName": null, "length": 255, "scale": null, "position": 2, "optional": false, "autoIncremented": false, "generated": false }, { "name": "LAST_NAME", "jdbcType": 12, "nativeType": null, "typeName": "varchar", "typeExpression": "varchar", "charsetName": null, "length": 255, "scale": null, "position": 3, "optional": false, "autoIncremented": false, "generated": false }, { "name": "EMAIL", "jdbcType": 12, "nativeType": null, "typeName": "varchar", "typeExpression": "varchar", "charsetName": null, "length": 255, "scale": null, "position": 4, "optional": false, "autoIncremented": false, "generated": false } ] } } ] } }
項目 | フィールド名 | 説明 |
---|---|---|
1 |
| 変更が含まれるデータベースとスキーマを識別します。 |
2 |
|
Db2 コネクターの場合は常に |
3 |
| DDL コマンドによって生成されるスキーマの変更が含まれる 1 つ以上の項目の配列。 |
4 |
| 変更の種類を説明します。値は以下のいずれかになります。
|
5 |
| 作成、変更、または破棄されたテーブルの完全な識別子。 |
6 |
| 適用された変更後のテーブルメタデータを表します。 |
7 |
| テーブルのプライマリーキーを設定する列のリスト。 |
8 |
| 変更されたテーブルの各列のメタデータ。 |
コネクターがスキーマ変更トピックに送信するメッセージでは、メッセージキーはスキーマの変更が含まれるデータベースの名前です。以下の例では、payload
フィールドにキーが含まれます。
{ "schema": { "type": "struct", "fields": [ { "type": "string", "optional": false, "field": "databaseName" } ], "optional": false, "name": "io.debezium.connector.db2.SchemaChangeKey" }, "payload": { "databaseName": "TESTDB" } }
3.2.5. トランザクション境界を表す Debezium Db2 コネクターによって生成されたイベント
Debezium は、トランザクション境界を表し、変更データイベントメッセージをエンリッチするイベントを生成できます。
Debezium は、コネクターのデプロイ後に発生するトランザクションに対してのみメタデータを登録し、受信します。コネクターをデプロイする前に発生するトランザクションのメタデータは利用できません。
Debezium は、すべてのトランザクションで BEGIN
および END
区切り文字のトランザクション境界イベントを生成します。トランザクション境界イベントには以下のフィールドが含まれます。
status
-
BEGIN
またはEND
id
- 一意のトランザクション識別子の文字列表現。
event_count
(END
イベント用)- トランザクションによって出力されるイベントの合計数。
data_collections
(END
イベント用)-
data_collection
とevent_count
要素のペアの配列。これは、コネクターがデータコレクションから発信された変更に対して出力するイベントの数を示します。
例
{ "status": "BEGIN", "id": "00000025:00000d08:0025", "event_count": null, "data_collections": null } { "status": "END", "id": "00000025:00000d08:0025", "event_count": 2, "data_collections": [ { "data_collection": "testDB.dbo.tablea", "event_count": 1 }, { "data_collection": "testDB.dbo.tableb", "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": "00000025:00000d08:0025", "total_order": "1", "data_collection_order": "1" } }
3.3. Debezium Db2 コネクターのデータ変更イベントの説明
Debezium Db2 コネクターは、行レベルの 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 つ目の |
デフォルトでは、コネクターによって、変更イベントレコードがイベントの元のテーブルと同じ名前を持つトピックにストリーミングされます。トピック名 を参照してください。
Debezium Db2 コネクターは、すべての Kafka Connect スキーマ名が Avro スキーマ名の形式 に準拠するようにします。つまり、論理サーバー名はアルファベットまたはアンダースコア (a-z、A-Z、または _) で始まる必要があります。論理サーバー名の残りの各文字と、データベース名とテーブル名の各文字は、アルファベット、数字、またはアンダースコア ( a-z、A-Z、0-9、または \_) でなければなりません。無効な文字がある場合は、アンダースコアに置き換えられます。
論理サーバー名、データベース名、またはテーブル名に無効な文字が含まれ、名前を区別する唯一の文字が無効であると、無効な文字はすべてアンダースコアに置き換えられるため、予期せぬ競合が発生する可能性があります。
また、データベース、スキーマ、およびテーブルの Db2 名では、大文字と小文字を区別することができます。つまり、コネクターは同じ Kafka トピックに複数のテーブルのイベントレコードを出力できます。
詳細は以下を参照してください。
3.3.1. Debezium db2 変更イベントのキー
変更イベントのキーには、変更されたテーブルのキーのスキーマと、変更された行の実際のキーのスキーマが含まれます。スキーマとそれに対応するペイロードの両方には、コネクターによってイベントが作成された時点において、変更されたテーブルの PRIMARY KEY
(または一意の制約) に存在した各列のフィールドが含まれます。
以下の customers
テーブルについて考えてみましょう。この後に、このテーブルの変更イベントキーの例を示します。
テーブルの例
CREATE TABLE customers ( ID INTEGER IDENTITY(1001,1) NOT NULL PRIMARY KEY, FIRST_NAME VARCHAR(255) NOT NULL, LAST_NAME VARCHAR(255) NOT NULL, EMAIL VARCHAR(255) NOT NULL UNIQUE );
変更イベントキーの例
customers
テーブルへの変更をキャプチャーする変更イベントのすべてに、イベントキースキーマがあります。customers
テーブルに前述の定義がある限り、customers
テーブルへの変更をキャプチャーする変更イベントのキー構造はすべて以下のようになります。JSON では、以下のようになります。
{ "schema": { 1 "type": "struct", "fields": [ 2 { "type": "int32", "optional": false, "field": "ID" } ], "optional": false, 3 "name": "mydatabase.MYSCHEMA.CUSTOMERS.Key" 4 }, "payload": { 5 "ID": 1004 } }
項目 | フィールド名 | 説明 |
---|---|---|
1 |
|
キーのスキーマ部分は、キーの |
2 |
|
各フィールドの名前、型、および必要かどうかなど、 |
3 |
|
イベントキーの |
4 |
|
キーのペイロードの構造を定義するスキーマの名前。このスキーマは、変更されたテーブルのプライマリーキーの構造を記述します。キースキーマ名の形式は connector-name.database-name.table-name.
|
5 |
|
この変更イベントが生成された行のキーが含まれます。この例では、キーには値が |
3.3.2. Debezium Db2 変更イベントの値
変更イベントの値はキーよりも若干複雑です。キーと同様に、値には schema
セクションと payload
セクションがあります。schema
セクションには、入れ子のフィールドを含む、 Envelope
セクションの payload
構造を記述するスキーマが含まれています。データを作成、更新、または削除する操作のすべての変更イベントには、Envelope 構造を持つ値 payload があります。
変更イベントキーの例を紹介するために使用した、同じサンプルテーブルについて考えてみましょう。
テーブルの例
CREATE TABLE customers ( ID INTEGER IDENTITY(1001,1) NOT NULL PRIMARY KEY, FIRST_NAME VARCHAR(255) NOT NULL, LAST_NAME VARCHAR(255) NOT NULL, EMAIL VARCHAR(255) NOT NULL UNIQUE );
customers
テーブルのすべての変更イベントのイベント値部分は同じスキーマを指定します。イベント値のペイロードは、イベント型によって異なります。
作成 イベント
以下の例は、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": "mydatabase.MYSCHEMA.CUSTOMERS.Value", 2 "field": "before" }, { "type": "struct", "fields": [ { "type": "int32", "optional": false, "field": "ID" }, { "type": "string", "optional": false, "field": "FIRST_NAME" }, { "type": "string", "optional": false, "field": "LAST_NAME" }, { "type": "string", "optional": false, "field": "EMAIL" } ], "optional": true, "name": "mydatabase.MYSCHEMA.CUSTOMERS.Value", "field": "after" }, { "type": "struct", "fields": [ { "type": "string", "optional": false, "field": "version" }, { "type": "string", "optional": false, "field": "connector" }, { "type": "string", "optional": false, "field": "name" }, { "type": "int64", "optional": false, "field": "ts_ms" }, { "type": "boolean", "optional": true, "default": false, "field": "snapshot" }, { "type": "string", "optional": false, "field": "db" }, { "type": "string", "optional": false, "field": "schema" }, { "type": "string", "optional": false, "field": "table" }, { "type": "string", "optional": true, "field": "change_lsn" }, { "type": "string", "optional": true, "field": "commit_lsn" }, ], "optional": false, "name": "io.debezium.connector.db2.Source", 3 "field": "source" }, { "type": "string", "optional": false, "field": "op" }, { "type": "int64", "optional": true, "field": "ts_ms" } ], "optional": false, "name": "mydatabase.MYSCHEMA.CUSTOMERS.Envelope" 4 }, "payload": { 5 "before": null, 6 "after": { 7 "ID": 1005, "FIRST_NAME": "john", "LAST_NAME": "doe", "EMAIL": "john.doe@example.org" }, "source": { 8 "version": "1.7.2.Final", "connector": "db2", "name": "myconnector", "ts_ms": 1559729468470, "snapshot": false, "db": "mydatabase", "schema": "MYSCHEMA", "table": "CUSTOMERS", "change_lsn": "00000027:00000758:0003", "commit_lsn": "00000027:00000758:0005", }, "op": "c", 9 "ts_ms": 1559729471739 10 } }
項目 | フィールド名 | 説明 |
---|---|---|
1 |
| 値のペイロードの構造を記述する、値のスキーマ。変更イベントの値スキーマは、コネクターが特定のテーブルに生成するすべての変更イベントで同じになります。 |
2 |
|
|
3 |
|
|
4 |
|
|
5 |
|
値の実際のデータ。これは、変更イベントが提供する情報です。 |
6 |
|
イベント発生前の行の状態を指定する任意のフィールド。この例のように、 |
7 |
|
イベント発生後の行の状態を指定する任意のフィールド。この例では、 |
8 |
|
イベントのソースメタデータを記述する必須のフィールド。
|
9 |
|
コネクターによってイベントが生成される原因となった操作の型を記述する必須文字列。この例では、
|
10 |
|
コネクターがイベントを処理した時間を表示する任意のフィールド。この時間は、Kafka Connect タスクを実行している JVM のシステムクロックを基にします。 |
更新イベント
サンプル customers
テーブルにある更新の変更イベントの値には、そのテーブルの 作成 イベントと同じスキーマがあります。同様に、更新イベント値のペイロードは同じ構造を持ちます。ただし、イベント値ペイロードでは 更新 イベントに異なる値が含まれます。以下は、コネクターによって customers
テーブルでの更新に生成されるイベントの変更イベント値の例になります。
{ "schema": { ... }, "payload": { "before": { 1 "ID": 1005, "FIRST_NAME": "john", "LAST_NAME": "doe", "EMAIL": "john.doe@example.org" }, "after": { 2 "ID": 1005, "FIRST_NAME": "john", "LAST_NAME": "doe", "EMAIL": "noreply@example.org" }, "source": { 3 "version": "1.7.2.Final", "connector": "db2", "name": "myconnector", "ts_ms": 1559729995937, "snapshot": false, "db": "mydatabase", "schema": "MYSCHEMA", "table": "CUSTOMERS", "change_lsn": "00000027:00000ac0:0002", "commit_lsn": "00000027:00000ac0:0007", }, "op": "u", 4 "ts_ms": 1559729998706 5 } }
項目 | フィールド名 | 説明 |
---|---|---|
1 |
|
イベント発生前の行の状態を指定する任意のフィールド。更新 イベント値の |
2 |
|
イベント発生後の行の状態を指定する任意のフィールド。 |
3 |
|
イベントのソースメタデータを記述する必須のフィールド。
|
4 |
|
操作の型を記述する必須の文字列。更新 イベントの値では、 |
5 |
|
コネクターがイベントを処理した時間を表示する任意のフィールド。この時間は、Kafka Connect タスクを実行している JVM のシステムクロックを基にします。 |
行のプライマリーキー/一意キーの列を更新すると、行のキーの値が変更されます。キーが変更されると、3 つのイベントが Debezium によって出力されます。3 つのイベントとは、DELETE
イベント、行の古いキーを持つ 廃棄 (tombstone)、およびそれに続く行の新しいキーを持つイベントです。
削除 イベント
削除 変更イベントの値は、同じテーブルの 作成 および 更新 イベントと同じ schema
の部分になります。サンプル customers
テーブルの 削除 イベントのイベント値 payload
は以下のようになります。
{ "schema": { ... }, }, "payload": { "before": { 1 "ID": 1005, "FIRST_NAME": "john", "LAST_NAME": "doe", "EMAIL": "noreply@example.org" }, "after": null, 2 "source": { 3 "version": "1.7.2.Final", "connector": "db2", "name": "myconnector", "ts_ms": 1559730445243, "snapshot": false, "db": "mydatabase", "schema": "MYSCHEMA", "table": "CUSTOMERS", "change_lsn": "00000027:00000db0:0005", "commit_lsn": "00000027:00000db0:0007" }, "op": "d", 4 "ts_ms": 1559730450205 5 } }
項目 | フィールド名 | 説明 |
---|---|---|
1 |
|
イベント発生前の行の状態を指定する任意のフィールド。削除 イベント値の |
2 |
|
イベント発生後の行の状態を指定する任意のフィールド。削除 イベント値の |
3 |
|
イベントのソースメタデータを記述する必須のフィールド。削除 イベント値の
|
4 |
|
操作の型を記述する必須の文字列。 |
5 |
|
コネクターがイベントを処理した時間を表示する任意のフィールド。この時間は、Kafka Connect タスクを実行している JVM のシステムクロックを基にします。 |
削除 変更イベントレコードは、この行の削除を処理するために必要な情報を持つコンシューマーを提供します。コンシューマーによっては、削除を適切に処理するために古い値が必要になることがあるため、古い値が含まれます。
Db2 コネクターイベントは、Kafka のログコンパクション と動作するように設計されています。ログコンパクションにより、少なくとも各キーの最新のメッセージが保持される限り、一部の古いメッセージを削除できます。これにより、トピックに完全なデータセットが含まれ、キーベースの状態のリロードに使用できるようにするとともに、Kafka がストレージ領域を確保できるようにします。
行が削除された場合でも、Kafka は同じキーを持つ以前のメッセージをすべて削除できるため、削除 イベントの値はログコンパクションで動作します。ただし、Kafka が同じキーを持つすべてのメッセージを削除するには、メッセージの値が null
である必要があります。これを可能にするために、Debezium の Db2 コネクターは 削除 イベントを出力した後に、null
値以外で同じキーを持つ特別な廃棄 (tombstone) イベントを出力します。
3.4. Debezium Db2 コネクターによるデータ型のマッピング方法
Db2 のデータ型の説明は Db2 SQL Data Types を参照してください。
Db2 コネクターは、行が存在するテーブルのように構造化されたイベントで行への変更を表します。イベントには、各列の値のフィールドが含まれます。その値がどのようにイベントで示されるかは、列の Db2 のデータ型によって異なります。ここでは、これらのマッピングについて説明します。
詳細は以下を参照してください。
基本型
以下の表では、各 Db2 データ型をイベントフィールドの リテラル型 および セマンティック型にマッピングする方法を説明します。
-
リテラル型 は、Kafka Connect スキーマ型を使用して、値を表す方法を記述します。
INT8
、INT16
、INT32
、INT64
、FLOAT32
、FLOAT64
、BOOLEAN
、STRING
、BYTES
、ARRAY
、MAP
、およびSTRUCT
。 - セマンティック型 は、フィールドの Kafka Connect スキーマの名前を使用して、Kafka Connect スキーマがフィールドの 意味 をキャプチャーする方法を記述します。
DB2 データ型 | リテラル型 (スキーマ型) | セマンティック型 (スキーマ名) および注記 |
---|---|---|
|
| BOOLEAN 型の列のあるテーブルからのみスナップショットを作成できます。現在、Db2 での SQL レプリケーションは BOOLEAN をサポートしないため、Debezium はこれらのテーブルで CDC を実行できません。別の型の使用を検討してください。 |
|
| 該当なし |
|
| 該当なし |
|
| 該当なし |
|
| 該当なし |
|
| 該当なし |
|
|
|
|
|
|
|
|
|
|
| 該当なし |
|
| 該当なし |
|
| 該当なし |
|
| 該当なし |
|
| 該当なし |
|
|
|
|
|
|
|
| 該当なし |
|
| 該当なし |
|
| 該当なし |
|
|
|
列のデフォルト値がある場合は、対応するフィールドの Kafka Connect スキーマに伝達されます。明示的な列値が指定されない限り、変更イベントにはフィールドのデフォルト値が含まれます。そのため、スキーマからデフォルト値を取得する必要はほとんどありません。
時間型
タイムゾーン情報が含まれる Db2 の DATETIMEOFFSET
データ型以外に、時間型がマッピングされる仕組みは time.precision.mode
コネクター設定プロパティーの値によって異なります。ここでは、以下のマッピングについて説明します。
time.precision.mode=adaptive
time.precision.mode
設定プロパティーがデフォルトの adaptive
に設定された場合、コネクターは列のデータ型定義に基づいてリテラル型とセマンティック型を決定します。これにより、イベントがデータベースの値を 正確 に表すようになります。
DB2 データ型 | リテラル型 (スキーマ型) | セマンティック型 (スキーマ名) および注記 |
---|---|---|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
time.precision.mode=connect
time.precision.mode
設定プロパティーが connect
に設定された場合、コネクターは Kafka Connect の論理型を使用します。これは、コンシューマーが組み込みの Kafka Connect の論理型のみを処理でき、可変精度の時間値を処理できない場合に便利です。ただし、Db2 はマイクロ秒の 10 分の 1 の精度をサポートするため、connect
時間精度を指定してコネクターによって生成されたイベントは、データベース列の少数秒の精度値が 3 よりも大きい場合に、精度が失われます。
DB2 データ型 | リテラル型 (スキーマ型) | セマンティック型 (スキーマ名) および注記 |
---|---|---|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
タイムスタンプ型
DATETIME
、SMALLDATETIME
および DATETIME2
タイプは、タイムゾーン情報のないタイムスタンプを表します。このような列は、UTC を基にして同等の Kafka Connect 値に変換されます。たとえば、2018-06-20 15:13:16.945104 という DATETIME2
の値は、1529507596945104 という値の io.debezium.time.MicroTimestamp
で表されます。
Kafka Connect および Debezium を実行している JVM のタイムゾーンは、この変換には影響しません。
DB2 データ型 | リテラル型 (スキーマ型) | セマンティック型 (スキーマ名) および注記 |
---|---|---|
|
|
|
|
|
|
|
|
|
|
|
|
3.5. Debezium コネクターを実行するための Db2 の設定
Db2 テーブルにコミットされた変更イベントを Debezium がキャプチャーするには、必要な権限を持つ Db2 データベース管理者が、変更データキャプチャー (CDC) のデータベースでテーブルを設定する必要があります。Debezium の実行を開始した後、キャプチャーエージェントの設定を調整してパフォーマンスを最適化できます。
Debezium コネクターと使用するために Db2 を設定する場合の詳細は、以下を参照してください。
3.5.1. 変更データキャプチャーの Db2 テーブルの設定
テーブルをキャプチャーモードにするために、Debezium ではユーザー定義関数 (UDF) のセットが提供されます。ここでは、これらの管理 UDF をインストールおよび実行する手順を説明します。また、Db2 制御コマンドを実行してテーブルをキャプチャーモードにすることもできます。その後、管理者は Debezium がキャプチャーする各テーブルに対して、CDC を有効にする必要があります。
前提条件
-
db2instl
ユーザーとして Db2 にログインしている。 - Db2 ホストの $HOME/asncdctools/src ディレクトリーで Debezium 管理 UDF を使用できる。UDF は Debezium サンプルリポジトリー から入手できます。
手順
Db2 で提供される
bldrtn
コマンドを使用して、Db2 サーバーホストで Debezium 管理 UDF をコンパイルします。cd $HOME/asncdctools/src
./bldrtn asncdc
データベースが稼働していない場合は起動します。
DB_NAME
は、Debezium が接続するデータベースの名前に置き換えます。db2 start db DB_NAME
JDBC が Db2 メタデータカタログを読み取りできるようにします。
cd $HOME/sqllib/bnd
db2 bind db2schema.bnd blocking all grant public sqlerror continue
データベースが最近バックアップされたことを確認します。ASN エージェントには、読み取りを始める最新の開始点が必要です。バックアップを実行する必要がある場合は、以下のコマンドを実行して、最新のバージョンのみを利用できるようにデータをプルーニングします。古いバージョンのデータを保持する必要がない場合は、バックアップの場所に
dev/null
を指定します。データベースをバックアップします。
DB_NAME
およびBACK_UP_LOCATION
を適切な値に置き換えます。db2 backup db DB_NAME to BACK_UP_LOCATION
データベースを再起動します。
db2 restart db DB_NAME
データベースに接続して、Debezium 管理 UDF をインストールします。
db2instl
ユーザーとしてログインしていることを前提とするため、UDF がdb2inst1
ユーザーにインストールされている必要があります。db2 connect to DB_NAME
Debezium 管理 UDF をコピーし、その権限を設定します。
cp $HOME/asncdctools/src/asncdc $HOME/sqllib/function
chmod 777 $HOME/sqllib/function
ASN キャプチャーエージェントを開始および停止する Debezium UDF を有効にします。
db2 -tvmf $HOME/asncdctools/src/asncdc_UDF.sql
ASN 制御テーブルを作成します。
$ db2 -tvmf $HOME/asncdctools/src/asncdctables.sql
テーブルをキャプチャーモードに追加し、キャプチャーモードからテーブルを削除する Debezium UDF を有効にします。
$ db2 -tvmf $HOME/asncdctools/src/asncdcaddremove.sql
Db2 サーバーを設定したら、UDF を使用して SQL コマンドで Db2 レプリケーション (ASN) を制御します。UDF によっては戻り値が必要な場合があります。この場合、SQL の
VALUE
ステートメントを使用して呼び出します。その他の UDF には、SQL のCALL
ステートメントを使用します。ASN エージェントを起動します。
VALUES ASNCDC.ASNCDCSERVICES('start','asncdc');
前述のステートメントは、以下のいずれかの結果を返します。
-
asncap is already running
start -->
<COMMAND>
この場合は、以下の例のように、ターミナルウィンドウに指定の
<COMMAND>
を入力します。/database/config/db2inst1/sqllib/bin/asncap capture_schema=asncdc capture_server=SAMPLE &
-
テーブルをキャプチャーモードにします。キャプチャーする各テーブルに対して、以下のステートメントを呼び出します。
MYSCHEMA
は、キャプチャーモードにするテーブルが含まれるスキーマの名前に置き換えます。同様に、MYTABLE
は、キャプチャーモードにするテーブルの名前に置き換えます。CALL ASNCDC.ADDTABLE('MYSCHEMA', 'MYTABLE');
ASN サービスを再初期化します。
VALUES ASNCDC.ASNCDCSERVICES('reinit','asncdc');
3.5.2. Db2 キャプチャーエージェント設定のサーバー負荷およびレイテンシーへの影響
データベース管理者がソーステーブルに対して変更データキャプチャーを有効にすると、キャプチャーエージェントの実行が開始されます。エージェントは新しい変更イベントレコードをトランザクションログから読み取り、イベントレコードをキャプチャーテーブルに複製します。変更がソーステーブルにコミットされてから、対応する変更テーブルに変更が反映される間、常に短いレイテンシーが間隔で発生します。この遅延間隔は、ソーステーブルで変更が発生したときから、Debezium がその変更を Apache Kafka にストリーミングできるようになるまでの差を表します。
データの変更に素早く対応する必要があるアプリケーションについては、ソースとキャプチャーテーブル間で密接に同期を維持するのが理想的です。キャプチャーエージェントを実行してできるだけ迅速に変更イベントを継続的に処理すると、スループットが増加し、レイテンシーが減少するため、イベントの発生後にほぼリアルタイムで新しいイベントレコードが変更テーブルに入力されることを想像するかもしれません。しかし、これは必ずしもそうであるとは限りません。同期を即時に行うとパフォーマンスに影響します。変更エージェントが新しいイベントレコードについてデータベースにクエリーを実行するたびに、データベースホストの CPU 負荷が増加します。サーバーへの負荷が増えると、データベース全体のパフォーマンスに悪影響を及ぼす可能性があり、特にデータベースの使用がピークに達するときにトランザクションの効率が低下する可能性があります。
データベースメトリクスを監視して、サーバーがキャプチャーエージェントのアクティビティーをサポートできなくなるレベルにデータベースが達した場合に認識できるようにすることが重要となります。キャプチャーエージェントの実行中にパフォーマンスの問題が発生した場合は、キャプチャーエージェント設定を調整して CPU の負荷を減らします。
3.5.3. DB2 キャプチャーエージェントの設定パラメーター
Db2 では、IBMSNAP_CAPPARMS
テーブルにはキャプチャーエージェントの動作を制御するパラメーターが含まれています。これらのパラメーターの値を調整して、キャプチャープロセスの設定を調整すると、CPU の負荷を減らしながら許容レベルのレイテンシーを維持することができます。
Db2 のキャプチャーエージェントパラメーターの設定方法に関する具体的なガイダンスは、本書の範囲外となります。
IBMSNAP_CAPPARMS
テーブルでは、CPU 負荷の削減に最も影響を与えるパラメーターは以下のとおりです。
COMMIT_INTERVAL
- キャプチャーエージェントがデータを変更データテーブルにコミットするまで待つ期間を秒単位で指定します。
- 値が大きいほど、データベースホストの負荷が減少し、レイテンシーが増加します。
-
デフォルト値は
30
です。
SLEEP_INTERVAL
- キャプチャーエージェントがアクティブなトランザクションログの最後に到達した後に、新しいコミットサイクルの開始まで待つ期間を秒単位で指定します。
- 値が大きいほど、サーバーの負荷が減少し、レイテンシーが増加します。
-
デフォルト値は
5
です。
関連情報
- キャプチャーエージェントパラメーターの詳細は、Db2 のドキュメントを参照してください。
3.6. Debezium Db2 コネクターのデプロイ
以下の方法のいずれかを使用して Debezium Db2 コネクターをデプロイできます。
Debezium Db2 コネクターでは、Db2 データベースに接続するために Db2JDBC ドライバーが必要です。ドライバーの入手方法については、Db2JDBC ドライバーの入手を参照してください。
3.6.1. Db2 JDBC ドライバーの取得
ライセンス要件により、Db2 JDBC ドライバーファイルは Debezium Db2 コネクターアーカイブに含まれていません。使用するデプロイメント方法に関係なく、デプロイメントを完了するためにドライバーファイルをダウンロードする必要があります。
以下の手順では、ドライバーを取得してお使いの環境で使用する方法を説明します。
手順
ブラウザーから、IBM サポートサイトに ナビゲートし、ご使用のバージョンの Db2 に一致する JDBC ドライバーをダウンロードします。
-
Dockerfile を使用してコネクターを構築する場合は、ダウンロードしたファイルを Debezium Db2 コネクターのファイルがあるディレクトリー、たとえば
<kafka_home>/libs
ディレクトリーにコピーします。 AMQ ストリームを使用してコネクターを Kafka Connect イメージに追加する場合:
- このドライバーを Maven リポジトリーまたは OpenShift クラスターで利用できる別の HTTP サーバーにデプロイします。
-
アーティファクト URL を
KafkaConnect
カスタムリソースに追加します。
-
Dockerfile を使用してコネクターを構築する場合は、ダウンロードしたファイルを Debezium Db2 コネクターのファイルがあるディレクトリー、たとえば
KafkaConnector
リソースを適用してコネクターをデプロイすると、コネクターは指定されたドライバーを使用するよう設定されます。
3.6.2. AMQ Streams を使用した Db2 コネクターデプロイメント
Debezium 1.7 以降、Debezium コネクターのデプロイに推奨される方法は、AMQ Streams を使用してコネクタープラグインが含まれる Kafka Connect コンテナーイメージをビルドすることです。
デプロイメントプロセス中に、以下のカスタムリソース (CR) を作成し、使用します。
-
Kafka Connect インスタンスを定義し、コネクターアーティファクトに関する情報をイメージに含める必要がある
KafkaConnect
CR。 -
コネクターがソースデータベースにアクセスするために使用する情報を提供する
KafkaConnector
CR。AMQStreams が Kafka Connect Pod を開始した後、KafkaConnector
CR を適用してコネクターを開始します。
Kafka Connect イメージのビルド仕様では、デプロイ可能なコネクターを指定できます。各コネクタープラグインに対して、デプロイメントに利用可能にする他のコンポーネントを指定することもできます。たとえば、Service Registry アーティファクトまたは Debezium スクリプトコンポーネントを追加できます。AMQ Streams が Kafka Connect イメージをビルドすると、指定のアーティファクトをダウンロードし、イメージに組み込みます。
Kafka Connect CR の spec.build.output
パラメーターは、生成される KafkaConnect
コンテナーイメージを格納する場所を指定します。コンテナーイメージは Docker レジストリーまたは OpenShift ImageStream に保存できます。イメージを ImageStream に保存するには、Kafka Connect をデプロイする前に ImageStream を作成する必要があります。イメージストリームは自動的に作成されません。
KafkaConnect
リソースを使用してクラスターを作成する場合は、Kafka Connect REST API を使用してコネクターを作成または更新できません。ただし、REST API を使用して情報を取得できます。
関連情報
- AMQ Streams on OpenShift の使用のKafka Connect の設定を参照してください。
- AMQ Streams を使用した新しいコンテナーイメージの自動作成と OpenShift での AMQ Streams のアップグレード
3.6.3. AMQ Streams を使用した Debezium Db2 コネクターのデプロイ
以前のバージョンの AMQ Streams では、OpenShift に Debezium コネクターをデプロイするには、最初にコネクター用の Kafka Connect イメージをビルドする必要がありました。コネクターを OpenShift にデプロイするのに現在推奨される方法は、AMQ Streams でビルド設定を使用して、使用する Debezium コネクタープラグインが含まれる Kafka Connect コンテナーイメージを自動的にビルドすることです。
ビルドプロセス中、AMQ Streams Operator は Debezium コネクター定義を含む KafkaConnect
カスタムリソースの入力パラメーターを Kafka Connect コンテナーイメージに変換します。このビルドは、Red Hat Maven リポジトリーまたは別の設定済みの HTTP サーバーから必要なアーティファクトをダウンロードします。新規に作成されたコンテナーは .spec.build.output
に指定されるコンテナーレジストリーにプッシュされ、Kafka Connect Pod のデプロイに使用されます。AMQ Streams が Kafka Connect イメージをビルドしたら、KafkaConnector
カスタムリソースを作成し、ビルドに含まれるコネクターを起動します。
前提条件
- クラスター Operator がインストールされている OpenShift クラスターにアクセスできる必要があります。
- AMQ Streams Operator が稼働している必要があります。
- Kafka クラスターは、Apache Open Shift での AMQ ストリームのデプロイとアップグレードに記載されているようにデプロイされます。
- Red Hat Integration ライセンスがある。
- Kafka Connect is deployed on AMQ Streams。
-
OpenShift
oc
CLI クライアントがインストールされている、または OpenShift Container Platform Web コンソールにアクセスできる。 Kafka Connect ビルドイメージの保存方法に応じて、レジストリーのパーミッションが必要であるか、ImageStream リソースを作成する必要があります。
- ビルドイメージを Red Hat Quay.io または Docker Hub などのイメージレジストリーに保存するには、以下を実行します。
- レジストリーでイメージを作成し、管理するためのアカウントおよびパーミッション。
- ビルドイメージをネイティブ OpenShift ImageStream として保存します。
- ImageStream リソースがクラスターにデプロイされている。クラスターの ImageStream を明示的に作成する必要があります。ImageStreams はデフォルトでは利用できません。
手順
- OpenShift クラスターにログインします。
コネクターの Debezium
KafkaConnect
カスタムリソース (CR) を作成するか、既存のリソースを変更します。たとえば、以下の例のようにmetadata.annotations
およびspec.build
プロパティーを指定するKafkaConnect
CR を作成します。dbz-connect.yaml
などの名前でファイルを保存します。例3.1 Debezium コネクターを含む
KafkaConnect
カスタムリソースを定義するdbz-connect.yaml
ファイルapiVersion: kafka.strimzi.io/v1beta2 kind: KafkaConnect metadata: name: debezium-kafka-connect-cluster annotations: strimzi.io/use-connector-resources: "true" 1 spec: version: 3.00 build: 2 output: 3 type: imagestream 4 image: debezium-streams-connect:latest plugins: 5 - name: debezium-connector-db2 artifacts: - type: zip 6 url: https://maven.repository.redhat.com/ga/io/debezium/debezium-connector-db2/1.7.2.Final-redhat-<build_number>/debezium-connector-db2-1.7.2.Final-redhat-<build_number>-plugin.zip 7 - type: zip url: https://maven.repository.redhat.com/ga/io/apicurio/apicurio-registry-distro-connect-converter/2.0-redhat-<build-number>/apicurio-registry-distro-connect-converter-2.0-redhat-<build-number>.zip - type: zip url: https://maven.repository.redhat.com/ga/io/debezium/debezium-scripting/1.7.2.Final/debezium-scripting-1.7.2.Final.zip bootstrapServers: debezium-kafka-cluster-kafka-bootstrap:9093
表3.13 Kafka Connect 設定の説明 項目 説明 1
strimzi.io/use-connector-resources
アノテーションをtrue
に設定して、クラスターオペレーターがKafkaConnector
リソースを使用してこの Kafka Connect クラスター内のコネクターを設定できるようにします。2
spec.build
設定は、ビルドイメージの保存場所を指定し、プラグインアーティファクトの場所と共にイメージに追加するプラグインを一覧表示します。3
build.output
は、新たにビルドされたイメージが保存されるレジストリーを指定します。4
イメージ出力の名前およびイメージ名を指定します。
output.type
の有効な値は、Docker Hub や Quay などのコンテナーレジストリーにプッシュする場合はdocker
、内部の OpenShift ImageStream にイメージをプッシュする場合はimagestream
です。ImageStream を使用するには、ImageStream リソースをクラスターにデプロイする必要があります。KafkaConnect 設定でbuild.output
の指定に関する詳細は、AMQ Streams Build スキーマ参照 のドキュメントを参照 してください。5
plugins
設定は、Kafka Connect イメージに追加するすべてのコネクターを一覧表示します。一覧の各エントリーについて、プラグインname
と、コネクターのビルドに必要なアーティファクトに関する情報を指定します。任意で、各コネクタープラグインに対して、コネクターと使用できる他のコンポーネントを含めることができます。たとえば、Service Registry アーティファクトまたは Debezium スクリプトコンポーネントを追加できます。6
artifacts.type
の値は、artifacts.url
で指定したアーティファクトのファイルタイプを指定します。有効なタイプはzip
、tgz
、またはjar
です。Debezium コネクターアーカイブは、.zip
ファイル形式で提供されます。JDBC ドライバーファイルは.jar
形式です。type
の値は、url
フィールドで参照されるファイルのタイプと一致する必要があります。7
artifacts.url
の値は、コネクターアーティファクトのファイルを格納する Maven リポジトリーなどの HTTP サーバーのアドレスを指定します。OpenShift クラスターは指定されたサーバーにアクセスできる必要があります。以下のコマンドを入力して、
KafkaConnect
ビルド仕様を OpenShift クラスターに適用します。oc create -f dbz-connect.yaml
Streams Operator はカスタムリソースで指定された設定に基づいて、デプロイする Kafka Connect イメージを準備します。
ビルドが完了すると、Operator はイメージを指定されたレジストリーまたは ImageStream にプッシュし、Kafka Connect クラスターを起動します。設定に一覧表示されているコネクターアーティファクトはクラスターで利用できます。KafkaConnector
リソースを作成し、デプロイする各コネクターのインスタンスを定義します。
たとえば、以下のKafkaConnector
CR を作成し、db2-inventory-connector.yaml
として保存します。例3.2 Debezium コネクターの
KafkaConnector
カスタムリソースを定義するdb2-inventory-connector.yaml
ファイルapiVersion: kafka.strimzi.io/v1beta2 kind: KafkaConnector metadata: labels: strimzi.io/cluster: debezium-kafka-connect-cluster name: inventory-connector-db2 1 spec: class: io.debezium.connector.db2.Db2ConnectorConnector 2 tasksMax: 1 3 config: 4 database.history.kafka.bootstrap.servers: 'debezium-kafka-cluster-kafka-bootstrap.debezium.svc.cluster.local:9092' database.history.kafka.topic: schema-changes.inventory database.hostname: db2.debezium-db2.svc.cluster.local 5 database.port: 3306 6 database.user: debezium 7 database.password: dbz 8 database.dbname: mydatabase 9 database.server.name: inventory_connector_db2 10 database.include.list: public.inventory 11
表3.14 コネクター設定の説明 項目 説明 1
Kafka Connect クラスターに登録するコネクターの名前。
2
コネクタークラスの名前。
3
同時に動作できるタスクの数。
4
コネクターの設定。
5
ホストデータベースインスタンスのアドレス。
6
データベースインスタンスのポート番号。
7
Debezium がデータベースに接続するユーザーアカウントの名前。
8
データベースユーザーアカウントのパスワード
9
変更をキャプチャーするデータベースの名前。
10
データベースインスタンスまたはクラスターの論理名。
指定の名前は英数字またはアンダースコアからのみ形成する必要があります。
論理名は、このコネクターから変更イベントを受信する Kafka トピックの接頭辞として使用されるため、名前はクラスターのコネクター間で一意である必要があります。
コネクターを Avro コネクターと統合する場合、名前空間は関連する Kafka Connect スキーマの名前や、対応する Avro スキーマの名前空間でも使用されます。11
コネクターが変更イベントをキャプチャーするテーブルの一覧。
以下のコマンドを実行してコネクターリソースを作成します。
oc create -n <namespace> -f <kafkaConnector>.yaml
以下に例を示します。
oc create -n debezium -f {context}-inventory-connector.yaml
コネクターは Kafka Connect クラスターに登録され、
KafkaConnector
CR のspec.config.database.dbname
で指定されたデータベースに対して実行を開始します。コネクター Pod の準備ができると、Debezium が実行されます。
これで、Debezium Db2 のデプロイメントを確認 する準備が整いました。
3.6.4. Dockerfile からカスタム Kafka Connect コンテナーイメージをビルドして Debezium Db2 コネクターのデプロイ
Debezium Db2 コネクターをデプロイするには、Debezium コネクターアーカイブが含まれるカスタム Kafka Connect コンテナーイメージをビルドし、このコンテナーイメージをコンテナーレジストリーにプッシュする必要があります。次に、以下のカスタムリソース (CR) を作成する必要があります。
-
Kafka Connect インスタンスを定義する
KafkaConnect
CR。image
は Debezium コネクターを実行するために作成したイメージの名前を指定します。この CR を、Red Hat AMQ Streams がデプロイされている OpenShift インスタンスに適用します。AMQ Streams は、Apache Kafka を OpenShift に取り入れる operator およびイメージを提供します。 -
Debezium Db2 コネクターを定義する
KafkaConnector
CR。この CR をKafkaConnect
CR を適用するのと同じ OpenShift インスタンスに適用します。
前提条件
- Db2 が実行中で、Db2 を設定して Debezium コネクターと連携する 手順が完了済みである必要があります。
- AMQ Streams は OpenShift にデプロイされ、Apache Kafka および Kafka Connect が稼働している必要があります。詳細は、Deploying and Upgrading AMQ Streams on OpenShift を参照してください。
- Podman または Docker がインストールされている。
- Db2 に必要な JDBC ドライバーを 取得している。
-
Debezium コネクターを実行するコンテナーを追加する予定のコンテナーレジストリー (
quay.io
やdocker.io
など) でコンテナーを作成および管理するアカウントとパーミッションを持っている。
手順
Kafka Connect の Debezium Db2 コンテナーを作成します。
- Debezium Db2 コネクターアーカイブをダウンロードします。
Debezium Db2 コネクターアーカイブを展開して、コネクタープラグインのディレクトリー構造を作成します。以下に例を示します。
./my-plugins/ ├── debezium-connector-db2 │ ├── ...
registry.redhat.io/amq7/amq-streams-kafka-30-rhel8:2.0.0
をベースイメージとして使用して、新規の Dockerfile を作成します。たとえば、ターミナルウィンドウから以下のコマンドを入力します。my-plugins
はプラグインディレクトリーの名前に置き換えます。cat <<EOF >debezium-container-for-db2.yaml 1 FROM registry.redhat.io/amq7/amq-streams-kafka-30-rhel8:2.0.0 USER root:root COPY ./<my-plugins>/ /opt/kafka/plugins/ 2 USER 1001 EOF
このコマンドは、現在のディレクトリーに
debezium-container-for-db2.yaml
という名前の Docker ファイルを作成します。前のステップで作成した
debezium-container-for-db2.yaml
Docker ファイルからコンテナーイメージをビルドします。ファイルが含まれるディレクトリーから、ターミナルウィンドウを開き、以下のコマンドのいずれかを入力します。podman build -t debezium-container-for-db2:latest .
docker build -t debezium-container-for-db2:latest .
上記のコマンドは、
debezium-container-for-db2
という名前のコンテナーイメージを構築します。カスタムイメージを quay.io などのコンテナーレジストリーまたは内部のコンテナーレジストリーにプッシュします。コンテナーレジストリーは、イメージをデプロイする OpenShift インスタンスで利用できる必要があります。以下のいずれかのコマンドを実行します。
podman push <myregistry.io>/debezium-container-for-db2:latest
docker push <myregistry.io>/debezium-container-for-db2:latest
新しい Debezium Db2
KafkaConnect
カスタムリソース (CR) を作成します。たとえば、以下の例のようにannotations
およびimage
プロパティーを指定するdbz-connect.yaml
という名前のKafkaConnect
CR を作成します。apiVersion: kafka.strimzi.io/v1beta2 kind: KafkaConnect metadata: name: my-connect-cluster annotations: strimzi.io/use-connector-resources: "true" 1 spec: #... image: debezium-container-for-db2 2
以下のコマンドを入力して、
KafkaConnect
CR を OpenShift Kafka Connect 環境に適用します。oc create -f dbz-connect.yaml
このコマンドは、Debezium コネクターを実行するために作成したイメージの名前を指定する Kafka Connect インスタンスを追加します。
Debezium Db2 コネクターインスタンスを設定する
KafkaConnector
カスタムリソースを作成します。通常、コネクターに使用できる設定プロパティーを使用して、
.yaml
ファイルに Debezium Db2 コネクターを設定します。コネクター設定は、Debezium に対して、スキーマおよびテーブルのサブセットにイベントを生成するよう指示する可能性があり、または機密性の高い、大きすぎる、または不必要な指定のコラムで Debezium が値を無視、マスク、または切り捨てするようにプロパティーを設定する可能性もあります。以下の例では、ポート
50000
で Db2 サーバーホスト192.168.99.100
に接続する Debezium コネクターを設定します。このホストには、mydatabase
という名前のデータベース、名前がinventory
というテーブルがあり、fulfillment
がサーバーの論理名です。Db2
inventory-connector.yaml
apiVersion: kafka.strimzi.io/v1beta2 kind: KafkaConnector metadata: name: inventory-connector 1 labels: strimzi.io/cluster: my-connect-cluster annotations: strimzi.io/use-connector-resources: 'true' spec: class: io.debezium.connector.db2.Db2Connector 2 tasksMax: 1 3 config: 4 database.hostname: 192.168.99.100 5 database.port: 50000 6 database.user: db2inst1 7 database.password: Password! 8 database.dbname: mydatabase 9 database.server.name: fullfillment 10 database.include.list: public.inventory 11
表3.15 コネクター設定の説明 項目 説明 1
Kafka Connect クラスターに登録する場合のコネクターの名前。
2
この Db2 コネクタークラスの名前。
3
1 度に 1 つのタスクのみが動作する必要があります。
4
コネクターの設定。
5
Db2 インスタンスのアドレスであるデータベースホスト。
6
Db2 インスタンスのポート番号。
7
Db2 ユーザーの名前。
8
Db2 ユーザーのパスワード。
9
変更をキャプチャーするデータベースの名前。
10
namespace を形成する Db2 インスタンス/クラスターの論理名で、コネクターが書き込む Kafka トピックの名前、Kafka Connect スキーマ名、および Arvo コネクター が使用される場合に対応する Avro スキーマの namespace のすべてに使用されます。
11
Debezium が変更をキャプチャーする必要があるすべてのテーブルのリスト。
Kafka Connect でコネクターインスタンスを作成します。たとえば、
KafkaConnector
リソースをinventory-connector.yaml
ファイルに保存した場合は、以下のコマンドを実行します。oc apply -f inventory-connector.yaml
上記のコマンドは
inventory-connector
を登録し、コネクターはKafkaConnector
CR に定義されているmydatabase
データベースに対して実行を開始します。
Debezium Db2 コネクターに設定できる設定プロパティーの完全リストは、Db2 コネクタープロパティー を参照してください。
結果
コネクターが起動すると、コネクターが変更をキャプチャーするように設定された Db2 データベーステーブルの 整合性スナップショット が実行されます。その後、コネクターは行レベルの操作のデータ変更イベントの生成を開始し、変更イベントレコードを Kafka トピックにストリーミングします。
3.6.5. Debezium Db2 コネクターが実行していることの確認
コネクターがエラーなしで正常に起動すると、コネクターがキャプチャーするように設定された各テーブルのトピックが作成されます。ダウンストリームアプリケーションは、これらのトピックをサブスクライブして、ソースデータベースで発生する情報イベントを取得できます。
コネクターが実行されていることを確認するには、OpenShift Container Platform Web コンソールまたは OpenShift CLI ツール (oc) から以下の操作を実行します。
- コネクターのステータスを確認します。
- コネクターがトピックを生成していることを確認します。
- 各テーブルの最初のスナップショットの実行中にコネクターが生成する読み取り操作 ("op":"r") のイベントがトピックに反映されていることを確認します。
前提条件
- Debezium コネクターは AMQ Streams on OpenShift にデプロイされている。
-
OpenShift
oc
CLI クライアントがインストールされている。 - OpenShift Container Platform Web コンソールへのアクセスがある。
手順
以下の方法のいずれかを使用して
KafkaConnector
リソースのステータスを確認します。OpenShift Container Platform Web コンソールから以下を実行します。
- Home → Search に移動します。
-
Search ページで Resources をクリックし、Select Resource ボックスを開き、
KafkaConnector
を入力します。 - KafkaConnectors リストから、チェックするコネクターの名前をクリックします (例: inventory-connector-db2)。
- Conditions セクションで、Type および Status 列の値が Ready および True に設定されていることを確認します。
ターミナルウィンドウから以下を実行します。
以下のコマンドを入力します。
oc describe KafkaConnector <connector-name> -n <project>
以下に例を示します。
oc describe KafkaConnector inventory-connector-db2 -n debezium
このコマンドは、以下の出力のようなステータス情報を返します。
例3.3
KafkaConnector
リソースのステータスName: inventory-connector-db2 Namespace: debezium Labels: strimzi.io/cluster=debezium-kafka-connect-cluster Annotations: <none> API Version: kafka.strimzi.io/v1beta2 Kind: KafkaConnector ... Status: Conditions: Last Transition Time: 2021-12-08T17:41:34.897153Z Status: True Type: Ready Connector Status: Connector: State: RUNNING worker_id: 10.131.1.124:8083 Name: inventory-connector-db2 Tasks: Id: 0 State: RUNNING worker_id: 10.131.1.124:8083 Type: source Observed Generation: 1 Tasks Max: 1 Topics: inventory_connector_db2 inventory_connector_db2.inventory.addresses inventory_connector_db2.inventory.customers inventory_connector_db2.inventory.geom inventory_connector_db2.inventory.orders inventory_connector_db2.inventory.products inventory_connector_db2.inventory.products_on_hand Events: <none>
コネクターによって Kafka トピックが作成されたことを確認します。
OpenShift Container Platform Web コンソールから以
- Home → Search に移動します。
-
Search ページで Resources をクリックし、Select Resource ボックスを開き、
KafkaTopic
を入力します。 - KafkaTopics リストから確認するトピックの名前をクリックします (例: inventory-connector-db2.inventory.orders---ac5e98ac6a5d91e04d8ec0dc9078a1ece439081d)。
- Conditions セクションで、Type および Status 列の値が Ready および True に設定されていることを確認します。
ターミナルウィンドウから以下を実行します。
以下のコマンドを入力します。
oc get kafkatopics
このコマンドは、以下の出力のようなステータス情報を返します。
例3.4
KafkaTopic
リソースのステータスNAME CLUSTER PARTITIONS REPLICATION FACTOR READY connect-cluster-configs debezium-kafka-cluster 1 1 True connect-cluster-offsets debezium-kafka-cluster 25 1 True connect-cluster-status debezium-kafka-cluster 5 1 True consumer-offsets---84e7a678d08f4bd226872e5cdd4eb527fadc1c6a debezium-kafka-cluster 50 1 True inventory-connector-db2---a96f69b23d6118ff415f772679da623fbbb99421 debezium-kafka-cluster 1 1 True inventory-connector-db2.inventory.addresses---1b6beaf7b2eb57d177d92be90ca2b210c9a56480 debezium-kafka-cluster 1 1 True inventory-connector-db2.inventory.customers---9931e04ec92ecc0924f4406af3fdace7545c483b debezium-kafka-cluster 1 1 True inventory-connector-db2.inventory.geom---9f7e136091f071bf49ca59bf99e86c713ee58dd5 debezium-kafka-cluster 1 1 True inventory-connector-db2.inventory.orders---ac5e98ac6a5d91e04d8ec0dc9078a1ece439081d debezium-kafka-cluster 1 1 True inventory-connector-db2.inventory.products---df0746db116844cee2297fab611c21b56f82dcef debezium-kafka-cluster 1 1 True inventory-connector-db2.inventory.products-on-hand---8649e0f17ffcc9212e266e31a7aeea4585e5c6b5 debezium-kafka-cluster 1 1 True schema-changes.inventory debezium-kafka-cluster 1 1 True strimzi-store-topic---effb8e3e057afce1ecf67c3f5d8e4e3ff177fc55 debezium-kafka-cluster 1 1 True strimzi-topic-operator-kstreams-topic-store-changelog---b75e702040b99be8a9263134de3507fc0cc4017b debezium-kafka-cluster 1 1 True
トピックの内容を確認します。
- 端末画面で、以下のコマンドを入力します。
oc exec -n <project> -it <kafka-cluster> -- /opt/kafka/bin/kafka-console-consumer.sh \ > --bootstrap-server localhost:9092 \ > --from-beginning \ > --property print.key=true \ > --topic=<topic-name>
以下に例を示します。
oc exec -n debezium -it debezium-kafka-cluster-kafka-0 -- /opt/kafka/bin/kafka-console-consumer.sh \ > --bootstrap-server localhost:9092 \ > --from-beginning \ > --property print.key=true \ > --topic=inventory_connector_db2.inventory.products_on_hand
トピック名を指定する形式は、手順 1 で返された
oc describe
コマンドと同じです (例:inventory_connector_db2.inventory.addresses
)。トピックの各イベントについて、このコマンドは、以下の出力のような情報を返します。
例3.5 Debezium 変更イベントの内容
{"schema":{"type":"struct","fields":[{"type":"int32","optional":false,"field":"product_id"}],"optional":false,"name":"inventory_connector_db2.inventory.products_on_hand.Key"},"payload":{"product_id":101}} {"schema":{"type":"struct","fields":[{"type":"struct","fields":[{"type":"int32","optional":false,"field":"product_id"},{"type":"int32","optional":false,"field":"quantity"}],"optional":true,"name":"inventory_connector_db2.inventory.products_on_hand.Value","field":"before"},{"type":"struct","fields":[{"type":"int32","optional":false,"field":"product_id"},{"type":"int32","optional":false,"field":"quantity"}],"optional":true,"name":"inventory_connector_db2.inventory.products_on_hand.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":"string","optional":true,"name":"io.debezium.data.Enum","version":1,"parameters":{"allowed":"true,last,false"},"default":"false","field":"snapshot"},{"type":"string","optional":false,"field":"db"},{"type":"string","optional":true,"field":"sequence"},{"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.db2.Source","field":"source"},{"type":"string","optional":false,"field":"op"},{"type":"int64","optional":true,"field":"ts_ms"},{"type":"struct","fields":[{"type":"string","optional":false,"field":"id"},{"type":"int64","optional":false,"field":"total_order"},{"type":"int64","optional":false,"field":"data_collection_order"}],"optional":true,"field":"transaction"}],"optional":false,"name":"inventory_connector_db2.inventory.products_on_hand.Envelope"},"payload":{"before":null,"after":{"product_id":101,"quantity":3},"source":{"version":"1.7.2.Final-redhat-00001","connector":"db2","name":"inventory_connector_db2","ts_ms":1638985247805,"snapshot":"true","db":"inventory","sequence":null,"table":"products_on_hand","server_id":0,"gtid":null,"file":"db2-bin.000003","pos":156,"row":0,"thread":null,"query":null},"op":"r","ts_ms":1638985247805,"transaction":null}}
上記の例では、
payload
値は、コネクタースナップショットがテーブルinventory.products_on_hand
から 読み込み (op" ="r"
) イベントを生成したことを示しています。product_id
レコードのbefore
状態はnull
であり、レコードに以前の値が存在しないことを示します。"after"
状態がproduct_id
101
で項目のquantity
を3
で示しています。
3.6.6. Debezium Db2 コネクター設定プロパティーの説明
Debezium Db2 コネクターには、アプリケーションに適したコネクター動作を実現するために使用できる設定プロパティーが多数あります。多くのプロパティーにはデフォルト値があります。プロパティーに関する情報は、以下のように設定されています。
- 必要な設定プロパティー
- 高度な設定プロパティー
Debezium がデータベース履歴トピックから読み取るイベントを処理する方法を制御する データベース履歴コネクター設定プロパティー。
- データベースドライバーの動作を制御する パススルーデータベースドライバープロパティー。
必要な Debezium Db2 コネクター設定プロパティー
以下の設定プロパティーは、デフォルト値がない場合は必須です。
プロパティー | デフォルト | 説明 |
---|---|---|
デフォルトなし | コネクターの一意名。同じ名前で再登録を試みると失敗します。このプロパティーはすべての Kafka Connect コネクターに必要です。 | |
デフォルトなし |
コネクターの Java クラスの名前。Db2 コネクターには、常に | |
| このコネクターのために作成する必要のあるタスクの最大数。Db2 コネクターは常に単一のタスクを使用するため、この値を使用しません。そのため、デフォルト値は常に許容されます。 | |
デフォルトなし | Db2 データベースサーバーの IP アドレスまたはホスト名。 | |
| Db2 データベースサーバーの整数のポート番号。 | |
デフォルトなし | Db2 データベースサーバーに接続するための Db2 データベースユーザーの名前。 | |
デフォルトなし | Db2 データベースサーバーへの接続時に使用するパスワード。 | |
デフォルトなし | 変更をストリーミングする Db2 データベースの名前 | |
デフォルトなし | Debezium が変更をキャプチャーするデータベースをホストする特定の Db2 データベースサーバーの namespace を特定および提供する論理名。データベースサーバーの論理名には英数字とハイフン、ドット、アンダースコアのみを使用する必要があります。論理名は、他のコネクター全体で一意となる必要があります。これは、このコネクターからレコードを受信するすべての Kafka トピックのトピック名接頭辞として使用されるためです。 | |
デフォルトなし |
コネクターで変更をキャプチャーするテーブルの完全修飾テーブル識別子と一致する正規表現のコンマ区切りリスト (任意)。include リストに含まれていないテーブルの変更はキャプチャーされません。各識別子の形式は schemaName.tableName です。デフォルトでは、コネクターはシステム以外のテーブルすべての変更をキャプチャーします。また、 | |
デフォルトなし |
コネクターで変更をキャプチャーしないテーブルの完全修飾テーブル識別子と一致する正規表現のコンマ区切りリスト (任意)。コネクターは exclude リストに含まれていないシステム以外のテーブルごとに変更をキャプチャーします。各識別子の形式は schemaName.tableName です。また、 | |
空の文字列 | 変更イベント値から除外する列の完全修飾名と一致する正規表現のコンマ区切りリスト (任意)。列の完全修飾名の形式は schemaName.tableName.columnName です。プライマリーキー列は、値から除外された場合でも、イベントのキーに常に含まれます。 | |
該当なし |
文字ベースの列の完全修飾名と一致する正規表現のコンマ区切りリスト (任意)。列の完全修飾名の形式は schemaName.tableName.columnName です。作成された変更イベントレコードでは、指定された列の値は仮名に置き換えられます。
仮名は、指定された hashAlgorithm と salt を適用すると得られるハッシュ化された値で設定されます。使用されるハッシュ関数に基づいて、参照整合性は維持され、列値は仮名に置き換えられます。サポートされるハッシュ関数は、Java Cryptography Architecture Standard Algorithm Name Documentation の MessageDigest section に説明されています。 column.mask.hash.SHA-256.with.salt.CzQMA0cB5K = inventory.orders.customerName, inventory.shipment.customerName
必要な場合は、仮名は自動的に列の長さに短縮されます。コネクター設定には、異なるハッシュアルゴリズムと salt を指定する複数のプロパティーを含めることができます。 | |
|
時間、日付、およびタイムスタンプは、異なる精度の種類で表すことができます。 | |
|
廃棄 (tombstone) イベントの後に削除イベントが続くかどうかを制御します。 | |
| コネクターがデータベーススキーマの変更を、データベースサーバー ID と同じ名前の Kafka トピックに公開するかどうかを指定するブール値。各スキーマの変更は、データベース名が含まれるキーと、スキーマ更新を記述する JSON 構造である値で記録されます。これは、コネクターがデータベース履歴を内部で記録する方法には依存しません。 | |
該当なし |
文字ベースの列の完全修飾名と一致する正規表現のコンマ区切りリスト (任意)。列の完全修飾名の形式は schemaName.tableName.columnName です。変更イベントレコードでは、これらの列の値がプロパティー名の 長さ によって指定される文字数よりも長い場合は切り捨てられます。単一の設定で、異なる長さを持つ複数のプロパティーを指定できます。長さは正の整数である必要があります (例: | |
該当なし |
文字ベースの列の完全修飾名と一致する正規表現のコンマ区切りリスト (任意)。列の完全修飾名の形式は schemaName.tableName.columnName です。変更イベント値では、指定のテーブルコラムの値はアスタリスク ( | |
該当なし |
列の完全修飾名と一致する正規表現のコンマ区切りリスト (任意)。列の完全修飾名の形式は、databaseName.tableName.columnName または databaseName.schemaName.tableName.columnName です。 | |
該当なし |
一部の列のデータベース固有のデータ型名と一致する正規表現のコンマ区切りリスト (任意)。完全修飾データ型名の形式は、databaseName.tableName.typeName または databaseName.schemaName.tableName.typeName です。 | |
空の文字列 | 指定のテーブルの Kafka トピックに公開する変更イベントレコードのカスタムメッセージキーを形成するためにコネクターが使用する列を指定する式のリスト。
デフォルトでは、Debezium はテーブルのプライマリーキー列を、出力するレコードのメッセージキーとして使用します。デフォルトの代わりに、またはプライマリーキーのないテーブルのキーを指定するには、1 つ以上の列をもとにカスタムメッセージキーを設定できます。
プロパティーは複数のテーブルのエントリーをリストできます。リスト内の異なるテーブルのエントリーは、セミコロンを使用して、区切ります。 |
高度なコネクター設定プロパティー
以下の 高度な 設定プロパティーには、ほとんどの状況で機能するデフォルト設定があるため、コネクターの設定で指定する必要はほとんどありません。
プロパティー | デフォルト | 説明 |
---|---|---|
|
コネクターの起動時にスナップショットを実行する基準を指定します。 | |
|
スナップショットの実行中に、トランザクション分離レベルとキャプチャーモードのテーブルをロックする期間を制御します。使用できる値は次のとおりです。 | |
|
イベントの処理中にコネクターが例外を処理する方法を指定します。使用できる値は次のとおりです。 | |
| コネクターがイベントのバッチの処理を開始する前に、新しい変更イベントの発生を待つ期間をミリ秒単位で指定する正の整数値。デフォルトは 1000 ミリ秒 (1 秒) です。 | |
|
ブロッキングキューの最大サイズの正の整数値。コネクターは、データベースログから読み取る変更イベントをブロッキングキューに配置してから Kafka に書き込みます。このキューは、たとえば Kafka へのレコードの書き込みが遅い場合や Kafka が利用できない場合などに、変更データテーブルを読み取るためのバックプレシャーを提供できます。キューに表示されるイベントは、コネクターによって定期的に記録されるオフセットには含まれません。 | |
| コネクターが処理するイベントの各バッチの最大サイズを指定する正の整数値。 | |
| ブロッキングキューの最大サイズ (バイト単位) の long 値。この機能はデフォルトで無効になっています。正の long 値が設定されると有効になります。 | |
|
コネクターがハートビートメッセージを Kafka トピックに送信する頻度を制御します。デフォルトの動作では、コネクターはハートビートメッセージを送信しません。 | |
|
コネクターがハートビートメッセージを送信するトピック名の接頭辞を指定します。このトピック名の形式は | |
デフォルトなし | コネクターの起動時にスナップショットを実行するまでコネクターが待つ必要がある間隔 (ミリ秒単位)。クラスターで複数のコネクターを起動する場合、このプロパティーは、コネクターのリバランスが行われる原因となるスナップショットの中断を防ぐのに役立ちます。 | |
| スナップショットの実行中、コネクターは行のバッチでテーブルの内容を読み取ります。このプロパティーは、バッチの行の最大数を指定します。 | |
|
スナップショットの実行時に、テーブルロックを取得するまで待つ最大時間 (ミリ秒単位) を指定する正の整数値。コネクターがこの間隔でテーブルロックを取得できないと、スナップショットは失敗します。詳細は、コネクターによるスナップショットの実行方法 を参照してください。他の設定可能な値は以下のとおりです。 | |
デフォルトなし | スナップショットに追加するテーブル行を指定します。スナップショットにテーブルの行のサブセットのみを含める場合は、プロパティーを使用します。このプロパティーはスナップショットにのみ影響します。コネクターがログから読み取るイベントには影響しません。
プロパティーには、
スナップショットにソフト削除以外のレコードのみを含める場合は、soft-delete 列 ( "snapshot.select.statement.overrides": "customer.orders", "snapshot.select.statement.overrides.customer.orders": "SELECT * FROM [customers].[orders] WHERE delete_flag = 0 ORDER BY id DESC"
作成されるスナップショットでは、コネクターには | |
コネクターが
そうでない場合は | Avro の命名要件 に準拠するためにフィールド名がサニタイズされるかどうかを示します。 | |
|
コネクターがトランザクション境界でイベントを生成し、トランザクションメタデータで変更イベントエンベロープを強化するかどうかを決定します。コネクターにこれを実行させる場合は | |
デフォルトなし |
ストリーミング中にスキップされる操作タイプのコンマ区切りリスト。操作には、 | |
デフォルトなし |
シグナルをコネクターへの送信に使用されるデータコレクションの完全修飾名 。 シグナル機能はテクノロジープレビュー機能です。 | |
| 増分スナップショットのチャンクの実行中にコネクターがメモリーを取得して読み取る行の最大数。スナップショットは、サイズが大きいスナップショットの場合にはクエリーが少なくなるため、チャンクサイズを増やすと効率が上がります。ただし、チャンクサイズが大きい場合には、スナップショットデータのバッファーにより多くのメモリーが必要になります。チャンクサイズは、環境で最適なパフォーマンスを発揮できる値に、調整します。 増分スナップショットはテクノロジープレビュー機能です。 |
Debezium コネクターデータベース履歴設定プロパティー
Debezium には、コネクターがスキーマ履歴トピックと対話する方法を制御する database.history.*
プロパティーのセットが含まれています。
以下の表は、Debezium コネクターを設定するための database.history
プロパティーについて説明しています。
プロパティー | デフォルト | 説明 |
---|---|---|
コネクターがデータベーススキーマの履歴を保存する Kafka トピックの完全名。 | ||
Kafka クラスターへの最初の接続を確立するために コネクターが使用するホストとポートのペアの一覧。このコネクションは、コネクターによって以前に保存されたデータベーススキーマ履歴の取得や、ソースデータベースから読み取られる各 DDL ステートメントの書き込みに使用されます。各ペアは、Kafka Connect プロセスによって使用される同じ Kafka クラスターを示す必要があります。 | ||
| 永続化されたデータのポーリングが行われている間にコネクターが起動/回復を待つ最大時間 (ミリ秒単位) を指定する整数値。デフォルトは 100 ミリ秒です。 | |
|
エラーでコネクターのリカバリーが失敗する前に、コネクターが永続化された履歴データの読み取りを試行する最大回数。データが受信されなかった場合に最大待機する時間は、 | |
|
コネクターが不正または不明なデータベースのステートメントを無視するかどうか、または人が問題を修正するために処理を停止するかどうかを指定するブール値。安全なデフォルトは | |
今後のリリースで非推奨になり、削除される予定です。代わりに |
|
コネクターがすべての DDL ステートメントを記録するかどうかを指定するブール値
安全なデフォルトは |
|
コネクターがすべての DDL ステートメントを記録するかどうかを指定するブール値
安全なデフォルトは |
プロデューサーおよびコンシューマークライアントを設定するためのパススルーデータベース履歴プロパティー
Debezium は、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
Debezium は、プロパティーを Kafka クライアントに渡す前に、プロパティー名から接頭辞を削除します。
Kafka プロデューサー設定プロパティー および Kafka コンシューマー設定プロパティーの詳細は、Kafka のドキュメントを参照してください。
Debezium コネクターのパススルーデータベースドライバー設定プロパティー
Debezium コネクターでは、データベースドライバーのパススルー設定が可能です。パススルーデータベースプロパティーは、接頭辞 database.*
で始まります。たとえば、コネクターは database.foobar=false
などのプロパティーを JDBC URL に渡します。
データベース履歴クライアントのパススループロパティー の場合のように、Debezium はプロパティーから接頭辞を削除してからデータベースドライバーに渡します。
3.7. Debezium Db2 コネクターのパフォーマンスの監視
Debezium Db2 コネクターは、Apache Zookeeper、Apache Kafka、および Kafka Connect によって提供される JMX メトリクスの組み込みサポートに加えて、3 種類のメトリクスを提供します。
- スナップショットメトリクス は、スナップショットの実行中にコネクター操作に関する情報を提供します。
- メトリクスのストリーミング は、コネクターが変更をキャプチャーし、変更イベントレコードをストリーミングするときにコネクター操作に関する情報を提供します。
- スキーマ履歴メトリクス は、コネクターのスキーマ履歴の状態に関する情報を提供します。
Debezium モニターリングのドキュメント では、JMX を使用してこれらのメトリクスを公開する方法の詳細を提供します。
3.7.1. Db2 データベースのスナップショット作成時の Debezium の監視
MBean は debezium.db2:type=connector-metrics,context=snapshot,server=<db2.server.name>
です。
スナップショット操作がアクティブでない場合や、最後のコネクターの起動後にスナップショットの作成が発生した場合に、スナップショットメトリクスは公開されません。
以下の表は、利用可能なスナップショットのメトリックの一覧です。
属性 | 型 | 説明 |
---|---|---|
| コネクターが読み取りした最後のスナップショットイベント。 | |
| コネクターが最新のイベントを読み取りおよび処理してからの経過時間 (ミリ秒単位)。 | |
| 前回の開始またはリセット以降にコネクターで確認されたイベントの合計数。 | |
| コネクターに設定された include/exclude リストのフィルターリングルールによってフィルターされたイベントの数。 | |
|
| コネクターによって監視されるテーブルの一覧。 |
| コネクターによって取得されるテーブルの一覧。 | |
| snapshotter とメインの Kafka Connect ループの間でイベントを渡すために使用されるキューの長さ。 | |
| snapshotter とメインの Kafka Connect ループの間でイベントを渡すために使用されるキューの空き容量。 | |
| スナップショットに含まれているテーブルの合計数。 | |
| スナップショットによってまだコピーされていないテーブルの数。 | |
| スナップショットが起動されたかどうか。 | |
| スナップショットが中断されたかどうか。 | |
| スナップショットが完了したかどうか。 | |
| スナップショットが完了したかどうかに関わらず、これまでスナップショットにかかった時間 (秒単位)。 | |
| スナップショットの各テーブルに対してスキャンされる行数が含まれるマップ。テーブルは、処理中に増分がマップに追加されます。スキャンされた 10,000 行ごとに、テーブルの完成時に更新されます。 | |
|
キューの最大バッファー (バイト単位)。 | |
| キュー内のレコードの現在のデータ (バイト単位)。 |
コネクターは、増分スナップショットの実行時に、以下の追加のスナップショットメトリクスも提供します。
属性 | 型 | 説明 |
---|---|---|
| 現在のスナップショットチャンクの識別子。 | |
| 現在のチャンクを定義するプライマリーキーセットの下限。 | |
| 現在のチャンクを定義するプライマリーキーセットの上限。 | |
| 現在スナップショットされているテーブルのプライマリーキーセットの下限。 | |
| 現在スナップショットされているテーブルのプライマリーキーセットの上限。 |
増分スナップショットの使用はテクノロジープレビュー機能です。テクノロジープレビュー機能は、Red Hat 製品のサービスレベルアグリーメント (SLA) の対象外であり、機能的に完全ではないことがあります。Red Hat は実稼働環境でこれらを使用することを推奨していません。テクノロジープレビューの機能は、最新の製品機能をいち早く提供して、開発段階で機能のテストを行い、フィードバックを提供していただくことを目的としています。Red Hat のテクノロジープレビュー機能のサポート範囲に関する詳細は、テクノロジープレビュー機能のサポート範囲 を参照してください。
3.7.2. Debezium Db2 コネクターレコードストリーミングの監視
MBean は debezium.db2:type=connector-metrics,context=streaming,server=<db2.server.name>
です。
以下の表は、利用可能なストリーミングメトリクスの一覧です。
属性 | 型 | 説明 |
---|---|---|
| コネクターが読み取られた最後のストリーミングイベント。 | |
| コネクターが最新のイベントを読み取りおよび処理してからの経過時間 (ミリ秒単位)。 | |
| 前回の開始またはリセット以降にコネクターで確認されたイベントの合計数。 | |
| コネクターに設定された include/exclude リストのフィルターリングルールによってフィルターされたイベントの数。 | |
|
| コネクターによって監視されるテーブルの一覧。 |
| コネクターによって取得されるテーブルの一覧。 | |
| ストリーマーとメイン Kafka Connect ループの間でイベントを渡すために使用されるキューの長さ。 | |
| ストリーマーとメインの Kafka Connect ループの間でイベントを渡すために使用されるキューの空き容量。 | |
| コネクターが現在データベースサーバーに接続されているかどうかを示すフラグ。 | |
| 最後の変更イベントのタイムスタンプとそれを処理するコネクターとの間の期間 (ミリ秒単位)。この値は、データベースサーバーとコネクターが稼働しているマシンのクロック間の差異に対応します。 | |
| コミットされた処理済みトランザクションの数。 | |
| 最後に受信したイベントの位置。 | |
| 最後に処理されたトランザクションのトランザクション識別子。 | |
| キューの最大バッファー (バイト単位)。 | |
| キュー内のレコードの現在のデータ (バイト単位)。 |
3.7.3. Debezium Db2 コネクターのスキーマ履歴の監視
MBean は debezium.db2:type=connector-metrics,context=schema-history,server=<db2.server.name>
です。
以下の表は、利用可能なスキーマ履歴メトリクスの一覧です。
属性 | 型 | 説明 |
---|---|---|
|
データベース履歴の状態を示す | |
| リカバリーが開始された時点のエポック秒の時間。 | |
| リカバリーフェーズ中に読み取られた変更の数。 | |
| リカバリーおよびランタイム中に適用されるスキーマ変更の合計数。 | |
| 最後の変更が履歴ストアから復元された時点からの経過時間 (ミリ秒単位)。 | |
| 最後の変更が適用された時点からの経過時間 (ミリ秒単位)。 | |
| 履歴ストアから復元された最後の変更の文字列表現。 | |
| 最後に適用された変更の文字列表現。 |
3.8. Debezium Db2 コネクターの管理
Debezium Db2 コネクターをデプロイしたら、Debezium 管理 UDF を使用して、SQL コマンドで Db2 レプリケーション (ASN) を制御します。UDF によっては戻り値が必要な場合があります。この場合、SQL の VALUE
ステートメントを使用して呼び出します。その他の UDF には、SQL の CALL
ステートメントを使用します。
タスク | コマンドおよび注記 |
---|---|
| |
| |
| |
| |
| |
|
3.9. Debezium コネクターでのキャプチャーモードの Db2 テーブルのスキーマの更新
Debezium Db2 コネクターはスキーマ変更をキャプチャーできますが、スキーマを更新するには、データベース管理者と協力してコネクターが変更イベントの生成を継続するようにする必要があります。これは、Db2 がレプリケーションを実装する方法に必要です。
Db2 のレプリケーション機能は、キャプチャーモードのテーブルごとに、すべての変更が含まれる変更データテーブルをそのソーステーブルに作成します。ただし、変更データテーブルスキーマは静的です。キャプチャーモードのテーブルのスキーマを更新する場合は、対応する変更データテーブルのスキーマを更新する必要もあります。Debezium Db2 コネクターはこれを実行できません。昇格された権限を持つデータベース管理者は、キャプチャーモードのテーブルのスキーマを更新する必要があります。
同じテーブルの新しいスキーマ更新の前に、スキーマ更新の手順を完全に実行することが重要です。そのため、スキーマ更新の手順を 1 度で完了するために、すべての DDL を 1 つのバッチで実行することが推奨されます。
通常、テーブルスキーマを更新する手順は 2 つあります。
それぞれの方法に長所と短所があります。
3.9.1. Debezium Db2 コネクターでのオフラインスキーマ更新の実行
オフラインでスキーマの更新を行う前に、Debezium Db2 コネクターを停止します。これはより安全なスキーマ更新の手順ですが、高可用性の要件のあるアプリケーションには実現できない可能性があります。
前提条件
- スキーマの更新が必要なキャプチャーモードのテーブル 1 つ以上。
手順
- データベースを更新するアプリケーションを一時停止します。
- Debezium コネクターがストリーミングされていない変更イベントレコードをすべてストリーミングするまで待ちます。
- Debezium コネクターを停止します。
- すべての変更をソーステーブルスキーマに適用します。
-
ASN レジスターテーブルで、スキーマが更新されたテーブルを
INACTIVE
でマーク付けします。 - ASN キャプチャーサービスを再初期化します。
- キャプチャーモードからテーブルを削除するために Debezium UDF を実行 して、キャプチャーモードから古いスキーマを持つソーステーブルを削除します。
- テーブルをキャプチャーモードに追加するために Debezium UDF を実行 して、新しいスキーマを持つソーステーブルをキャプチャーモードに追加します。
-
ASN レジスターテーブルで、更新されたソーステーブルを
ACTIVE
としてマーク付けします。 - ASN キャプチャーサービスを再初期化します。
- データベースを更新するアプリケーションを再開します。
- Debezium コネクターを再起動します。
3.9.2. Debezium Db2 コネクターでのオンラインスキーマ更新の実行
オンラインスキーマの更新ではアプリケーションやデータ処理のダウンタイムは必要ありません。そのため、オンラインスキーマの更新を実行する前に Debezium Db2 コネクターを停止しません。また、オンラインスキーマの更新手順は、オフラインスキーマの更新手順よりも簡単です。
ただし、テーブルがキャプチャーモードの場合は、列名の変更後も Db2 レプリケーション機能は引き続き古い列名を使用します。新しい列名は、Debezium の変更イベントでは表示されません。変更イベントにある新しい列名を確認するには、コネクターを再起動する必要があります。
前提条件
- スキーマの更新が必要なキャプチャーモードのテーブル 1 つ以上。
テーブルの最後に列を追加する場合の手順
- 変更するスキーマのソーステーブルをロックします。
-
ASN レジスターテーブルで、ロックされたテーブルを
INACTIVE
としてマーク付けします。 - ASN キャプチャーサービスを再初期化します。
- ソーステーブルのスキーマにすべての変更を適用します。
- 対応する変更データテーブルのスキーマにすべての変更を適用します。
-
ASN レジスターテーブルで、ソーステーブルを
ACTIVE
としてマーク付けします。 - ASN キャプチャーサービスを再初期化します。
- 任意手順:コネクターを再起動して、変更イベントにある更新された列名を確認します。
テーブルの中に列を追加する場合の手順
- 変更するソーステーブルをロックします。
-
ASN レジスターテーブルで、ロックされたテーブルを
INACTIVE
としてマーク付けします。 - ASN キャプチャーサービスを再初期化します。
変更するソーステーブルごとに以下を行います。
- ソーステーブルのデータをエクスポートします。
- ソーステーブルを切り捨てます。
- ソーステーブルを変更して列を追加します。
- エクスポートしたデータを変更したソーステーブルに読み込みます。
- ソーステーブルの対応する変更データテーブルのデータをエクスポートします。
- 変更データテーブルを切り捨てます。
- 変更データテーブルを変更して、列を追加します。
- エクスポートしたデータを変更した変更データテーブルに読み込みます。
-
ASN レジスターテーブルで、テーブルを
INACTIVE
としてマーク付けします。これにより、古い変更データテーブルが非アクティブとしてマーク付けされるため、それらのテーブルにあるデータは保持されますが、更新されなくなります。 - ASN キャプチャーサービスを再初期化します。
- 任意手順:コネクターを再起動して、変更イベントにある更新された列名を確認します。
第4章 MongoDB の Debezium コネクター
Debezium の MongoDB コネクターは、データベースおよびコレクションにおけるドキュメントの変更に対して、MongoDB レプリカセットまたは MongoDB シャードクラスターを追跡し、これらの変更を Kafka トピックのイベントとして記録します。コネクターは、シャードクラスターにおけるシャードの追加または削除、各レプリカセットのメンバーシップの変更、各レプリカセット内の選出、および通信問題の解決待ちを自動的に処理します。
このコネクターと互換性のある MongoDB のバージョンについては、Debezium でサポートされる設定ページを参照してください。
Debezium MongoDB コネクターを使用するための情報および手順は、以下のように設定されています。
4.1. Debezium MongoDB コネクターの概要
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. Debezium MongoDB コネクターの仕組み
コネクターがサポートする MongoDB トポロジーの概要は、アプリケーションを計画するときに役立ちます。
MongoDB コネクターが設定およびデプロイされると、シードアドレスの MongoDB サーバーに接続して起動し、利用可能な各レプリカセットの詳細を判断します。各レプリカセットには独立した独自の oplog があるため、コネクターはレプリカセットごとに個別のタスクの使用を試みます。コネクターは、使用するタスクの最大数を制限でき、十分なタスクが利用できない場合は、コネクターは各タスクに複数のレプリカセットを割り当てます。ただし、タスクはレプリカセットごとに個別のスレッドを使用します。
シャードクラスターに対してコネクターを実行する場合は、レプリカセットの数よりも大きい tasks.max
の値を使用します。これにより、コネクターはレプリカセットごとに 1 つのタスクを作成でき、Kafka Connect が利用可能なワーカープロセス全体でタスクを調整、配布、および管理できるようにします。
Debezium MongoDB コネクターの仕組みの詳細は、以下を参照してください。
- 「Debezium コネクターでサポートされる MongoDB トポロジー」
- 「Debezium MongoDB コネクターでレプリカセットおよびシャードクラスターに論理名を使用する方法」
- 「Debezium MongoDB コネクターでのスナップショットの実行方法」
- 「Debezium MongoDB コネクターでの変更イベントレコードのストリーミング方法」
- 「Debezium MongoDB 変更イベントレコードを受信する Kafka トピックのデフォルト名」
- 「イベントキーが Debezium MongoDB コネクターのトピックパーティション設定を制御する方法」
- 「トランザクション境界を表す Debezium MongoDB コネクターによって生成されたイベント」
4.2.1. Debezium コネクターでサポートされる MongoDB トポロジー
MongoDB コネクターは以下の MongoDB トポロジーをサポートします。
- MongoDB レプリカセット
Debezium MongoDB コネクターは単一の MongoDB レプリカセットから変更をキャプチャーできます。実稼働のレプリカセットには、少なくとも 3 つのメンバー が必要です。
レプリカセットで MongoDB コネクターを使用するには、コネクターの
mongodb.hosts
プロパティーを使用して、1 つ以上のレプリカセットサーバーのアドレスを シードアドレス として提供します。コネクターはこれらのシードを使用してレプリカセットに接続した後、レプリカセットからメンバーの完全セットを取得し、どのメンバーがプライマリーであるかを認識します。コネクターは、プライマリーに接続するタスクを開始し、プライマリーの oplog から変更をキャプチャーします。レプリカセットが新しいプライマリーを選出すると、タスクは自動的に新しいプライマリーに切り替えます。注記MongoDB がプロキシーと面する場合 (Docker on OS X や Windows などのように)、クライアントがレプリカセットに接続し、メンバーを検出すると、MongoDB クライアントはプロキシーを有効なメンバーから除外し、プロキシーを経由せずに直接メンバーに接続しようとし、失敗します。
このような場合、コネクターのオプションの
mongodb.members.auto.discover
設定プロパティーをfalse
に設定して、コネクターにメンバーシップの検出を見送るように指示し、代わりに最初のシードアドレス (mongodb.hosts
プロパティーによって指定) をプライマリーノードとして使用するよう指示します。これは機能する可能性がありますが、選出が行われるときに問題が発生します。
- MongoDB のシャードクラスター
MongoDB のシャードクラスター は以下で設定されます。
- レプリカセットとしてデプロイされる 1 つ以上のシャード。
- クラスターの設定サーバーとして動作する個別のレプリカセット。
クライアントが接続し、要求を適切なシャードにルーティングする 1 つ以上の ルーター (
mongos
とも呼ばれます)。シャードクラスターで MongoDB コネクターを使用するには、コネクターを設定サーバーレプリカセットのホストアドレスで設定します。コネクターがこのレプリカセットに接続すると、シャードクラスターの設定サーバーとして動作していることを検出し、クラスターでシャードとして使用される各レプリカセットに関する情報を検出した後、各レプリカセットから変更をキャプチャーするために別のタスクを起動します。新しいシャードがクラスターに追加される場合または既存のシャードが削除される場合、コネクターはそのタスクを自動的に調整します。
- MongoDB スタンドアロンサーバー
- スタンドアロンサーバーには oplog がないため、MongoDB コネクターはスタンドアロン MongoDB サーバーの変更を監視できません。スタンドアロンサーバーが 1 つのメンバーを持つレプリカセットに変換されると、コネクターが動作します。
MongoDB は、実稼働でのスタンドアロンサーバーの実行を推奨しません。詳細は MariaDB のドキュメント を参照してください。
4.2.2. Debezium MongoDB コネクターでレプリカセットおよびシャードクラスターに論理名を使用する方法
コネクター設定プロパティー mongodb.name
は、MongoDB レプリカセットまたはシャードされたクラスターの 論理名 として提供されます。コネクターは、論理名をさまざまな方法で使用します。すべてトピック名のプレフィックとして使用したり、各レプリカセットの oplog の位置を記録するときに一意の識別子として使用したりします。
各 MongoDB コネクターに、ソース MongoDB システムを意味する一意の論理名を命名する必要があります。論理名は、アルファベットまたはアンダースコアで始まり、残りの文字を英数字またはアンダースコアとすることが推奨されます。
4.2.3. Debezium MongoDB コネクターでのスナップショットの実行方法
タスクがレプリカセットを使用して起動すると、コネクターの論理名とレプリカセット名を使用して、コネクターが変更の読み取りを停止した位置を示す オフセット を検出します。オフセットが検出され、oplog に存在する場合、タスクは記録されたオフセットの位置から即座に ストリームの変更 を続行します。
ただし、オフセットが見つからない場合や、oplog にその位置が含まれなくなった場合、タスクは スナップショット を実行してレプリカセットの内容の現在の状態を取得する必要があります。このプロセスは、oplog の現在の位置を記録して開始され、オフセット (スナップショットが開始されたことを示すフラグとともに) として記録します。その後、タスクは各コレクションをコピーし、できるだけ多くのスレッドを生成し (snapshot.max.threads
設定プロパティーの値まで)、この作業を並行して行います。コネクターは、確認した各ドキュメントの個別の 読み取りイベント を記録します。読み取りイベントにはオブジェクトの識別子、オブジェクトの完全な状態、およびオブジェクトが見つかった MongoDB レプリカセットの ソース 情報が含まれます。ソース情報には、スナップショット中にイベントが生成されたことを示すフラグも含まれます。
このスナップショットは、コネクターのフィルターと一致するすべてのコレクションがコピーされるまで継続されます。タスクのスナップショットが完了する前にコネクターが停止した場合は、コネクターを再起動すると、再びスナップショットを開始します。
コネクターがレプリカセットのスナップショットを実行している間は、タスクの再割り当てと再設定をしないようにします。コネクターはスナップショットの進捗とともにメッセージをログに記録します。最大限の制御を行う場合は、各コネクターに対して Kafka Connect の個別のクラスターを実行します。
4.2.4. Debezium MongoDB コネクターでの変更イベントレコードのストリーミング方法
レプリカセットレコードのコネクタータスクがオフセットを取得すると、オフセットを使用して変更のストリーミングを開始する oplog の位置を判断します。その後、タスクはレプリカセットのプライマリーノードに接続し、その位置から変更のストリーミングを開始します。すべての作成、挿入、および削除操作を処理して Debezium の 変更イベント に変換します。各変更イベントには操作が検出された oplog の位置が含まれ、コネクターはこれを最新のオフセットとして定期的に記録します。オフセットが記録される間隔は、Kafka Connect ワーカー設定プロパティーである offset.flush.interval.ms
によって制御されます。
コネクターが正常に停止されると、処理された最後のオフセットが記録され、再起動時にコネクターは停止した場所から続行されます。しかし、コネクターのタスクが予期せず終了した場合、最後にオフセットが記録された後、最後のオフセットが記録される前に、タスクによってイベントが処理および生成されることがあります。再起動時に、コネクターは最後に 記録された オフセットから開始し、クラッシュの前に生成された同じイベントを生成する可能性があります。
すべてが通常どおり動作している場合、Kafka コンシューマーは実際にすべてのメッセージを 1 度だけ 確認します。ただし、問題が発生した場合は、Kafka はコンシューマーが 少なくとも 1 度 各メッセージを確認することのみを保証します。したがって、コンシューマーが複数回メッセージを確認することを想定する必要があります。
前述のように、コネクタータスクは常にレプリカセットのプライマリーノードを使用して oplog からの変更をストリーミングし、コネクターが可能な限り最新の操作を確認できるようにし、代わりにセカンダリーが使用された場合よりも短いレイテンシーで変更をキャプチャーできるようにします。レプリカセットが新しいプライマリーを選出すると、コネクターは即座に変更のストリーミングを停止し、新しいプライマリーに接続して、同じ場所にある新しいプライマリーノードから変更のストリーミングを開始します。同様に、コネクターとレプリカセットメンバーとの通信で問題が発生した場合は、レプリカセットが過剰にならないように指数バックオフを使用して再接続を試みます。接続の確立後、停止した場所から変更のストリーミングを続行します。これにより、コネクターはレプリカセットメンバーシップの変更を動的に調整でき、通信障害を自動的に処理できます。
要約すると、MongoDB コネクターはほとんどの状況で実行を継続します。通信の問題により、問題が解決されるまでコネクターが待機する可能性があります。
4.2.5. Debezium MongoDB 変更イベントレコードを受信する Kafka トピックのデフォルト名
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.2.6. イベントキーが Debezium MongoDB コネクターのトピックパーティション設定を制御する方法
MongoDB コネクターは、イベントのトピックパーティションを明示的に決定しません。代わりに、Kafka はイベントキーに基づいてトピックのパーティションを作成する方法を決定できます。Kafka Connect ワーカー設定に Partitioner
実装の名前を定義することで、Kafka のパーティショニングロジックを変更できます。
Kafka は、1 つのトピックパーティションに書き込まれたイベントのみ、合計順序を維持します。キーでイベントのパーティションを行うと、同じキーを持つすべてのイベントは常に同じパーティションに移動します。これにより、特定のドキュメントのすべてのイベントが常に完全に順序付けされます。
4.2.7. トランザクション境界を表す Debezium MongoDB コネクターによって生成されたイベント
Debezium は、トランザクションメタデータ境界を表すイベントを生成でき、データイベントメッセージを補完できます。
Debezium は、コネクターのデプロイ後に発生するトランザクションに対してのみメタデータを登録し、受信します。コネクターをデプロイする前に発生するトランザクションのメタデータは利用できません。
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.collectiona", "event_count": 1 }, { "data_collection": "rs0.testDB.collectionb", "event_count": 1 } ] }
transaction.topic
オプションでオーバーライドされない限り、トランザクションイベントは database.server.name.transaction
という名前のトピックに書き込まれます。
変更データイベントのエンリッチメント
トランザクションメタデータを有効にすると、データメッセージ Envelope
は新しい transaction
フィールドでエンリッチされます。このフィールドは、複合フィールドの形式ですべてのイベントに関する情報を提供します。
id
- 一意のトランザクション識別子の文字列表現。
total_order
- トランザクションによって生成されたすべてのイベントを対象とするイベントの絶対位置。
data_collection_order
- トランザクションによって出力されたすべてのイベントを対象とするイベントのデータコレクションごとの位置。
以下は、メッセージの内容の例です。
{ "patch": null, "after": "{\"_id\" : {\"$numberLong\" : \"1004\"},\"first_name\" : \"Anne\",\"last_name\" : \"Kretchmar\",\"email\" : \"annek@noanswer.org\"}", "source": { ... }, "op": "c", "ts_ms": "1580390884335", "transaction": { "id": "1462833718356672513", "total_order": "1", "data_collection_order": "1" } }
4.3. Debezium MongoDB コネクターのデータ変更イベントの説明
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.3.1. Debezium MongoDB 変更イベントのキー
変更イベントのキーには、変更されたドキュメントのキーのスキーマと、変更されたドキュメントの実際のキーのスキーマが含まれます。特定のコレクションでは、スキーマとそれに対応するペイロードの両方に単一の 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 |
|
文字列 | "1234" |
|
Document |
|
|
ObjectId |
|
|
バイナリー |
|
|
4.3.2. Debezium MongoDB 変更イベントの値
変更イベントの値はキーよりも若干複雑です。キーと同様に、値には schema
セクションと payload
セクションがあります。schema
セクションには、入れ子のフィールドを含む、 Envelope
セクションの payload
構造を記述するスキーマが含まれています。データを作成、更新、または削除する操作のすべての変更イベントには、Envelope 構造を持つ値 payload があります。
変更イベントキーの例を紹介するために使用した、同じサンプルドキュメントについて考えてみましょう。
ドキュメントの例
{ "_id": 1004, "first_name": "Anne", "last_name": "Kretchmar", "email": "annek@noanswer.org" }
このドキュメントへの変更に対する変更イベントの値部分には、以下の各イベントタイプについて記述されています。
作成 イベント
以下の例は、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": "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.7.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 のシステムクロックを基にします。 |
更新イベント
サンプル 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.7.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 コネクターはこの状態を維持できません。
削除 イベント
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.7.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. Debezium コネクターと連携する MongoDB の設定
MongoDB コネクターは MongoDB の oplog を使用して変更をキャプチャーするため、コネクターは MongoDB レプリカセットと、各シャードが個別のレプリカセットであるシャードクラスターとのみ動作します。レプリカセット または シャードクラスター の設定については、MongoDB ドキュメントを参照してください。また、レプリカセットで アクセス制御と認証 を有効にする方法についても理解するようにしてください。
oplog が読み取られる admin
データベースを読み取るために適切なロールを持つ MongoDB ユーザーも必要です。さらに、ユーザーはシャードクラスターの設定サーバーで config
データベースを読み取りできる必要もあり、listDatabases
権限も必要です。
4.5. Debezium MongoDB コネクターのデプロイメント
以下の方法のいずれかを使用して Debezium MongoDB コネクターをデプロイできます。
4.5.1. AMQ Streams を使用した MongoDB コネクターデプロイメント
Debezium 1.7 以降、Debezium コネクターのデプロイに推奨される方法は、AMQ Streams を使用してコネクタープラグインが含まれる Kafka Connect コンテナーイメージをビルドすることです。
デプロイメントプロセス中に、以下のカスタムリソース (CR) を作成し、使用します。
-
Kafka Connect インスタンスを定義し、コネクターアーティファクトに関する情報をイメージに含める必要がある
KafkaConnect
CR。 -
コネクターがソースデータベースにアクセスするために使用する情報を提供する
KafkaConnector
CR。AMQStreams が Kafka Connect Pod を開始した後、KafkaConnector
CR を適用してコネクターを開始します。
Kafka Connect イメージのビルド仕様では、デプロイ可能なコネクターを指定できます。各コネクタープラグインに対して、デプロイメントに利用可能にする他のコンポーネントを指定することもできます。たとえば、Service Registry アーティファクトまたは Debezium スクリプトコンポーネントを追加できます。AMQ Streams が Kafka Connect イメージをビルドすると、指定のアーティファクトをダウンロードし、イメージに組み込みます。
Kafka Connect CR の spec.build.output
パラメーターは、生成される KafkaConnect
コンテナーイメージを格納する場所を指定します。コンテナーイメージは Docker レジストリーまたは OpenShift ImageStream に保存できます。イメージを ImageStream に保存するには、Kafka Connect をデプロイする前に ImageStream を作成する必要があります。イメージストリームは自動的に作成されません。
KafkaConnect
リソースを使用してクラスターを作成する場合は、Kafka Connect REST API を使用してコネクターを作成または更新できません。ただし、REST API を使用して情報を取得できます。
関連情報
- AMQ Streams on OpenShift の使用のKafka Connect の設定を参照してください。
- AMQ Streams を使用した新しいコンテナーイメージの自動作成と OpenShift での AMQ Streams のアップグレード
4.5.2. AMQ Streams を使用した Debezium MongoDB コネクターのデプロイ
以前のバージョンの AMQ Streams では、OpenShift に Debezium コネクターをデプロイするには、最初にコネクター用の Kafka Connect イメージをビルドする必要がありました。コネクターを OpenShift にデプロイするのに現在推奨される方法は、AMQ Streams でビルド設定を使用して、使用する Debezium コネクタープラグインが含まれる Kafka Connect コンテナーイメージを自動的にビルドすることです。
ビルドプロセス中、AMQ Streams Operator は Debezium コネクター定義を含む KafkaConnect
カスタムリソースの入力パラメーターを Kafka Connect コンテナーイメージに変換します。このビルドは、Red Hat Maven リポジトリーまたは別の設定済みの HTTP サーバーから必要なアーティファクトをダウンロードします。新規に作成されたコンテナーは .spec.build.output
に指定されるコンテナーレジストリーにプッシュされ、Kafka Connect Pod のデプロイに使用されます。AMQ Streams が Kafka Connect イメージをビルドしたら、KafkaConnector
カスタムリソースを作成し、ビルドに含まれるコネクターを起動します。
前提条件
- クラスター Operator がインストールされている OpenShift クラスターにアクセスできる必要があります。
- AMQ Streams Operator が稼働している必要があります。
- Kafka クラスターは、Apache Open Shift での AMQ ストリームのデプロイとアップグレードに記載されているようにデプロイされます。
- Red Hat Integration ライセンスがある。
- Kafka Connect is deployed on AMQ Streams。
-
OpenShift
oc
CLI クライアントがインストールされている、または OpenShift Container Platform Web コンソールにアクセスできる。 Kafka Connect ビルドイメージの保存方法に応じて、レジストリーのパーミッションが必要であるか、ImageStream リソースを作成する必要があります。
- ビルドイメージを Red Hat Quay.io または Docker Hub などのイメージレジストリーに保存するには、以下を実行します。
- レジストリーでイメージを作成し、管理するためのアカウントおよびパーミッション。
- ビルドイメージをネイティブ OpenShift ImageStream として保存します。
- ImageStream リソースがクラスターにデプロイされている。クラスターの ImageStream を明示的に作成する必要があります。ImageStreams はデフォルトでは利用できません。
手順
- OpenShift クラスターにログインします。
コネクターの Debezium
KafkaConnect
カスタムリソース (CR) を作成するか、既存のリソースを変更します。たとえば、以下の例のようにmetadata.annotations
およびspec.build
プロパティーを指定するKafkaConnect
CR を作成します。dbz-connect.yaml
などの名前でファイルを保存します。例4.1 Debezium コネクターを含む
KafkaConnect
カスタムリソースを定義するdbz-connect.yaml
ファイルapiVersion: kafka.strimzi.io/v1beta2 kind: KafkaConnect metadata: name: debezium-kafka-connect-cluster annotations: strimzi.io/use-connector-resources: "true" 1 spec: version: 3.00 build: 2 output: 3 type: imagestream 4 image: debezium-streams-connect:latest plugins: 5 - name: debezium-connector-mongodb artifacts: - type: zip 6 url: https://maven.repository.redhat.com/ga/io/debezium/debezium-connector-mongodb/1.7.2.Final-redhat-<build_number>/debezium-connector-mongodb-1.7.2.Final-redhat-<build_number>-plugin.zip 7 - type: zip url: https://maven.repository.redhat.com/ga/io/apicurio/apicurio-registry-distro-connect-converter/2.0-redhat-<build-number>/apicurio-registry-distro-connect-converter-2.0-redhat-<build-number>.zip - type: zip url: https://maven.repository.redhat.com/ga/io/debezium/debezium-scripting/1.7.2.Final/debezium-scripting-1.7.2.Final.zip bootstrapServers: debezium-kafka-cluster-kafka-bootstrap:9093
表4.7 Kafka Connect 設定の説明 項目 説明 1
strimzi.io/use-connector-resources
アノテーションをtrue
に設定して、クラスターオペレーターがKafkaConnector
リソースを使用してこの Kafka Connect クラスター内のコネクターを設定できるようにします。2
spec.build
設定は、ビルドイメージの保存場所を指定し、プラグインアーティファクトの場所と共にイメージに追加するプラグインを一覧表示します。3
build.output
は、新たにビルドされたイメージが保存されるレジストリーを指定します。4
イメージ出力の名前およびイメージ名を指定します。
output.type
の有効な値は、Docker Hub や Quay などのコンテナーレジストリーにプッシュする場合はdocker
、内部の OpenShift ImageStream にイメージをプッシュする場合はimagestream
です。ImageStream を使用するには、ImageStream リソースをクラスターにデプロイする必要があります。KafkaConnect 設定でbuild.output
の指定に関する詳細は、AMQ Streams Build スキーマ参照 のドキュメントを参照 してください。5
plugins
設定は、Kafka Connect イメージに追加するすべてのコネクターを一覧表示します。一覧の各エントリーについて、プラグインname
と、コネクターのビルドに必要なアーティファクトに関する情報を指定します。任意で、各コネクタープラグインに対して、コネクターと使用できる他のコンポーネントを含めることができます。たとえば、Service Registry アーティファクトまたは Debezium スクリプトコンポーネントを追加できます。6
artifacts.type
の値は、artifacts.url
で指定したアーティファクトのファイルタイプを指定します。有効なタイプはzip
、tgz
、またはjar
です。Debezium コネクターアーカイブは、.zip
ファイル形式で提供されます。JDBC ドライバーファイルは.jar
形式です。type
の値は、url
フィールドで参照されるファイルのタイプと一致する必要があります。7
artifacts.url
の値は、コネクターアーティファクトのファイルを格納する Maven リポジトリーなどの HTTP サーバーのアドレスを指定します。OpenShift クラスターは指定されたサーバーにアクセスできる必要があります。以下のコマンドを入力して、
KafkaConnect
ビルド仕様を OpenShift クラスターに適用します。oc create -f dbz-connect.yaml
Streams Operator はカスタムリソースで指定された設定に基づいて、デプロイする Kafka Connect イメージを準備します。
ビルドが完了すると、Operator はイメージを指定されたレジストリーまたは ImageStream にプッシュし、Kafka Connect クラスターを起動します。設定に一覧表示されているコネクターアーティファクトはクラスターで利用できます。KafkaConnector
リソースを作成し、デプロイする各コネクターのインスタンスを定義します。
たとえば、以下のKafkaConnector
CR を作成し、mongodb-inventory-connector.yaml
として保存します。例4.2 Debezium コネクターの
KafkaConnector
カスタムリソースを定義するmongodb-inventory-connector.yaml
ファイルapiVersion: kafka.strimzi.io/v1beta2 kind: KafkaConnector metadata: labels: strimzi.io/cluster: debezium-kafka-connect-cluster name: inventory-connector-mongodb 1 spec: class: io.debezium.connector.mongodb.MongoDbConnector 2 tasksMax: 1 3 config: 4 database.history.kafka.bootstrap.servers: 'debezium-kafka-cluster-kafka-bootstrap.debezium.svc.cluster.local:9092' database.history.kafka.topic: schema-changes.inventory database.hostname: mongodb.debezium-mongodb.svc.cluster.local 5 database.port: 3306 6 database.user: debezium 7 database.password: dbz 8 database.dbname: mydatabase 9 database.server.name: inventory_connector_mongodb 10 database.include.list: public.inventory 11
表4.8 コネクター設定の説明 項目 説明 1
Kafka Connect クラスターに登録するコネクターの名前。
2
コネクタークラスの名前。
3
同時に動作できるタスクの数。
4
コネクターの設定。
5
ホストデータベースインスタンスのアドレス。
6
データベースインスタンスのポート番号。
7
Debezium がデータベースに接続するユーザーアカウントの名前。
8
データベースユーザーアカウントのパスワード
9
変更をキャプチャーするデータベースの名前。
10
データベースインスタンスまたはクラスターの論理名。
指定の名前は英数字またはアンダースコアからのみ形成する必要があります。
論理名は、このコネクターから変更イベントを受信する Kafka トピックの接頭辞として使用されるため、名前はクラスターのコネクター間で一意である必要があります。
コネクターを Avro コネクターと統合する場合、名前空間は関連する Kafka Connect スキーマの名前や、対応する Avro スキーマの名前空間でも使用されます。11
コネクターが変更イベントをキャプチャーするテーブルの一覧。
以下のコマンドを実行してコネクターリソースを作成します。
oc create -n <namespace> -f <kafkaConnector>.yaml
以下に例を示します。
oc create -n debezium -f {context}-inventory-connector.yaml
コネクターは Kafka Connect クラスターに登録され、
KafkaConnector
CR のspec.config.database.dbname
で指定されたデータベースに対して実行を開始します。コネクター Pod の準備ができると、Debezium が実行されます。
これで、Debezium MongoDB のデプロイメントを確認 する準備が整いました。
4.5.3. Dockerfile からカスタム Kafka Connect コンテナーイメージをビルドして Debezium MongoDB コネクターのデプロイ
Debezium MongoDB コネクターをデプロイするには、Debezium コネクターアーカイブが含まれるカスタム Kafka Connect コンテナーイメージをビルドし、このコンテナーイメージをコンテナーレジストリーにプッシュする必要があります。次に、2 つのカスタムリソース (CR) を作成します。
-
Kafka Connect インスタンスを定義する
KafkaConnect
CR。image
は Debezium コネクターを実行するために作成したイメージの名前を指定します。この CR を、Red Hat AMQ Streams がデプロイされている OpenShift インスタンスに適用します。AMQ Streams は、Apache Kafka を OpenShift に取り入れる operator およびイメージを提供します。 -
Debezium MongoDB コネクターを定義する
KafkaConnector
CR。この CR をKafkaConnect
CR を適用するのと同じ OpenShift インスタンスに適用します。
前提条件
- MongoDB が稼働し、MongoDB を設定して Debezium コネクターと連携する 手順が完了済みである必要があります。
- AMQ Streams は OpenShift にデプロイされ、Apache Kafka および Kafka Connect が稼働している必要があります。詳細は、Deploying and Upgrading AMQ Streams on OpenShift を参照してください。
- Podman または Docker がインストールされている。
-
Debezium コネクターを実行するコンテナーを追加する予定のコンテナーレジストリー (
quay.io
やdocker.io
など) でコンテナーを作成および管理するアカウントとパーミッションを持っている。
手順
Kafka Connect の Debezium MongoDB コンテナーを作成します。
- Debezium MongoDB コネクターアーカイブをダウンロードします。
Debezium MongoDB コネクターアーカイブを展開して、コネクタープラグインのディレクトリー構造を作成します。以下に例を示します。
./my-plugins/ ├── debezium-connector-mongodb │ ├── ...
registry.redhat.io/amq7/amq-streams-kafka-30-rhel8:2.0.0
をベースイメージとして使用して、新規の Dockerfile を作成します。たとえば、ターミナルウィンドウから以下のコマンドを入力します。my-plugins
はプラグインディレクトリーの名前に置き換えます。cat <<EOF >debezium-container-for-mongodb.yaml 1 FROM registry.redhat.io/amq7/amq-streams-kafka-30-rhel8:2.0.0 USER root:root COPY ./<my-plugins>/ /opt/kafka/plugins/ 2 USER 1001 EOF
このコマンドは、現在のディレクトリーに
debezium-container-for-mongodb.yaml
という名前の Dockerfile を作成します。前のステップで作成した
debezium-container-for-mongodb.yaml
Docker ファイルからコンテナーイメージをビルドします。ファイルが含まれるディレクトリーから、ターミナルウィンドウを開き、以下のコマンドのいずれかを入力します。podman build -t debezium-container-for-mongodb:latest .
docker build -t debezium-container-for-mongodb:latest .
上記のコマンドは、
debezium-container-for-mongodb
という名前のコンテナーイメージを構築します。カスタムイメージを
quay.io
などのコンテナーレジストリーまたは内部のコンテナーレジストリーにプッシュします。コンテナーレジストリーは、イメージをデプロイする OpenShift インスタンスで利用できる必要があります。以下のいずれかのコマンドを実行します。podman push <myregistry.io>/debezium-container-for-mongodb:latest
docker push <myregistry.io>/debezium-container-for-mongodb:latest
新しい Debezium MongoDB
KafkaConnect
カスタムリソース (CR) を作成します。たとえば、以下の例のようにannotations
およびimage
プロパティーを指定するdbz-connect.yaml
という名前のKafkaConnect
CR を作成します。apiVersion: kafka.strimzi.io/v1beta2 kind: KafkaConnect metadata: name: my-connect-cluster annotations: strimzi.io/use-connector-resources: "true" 1 spec: #... image: debezium-container-for-mongodb 2
以下のコマンドを入力して、
KafkaConnect
CR を OpenShift Kafka Connect 環境に適用します。oc create -f dbz-connect.yaml
このコマンドは、Debezium コネクターを実行するために作成したイメージの名前を指定する Kafka Connect インスタンスを追加します。
Debezium PostgreSQL コネクターインスタンスを設定する
KafkaConnector
カスタムリソースを作成します。通常、コネクターに使用できる設定プロパティーを使用して、
.yaml
ファイルに Debezium MongoDB コネクターを設定します。コネクター設定で、Debezium に指示を出して MongoDB レプリカセットまたはシャードクラスターのサブセットの変更イベントを生成する場合があります。任意で、不必要なコレクションを除外するプロパティーを設定できます。以下の例では、MongoDB レプリカセット
rs0
を192.168.99.100
のポート27017
に接続し、インベントリー
コレクションで発生した変更をキャプチャーする Debezium コネクターを設定します。fullfillment
はレプリカセットの論理名です。MongoDB
inventory-connector.yaml
apiVersion: kafka.strimzi.io/v1beta2 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
- コネクターを Kafka Connect に登録するために使用される名前。
- 2
- MongoDB コネクタークラスの名前。
- 3
- MongoDB レプリカセットへの接続に使用するホストアドレス。
- 4
- 生成されたイベントの namespace を形成する MongoDB レプリカセットの論理名。コネクターが書き込む Kafka トピックの名前、Kafka Connect スキーマ名、および Arvo コンバーターが使用される場合に対応する Avro スキーマの namespace のすべてに使用されます。
- 5
- 監視するすべてのコレクションのコレクション namespace (例: <dbName>.<collectionName>) と一致する正規表現の任意リスト。
Kafka Connect でコネクターインスタンスを作成します。たとえば、
KafkaConnector
リソースをinventory-connector.yaml
ファイルに保存した場合は、以下のコマンドを実行します。oc apply -f inventory-connector.yaml
上記のコマンドは
inventory-connector
を登録し、コネクターはKafkaConnector
CR に定義されているinventory
コレクションに対して実行を開始します。
Debezium MongoDB コネクターに設定できる設定プロパティーの完全リストは、MongoDB コネクター設定プロパティーを参照してください。
結果
コネクターが起動したら、以下のアクションを完了します。
- MongoDB レプリカセットでコレクションの スナップショット 一貫性をもたせて実行する。
- レプリカセットの oplogs を読み取る。
- 挿入、更新、削除されたすべてのドキュメントの変更イベントを生成する。
- Kafka トピックに変更イベントレコードをストリーミングする。
4.5.4. Debezium MongoDB コネクターが実行していることの確認
コネクターがエラーなしで正常に起動すると、コネクターがキャプチャーするように設定された各テーブルのトピックが作成されます。ダウンストリームアプリケーションは、これらのトピックをサブスクライブして、ソースデータベースで発生する情報イベントを取得できます。
コネクターが実行されていることを確認するには、OpenShift Container Platform Web コンソールまたは OpenShift CLI ツール (oc) から以下の操作を実行します。
- コネクターのステータスを確認します。
- コネクターがトピックを生成していることを確認します。
- 各テーブルの最初のスナップショットの実行中にコネクターが生成する読み取り操作 ("op":"r") のイベントがトピックに反映されていることを確認します。
前提条件
- Debezium コネクターは AMQ Streams on OpenShift にデプロイされている。
-
OpenShift
oc
CLI クライアントがインストールされている。 - OpenShift Container Platform Web コンソールへのアクセスがある。
手順
以下の方法のいずれかを使用して
KafkaConnector
リソースのステータスを確認します。OpenShift Container Platform Web コンソールから以下を実行します。
- Home → Search に移動します。
-
Search ページで Resources をクリックし、Select Resource ボックスを開き、
KafkaConnector
を入力します。 - KafkaConnectors リストから、チェックするコネクターの名前をクリックします (例: inventory-connector-mongodb)。
- Conditions セクションで、Type および Status 列の値が Ready および True に設定されていることを確認します。
ターミナルウィンドウから以下を実行します。
以下のコマンドを入力します。
oc describe KafkaConnector <connector-name> -n <project>
以下に例を示します。
oc describe KafkaConnector inventory-connector-mongodb -n debezium
このコマンドは、以下の出力のようなステータス情報を返します。
例4.3
KafkaConnector
リソースのステータスName: inventory-connector-mongodb Namespace: debezium Labels: strimzi.io/cluster=debezium-kafka-connect-cluster Annotations: <none> API Version: kafka.strimzi.io/v1beta2 Kind: KafkaConnector ... Status: Conditions: Last Transition Time: 2021-12-08T17:41:34.897153Z Status: True Type: Ready Connector Status: Connector: State: RUNNING worker_id: 10.131.1.124:8083 Name: inventory-connector-mongodb Tasks: Id: 0 State: RUNNING worker_id: 10.131.1.124:8083 Type: source Observed Generation: 1 Tasks Max: 1 Topics: inventory_connector_mongodb inventory_connector_mongodb.inventory.addresses inventory_connector_mongodb.inventory.customers inventory_connector_mongodb.inventory.geom inventory_connector_mongodb.inventory.orders inventory_connector_mongodb.inventory.products inventory_connector_mongodb.inventory.products_on_hand Events: <none>
コネクターによって Kafka トピックが作成されたことを確認します。
OpenShift Container Platform Web コンソールから以
- Home → Search に移動します。
-
Search ページで Resources をクリックし、Select Resource ボックスを開き、
KafkaTopic
を入力します。 - KafkaTopics リストから確認するトピックの名前をクリックします (例: inventory-connector-mongodb.inventory.orders---ac5e98ac6a5d91e04d8ec0dc9078a1ece439081d)。
- Conditions セクションで、Type および Status 列の値が Ready および True に設定されていることを確認します。
ターミナルウィンドウから以下を実行します。
以下のコマンドを入力します。
oc get kafkatopics
このコマンドは、以下の出力のようなステータス情報を返します。
例4.4
KafkaTopic
リソースのステータスNAME CLUSTER PARTITIONS REPLICATION FACTOR READY connect-cluster-configs debezium-kafka-cluster 1 1 True connect-cluster-offsets debezium-kafka-cluster 25 1 True connect-cluster-status debezium-kafka-cluster 5 1 True consumer-offsets---84e7a678d08f4bd226872e5cdd4eb527fadc1c6a debezium-kafka-cluster 50 1 True inventory-connector-mongodb---a96f69b23d6118ff415f772679da623fbbb99421 debezium-kafka-cluster 1 1 True inventory-connector-mongodb.inventory.addresses---1b6beaf7b2eb57d177d92be90ca2b210c9a56480 debezium-kafka-cluster 1 1 True inventory-connector-mongodb.inventory.customers---9931e04ec92ecc0924f4406af3fdace7545c483b debezium-kafka-cluster 1 1 True inventory-connector-mongodb.inventory.geom---9f7e136091f071bf49ca59bf99e86c713ee58dd5 debezium-kafka-cluster 1 1 True inventory-connector-mongodb.inventory.orders---ac5e98ac6a5d91e04d8ec0dc9078a1ece439081d debezium-kafka-cluster 1 1 True inventory-connector-mongodb.inventory.products---df0746db116844cee2297fab611c21b56f82dcef debezium-kafka-cluster 1 1 True inventory-connector-mongodb.inventory.products-on-hand---8649e0f17ffcc9212e266e31a7aeea4585e5c6b5 debezium-kafka-cluster 1 1 True schema-changes.inventory debezium-kafka-cluster 1 1 True strimzi-store-topic---effb8e3e057afce1ecf67c3f5d8e4e3ff177fc55 debezium-kafka-cluster 1 1 True strimzi-topic-operator-kstreams-topic-store-changelog---b75e702040b99be8a9263134de3507fc0cc4017b debezium-kafka-cluster 1 1 True
トピックの内容を確認します。
- 端末画面で、以下のコマンドを入力します。
oc exec -n <project> -it <kafka-cluster> -- /opt/kafka/bin/kafka-console-consumer.sh \ > --bootstrap-server localhost:9092 \ > --from-beginning \ > --property print.key=true \ > --topic=<topic-name>
以下に例を示します。
oc exec -n debezium -it debezium-kafka-cluster-kafka-0 -- /opt/kafka/bin/kafka-console-consumer.sh \ > --bootstrap-server localhost:9092 \ > --from-beginning \ > --property print.key=true \ > --topic=inventory_connector_mongodb.inventory.products_on_hand
トピック名を指定する形式は、手順 1 で返された
oc describe
コマンドと同じです (例:inventory_connector_mongodb.inventory.addresses
)。トピックの各イベントについて、このコマンドは、以下の出力のような情報を返します。
例4.5 Debezium 変更イベントの内容
{"schema":{"type":"struct","fields":[{"type":"int32","optional":false,"field":"product_id"}],"optional":false,"name":"inventory_connector_mongodb.inventory.products_on_hand.Key"},"payload":{"product_id":101}} {"schema":{"type":"struct","fields":[{"type":"struct","fields":[{"type":"int32","optional":false,"field":"product_id"},{"type":"int32","optional":false,"field":"quantity"}],"optional":true,"name":"inventory_connector_mongodb.inventory.products_on_hand.Value","field":"before"},{"type":"struct","fields":[{"type":"int32","optional":false,"field":"product_id"},{"type":"int32","optional":false,"field":"quantity"}],"optional":true,"name":"inventory_connector_mongodb.inventory.products_on_hand.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":"string","optional":true,"name":"io.debezium.data.Enum","version":1,"parameters":{"allowed":"true,last,false"},"default":"false","field":"snapshot"},{"type":"string","optional":false,"field":"db"},{"type":"string","optional":true,"field":"sequence"},{"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.mongodb.Source","field":"source"},{"type":"string","optional":false,"field":"op"},{"type":"int64","optional":true,"field":"ts_ms"},{"type":"struct","fields":[{"type":"string","optional":false,"field":"id"},{"type":"int64","optional":false,"field":"total_order"},{"type":"int64","optional":false,"field":"data_collection_order"}],"optional":true,"field":"transaction"}],"optional":false,"name":"inventory_connector_mongodb.inventory.products_on_hand.Envelope"},"payload":{"before":null,"after":{"product_id":101,"quantity":3},"source":{"version":"1.7.2.Final-redhat-00001","connector":"mongodb","name":"inventory_connector_mongodb","ts_ms":1638985247805,"snapshot":"true","db":"inventory","sequence":null,"table":"products_on_hand","server_id":0,"gtid":null,"file":"mongodb-bin.000003","pos":156,"row":0,"thread":null,"query":null},"op":"r","ts_ms":1638985247805,"transaction":null}}
上記の例では、
payload
値は、コネクタースナップショットがテーブルinventory.products_on_hand
から 読み込み (op" ="r"
) イベントを生成したことを示しています。product_id
レコードのbefore
状態はnull
であり、レコードに以前の値が存在しないことを示します。"after"
状態がproduct_id
101
で項目のquantity
を3
で示しています。
4.5.5. Debezium Db2 コネクター設定プロパティーの説明
Debezium MongoDB コネクターには、アプリケーションに適したコネクター動作を実現するために使用できる設定プロパティーが多数あります。多くのプロパティーにはデフォルト値があります。プロパティーに関する情報は、以下のように設定されています。
以下の設定プロパティーは、デフォルト値がない場合は必須です。
プロパティー | デフォルト | 説明 |
---|---|---|
コネクターの一意名。同じ名前で再登録を試みると失敗します。(このプロパティーはすべての Kafka Connect コネクターに必要です) | ||
コネクターの Java クラスの名前。MongoDB コネクターには、常に | ||
レプリカセットでの MongoDB サーバーのホスト名とポートのペア ('host' または 'host:port' 形式) のコンマ区切りリスト。リストには、ホスト名とポートのペアを 1 つ含めることができます。 | ||
このコネクターが監視するコネクターや MongoDB レプリカセット、またはシャードクラスターを識別する一意の名前。このサーバー名は、MongoDB レプリカセットまたはクラスターから生成される永続化されたすべての Kafka トピックの接頭辞になるため、各サーバーは最大 1 つの Debezium コネクターによって監視される必要があります。使用できる文字は、英数字、ハイフン、ドット、アンダースコアのみです。 | ||
MongoDB への接続時に使用されるデータベースユーザーの名前。これは MongoDB が認証を使用するように設定されている場合にのみ必要です。 | ||
MongoDB への接続時に使用されるパスワード。これは MongoDB が認証を使用するように設定されている場合にのみ必要です。 | ||
|
MongoDB クレデンシャルが含まれるデータベース (認証ソース)。これは、MongoDB が | |
| コネクターは SSL を使用して MongoDB インスタンスに接続します。 | |
|
SSL が有効な場合、接続フェーズ中に厳密なホスト名のチェックを無効にするかどうかを制御する設定です。 | |
空の文字列 |
監視するデータベース名と一致する正規表現のコンマ区切りリスト (任意)。 | |
空の文字列 |
監視から除外されるデータベース名と一致する正規表現のコンマ区切りリスト (任意)。 | |
空の文字列 |
監視する MongoDB コレクションの完全修飾 namespace と一致する正規表現のコンマ区切りリスト (任意)。 | |
空の文字列 |
監視から除外される MongoDB コレクションの完全修飾 namespace と一致する正規表現のコンマ区切りリスト (任意)。 | |
| コネクターの起動時にスナップショットを実行する基準を指定します。デフォルトは initial で、オフセットが見つからない場合や oplog に以前のオフセットが含まれなくなった場合にコネクターがスナップショットを読み取るように指定します。never オプションは、コネクターはスナップショットを使用せずに、ログをの追跡を続行すべきであることを指定します。 | |
|
スナップショットを作成する | |
空の文字列 | 変更イベントメッセージ値から除外される必要があるフィールドの完全修飾名のコンマ区切りリスト (任意)。フィールドの完全修飾名の形式はdatabaseName.collectionName.fieldName.nestedFieldName で、databaseName および collectionName にはすべての文字と一致するワイルドカード (*) が含まれることがあります。 | |
空の文字列 | イベントメッセージ値のフィールドの名前を変更するために使用されるフィールドの完全修飾置換のコンマ区切りリスト (任意)。フィールドの完全修飾置換の形式は databaseName.collectionName.fieldName.nestedFieldName:newNestedFieldName で、databaseName および collectionName にはすべての文字と一致するワイルドカード (*) が含まれることがあります。コロン (:) は、フィールドの名前変更マッピングを決定するために使用されます。次のフィールドの置換は、リストの前のフィールド置換の結果に適用されるため、同じパスにある複数のフィールドの名前を変更する場合は、この点に注意してください。 | |
| このコネクターのために作成する必要のあるタスクの最大数。MongoDB コネクターは各レプリカセットに個別のタスクの使用しようとします。そのため、コネクターを単一の MongoDB レプリカセットと使用する場合は、デフォルトを使用できます。MongoDB のシャードクラスターでコネクターを使用する場合、クラスターのシャード数以上の値を指定して、各レプリカセットの作業が Kafka Connect によって分散されるようにすることが推奨されます。 | |
| レプリカセットでコレクションの最初の同期を実行するために使用されるスレッドの最大数を指定する正の整数値。デフォルトは 1 です。 | |
|
廃棄 (tombstone) イベントの後に削除イベントが続くかどうかを制御します。 | |
コネクターの起動後、スナップショットを取得するまで待機する間隔 (ミリ秒単位)。 | ||
|
スナップショットの実行中に各コレクションから 1 度に読み取る必要があるドキュメントの最大数を指定します。コネクターは、このサイズの複数のバッチでコレクションの内容を読み取ります。 |
以下の 高度な 設定プロパティーには、ほとんどの状況で機能する適切なデフォルト設定があるため、コネクターの設定で指定する必要はほとんどありません。
プロパティー | デフォルト | 説明 |
---|---|---|
|
データベースログから読み取られた変更イベントが Kafka に書き込まれる前に配置される、ブロッキングキューの最大サイズを指定する正の整数値。このキューは、Kafka への書き込みが遅い場合や Kafka が利用できない場合などに、oplog リーダーにバックプレシャーを提供できます。キューに発生するイベントは、このコネクターによって定期的に記録されるオフセットには含まれません。デフォルトは 8192 で、常に | |
| このコネクターの反復処理中に処理される必要があるイベントの各バッチの最大サイズを指定する正の整数値。デフォルトは 2048 です。 | |
| ブロッキングキューの最大サイズ (バイト単位) の long 値。この機能はデフォルトで無効になっています。正の long 値が設定されると有効になります。 | |
| 各反復処理の実行中に新しい変更イベントが表示されるまでコネクターが待機する時間 (ミリ秒単位) を指定する正の整数値。デフォルトは 1000 ミリ秒 (1 秒) です。 | |
| 最初に失敗した接続試行の後またはプライマリーが利用できない場合に、プライマリーへの再接続を試行するときの最初の遅延を指定する正の整数値。デフォルトは 1 秒 (1000 ミリ秒) です。 | |
| 接続試行に繰り返し失敗した後またはプライマリーが利用できない場合に、プライマリーへの再接続を試行するときの最大遅延を指定する正の整数値。デフォルトは 120 秒 (120,000 ミリ秒) です。 | |
|
レプリカセットのプライマリーへの接続を試行する場合の最大失敗回数を指定する正の整数値。この値を越えると、例外が発生し、タスクが中止されます。デフォルトは 16。 | |
|
'mongodb.hosts' 内のアドレスがクラスターまたはレプリカセットの全メンバーを検出するために使用されるシードであるかどうか ( | |
|
ハートビートメッセージが送信される頻度を制御します。
このプロパティーを | |
|
ハートビートメッセージが送信されるトピックの命名を制御します。 | |
コネクター設定が、Avro を使用するように | Avro の命名要件に準拠するためにフィールド名がサニタイズされるかどうか。 | |
ストリーミング中にスキップされる操作タイプのコンマ区切りリスト。操作には、 | ||
スナップショットに含まれるコレクション項目を制御します。このプロパティーはスナップショットにのみ影響します。databaseName.collectionName の形式でコレクション名のコンマ区切りリストを指定します。
指定する各コレクションに対して、別の設定プロパティー ( | ||
|
詳細は、トランザクションメタデータ を参照してください。 | |
10000 (10 秒) | 再試行可能なエラーが発生した後にコネクターを再起動するまで待機する時間 (ミリ秒単位)。 | |
| コネクターが新規、削除、または変更したレプリカセットをポーリングする間隔。 | |
10000 (10 秒) | 新しい接続試行が中断されるまでドライバーが待機する時間 (ミリ秒単位)。 | |
0 |
ソケットでの送受信がタイムアウトするまでにかかる時間 (ミリ秒単位)。 | |
30000 (30 秒) | ドライバーがタイムアウトし、エラーが出力される前に、サーバーが選択されるまでドライバーが待つ時間 (ミリ秒単位)。 | |
|
実行タイムアウトの例外を発生させる前に、oplog カーソルが結果を生成するのを待つ最大期間 (ミリ秒単位) を指定します。値 |
4.6. Debezium MongoDB コネクターのパフォーマンスの監視
Debezium MongoDB コネクターには、Zookeeper、Kafka、および Kafka Connect にある JMX メトリクスの組み込みサポートに加えて、2 種類のメトリクスがあります。
- スナップショットメトリクス は、スナップショットの実行中にコネクター操作に関する情報を提供します。
- メトリクスのストリーミング は、コネクターが変更をキャプチャーし、変更イベントレコードをストリーミングするときにコネクター操作に関する情報を提供します。
Debezium モニターリングのドキュメント では、JMX を使用してこれらのメトリクスを公開する方法の詳細を提供します。
4.6.1. MongoDB スナップショット作成中の Debezium の監視
MBean は debezium.mongodb:type=connector-metrics,context=snapshot,server=<mongodb.server.name>
です。
スナップショット操作がアクティブでない場合や、最後のコネクターの起動後にスナップショットの作成が発生した場合に、スナップショットメトリクスは公開されません。
以下の表は、利用可能なスナップショットのメトリックの一覧です。
属性 | 型 | 説明 |
---|---|---|
| コネクターが読み取りした最後のスナップショットイベント。 | |
| コネクターが最新のイベントを読み取りおよび処理してからの経過時間 (ミリ秒単位)。 | |
| 前回の開始またはリセット以降にコネクターで確認されたイベントの合計数。 | |
| コネクターに設定された include/exclude リストのフィルターリングルールによってフィルターされたイベントの数。 | |
|
| コネクターによって監視されるテーブルの一覧。 |
| コネクターによって取得されるテーブルの一覧。 | |
| snapshotter とメインの Kafka Connect ループの間でイベントを渡すために使用されるキューの長さ。 | |
| snapshotter とメインの Kafka Connect ループの間でイベントを渡すために使用されるキューの空き容量。 | |
| スナップショットに含まれているテーブルの合計数。 | |
| スナップショットによってまだコピーされていないテーブルの数。 | |
| スナップショットが起動されたかどうか。 | |
| スナップショットが中断されたかどうか。 | |
| スナップショットが完了したかどうか。 | |
| スナップショットが完了したかどうかに関わらず、これまでスナップショットにかかった時間 (秒単位)。 | |
| スナップショットの各テーブルに対してスキャンされる行数が含まれるマップ。テーブルは、処理中に増分がマップに追加されます。スキャンされた 10,000 行ごとに、テーブルの完成時に更新されます。 | |
|
キューの最大バッファー (バイト単位)。 | |
| キュー内のレコードの現在のデータ (バイト単位)。 |
Debezium MongoDB コネクターは、以下のカスタムスナップショットメトリクスも提供します。
属性 | 型 | 説明 |
---|---|---|
|
| データベースの切断数。 |
4.6.2. Debezium MongoDB コネクターレコードストリーミングの監視
MBean は debezium.mongodb:type=connector-metrics,context=streaming,server=<mongodb.server.name>
。
以下の表は、利用可能なストリーミングメトリクスの一覧です。
属性 | 型 | 説明 |
---|---|---|
| コネクターが読み取られた最後のストリーミングイベント。 | |
| コネクターが最新のイベントを読み取りおよび処理してからの経過時間 (ミリ秒単位)。 | |
| 前回の開始またはリセット以降にコネクターで確認されたイベントの合計数。 | |
| コネクターに設定された include/exclude リストのフィルターリングルールによってフィルターされたイベントの数。 | |
|
| コネクターによって監視されるテーブルの一覧。 |
| コネクターによって取得されるテーブルの一覧。 | |
| ストリーマーとメイン Kafka Connect ループの間でイベントを渡すために使用されるキューの長さ。 | |
| ストリーマーとメインの Kafka Connect ループの間でイベントを渡すために使用されるキューの空き容量。 | |
| コネクターが現在データベースサーバーに接続されているかどうかを示すフラグ。 | |
| 最後の変更イベントのタイムスタンプとそれを処理するコネクターとの間の期間 (ミリ秒単位)。この値は、データベースサーバーとコネクターが稼働しているマシンのクロック間の差異に対応します。 | |
| コミットされた処理済みトランザクションの数。 | |
| 最後に受信したイベントの位置。 | |
| 最後に処理されたトランザクションのトランザクション識別子。 | |
| キューの最大バッファー (バイト単位)。 | |
| キュー内のレコードの現在のデータ (バイト単位)。 |
Debezium MongoDB コネクターは、以下のカスタムストリーミングメトリクスも提供します。
属性 | 型 | 説明 |
---|---|---|
|
| データベースの切断数。 |
|
| プライマリーノードの選出数。 |
4.7. Debezium MongoDB コネクターによる障害および問題の処理方法
Debezium は、複数のアップストリームデータベースのすべての変更をキャプチャーする分散システムであり、イベントの見逃しや損失は発生しません。システムが正常に操作している場合や、慎重に管理されている場合は、Debezium は変更イベントごとに 1 度だけ 配信します。
障害が発生しても、システムからイベントがなくなることはありません。ただし、障害から復旧している間は、変更イベントが繰り返えされる可能性があります。このような状態では、Debezium は Kafka と同様に、変更イベントを 少なくとも 1 回 配信します。
以下のトピックでは、Debezium MongoDB コネクターがさまざまな種類の障害および問題を処理する方法を詳説します。
設定および起動エラー
以下の状況では、起動時にコネクターが失敗し、エラーまたは例外がログに記録され、実行が停止されます。
- コネクターの設定が無効である。
- 指定の接続パラメーターを使用してコネクターを MongoDB に接続できない。
失敗したら、コネクターは指数バックオフを使用して再接続を試みます。再接続試行の最大数を設定できます。
このような場合、エラーには問題の詳細が含まれ、場合によっては回避策が提示されます。設定が修正されたり、MongoDB の問題が解決された場合はコネクターを再起動できます。
再接の続試行は、3 つのプロパティーで制御されます。
-
connect.backoff.initial.delay.ms
- 初回の再接続を試みるまでの遅延。デフォルトは 1 秒 (1000 ミリ秒) です。 -
connect.backoff.max.delay.ms
- 再接続を試行するまでの最大遅延。デフォルトは 120 秒 (120,000 ミリ秒) です。 -
connect.max.attempts
- エラーが生成されるまでの最大試行回数。デフォルトは 16 です。
各遅延は、最大遅延以下で、前の遅延の 2 倍です。以下の表は、デフォルト値を指定した場合の、失敗した各接続試行の遅延と、失敗前の合計累積時間を表しています。
再接続試行回数 | 試行までの遅延 (秒単位) | 試行までの遅延合計 (分および秒単位) |
---|---|---|
1 | 1 | 00:01 |
2 | 2 | 00:03 |
3 | 4 | 00:07 |
4 | 8 | 00:15 |
5 | 16 | 00:31 |
6 | 32 | 01:03 |
7 | 64 | 02:07 |
8 | 120 | 04:07 |
9 | 120 | 06:07 |
10 | 120 | 08:07 |
11 | 120 | 10:07 |
12 | 120 | 12:07 |
13 | 120 | 14:07 |
14 | 120 | 16:07 |
15 | 120 | 18:07 |
16 | 120 | 20:07 |
Kafka Connect のプロセスは正常に停止する
Kafka Connect が分散モードで実行され、Kafka Connect プロセスが正常に停止された場合は、Kafka Connect はプロセスのシャットダウン前に、すべてのプロセスのコネクタータスクをそのグループの別の Kafka Connect プロセスに移行し、新しいコネクタータスクは、以前のタスクが停止した場所で開始されます。コネクタータスクが正常に停止され、新しいプロセスで再起動されるまでの間、プロセスに短い遅延が発生します。
グループにプロセスが 1 つだけあり、そのプロセスが正常に停止された場合、Kafka Connect はコネクターを停止し、各レプリカセットの最後のオフセットを記録します。再起動時に、レプリカセットタスクは停止した場所で続行されます。
Kafka Connect プロセスのクラッシュ
Kafka Connector プロセスが予期せず停止した場合、最後に処理されたオフセットを記録せずに、実行中のコネクタータスクが終了します。Kafka Connect が分散モードで実行されている場合は、他のプロセスでこれらのコネクタータスクを再起動します。ただし、MongoDB コネクターは以前のプロセスによって 記録 された最後のオフセットから再開します。つまり、新しい代替タスクによって、クラッシュの直前に処理された同じ変更イベントが生成される可能性があります。重複するイベントの数は、オフセットのフラッシュ期間とクラッシュの直前のデータ変更の量によって異なります。
障害からの復旧中に一部のイベントが重複された可能性があるため、コンシューマーは常に一部のイベントが重複している可能性があることを想定する必要があります。Debezium の変更はべき等であるため、一連のイベントは常に同じ状態になります。
Debezium の各変更イベントメッセージには、イベントの生成元に関するソース固有の情報が含まれます。これには、MongoDB イベントの一意なトランザクション識別子 (h
) やタイムスタンプ (sec
and ord
) が含まれます。コンシューマーはこれらの値の他の部分を追跡し、特定のイベントがすでに発生しているかどうかを確認することができます。
コネクターの長期間の停止
コネクターが正常に停止された場合、レプリカセットは引き続き使用でき、新しい変更は MongoDB の oplog に記録されます。コネクターが再起動されると、最後に停止した場所で各レプリカセットの変更のストリーミングを再開し、コネクターが停止した間に加えられたすべての変更の記録イベントを記録します。コネクターが一定期間停止し、コネクターが読み取っていない一部の操作を MongoDB が oplog からパージするようになると、コネクターは起動時にスナップショットを実行します。
Kafka クラスターを適切に設定すると、大量のスループットを実現できます。Kafka Connect は Kafka のベストプラクティスを使用して記述され、十分なリソースがあれば非常に多くのデータベース変更イベントを処理できます。そのため、コネクターがしばらくして再起動されると、データベースに追いつく可能性が非常に高くなりますが、遅れを取り戻すまでに掛かる時間は、Kafka の機能やパフォーマンスおよび MongoDB のデータへの変更の量に応じて異なります。
コネクターが長時間停止した場合、MongoDB が古い oplog ファイルをパージし、コネクターの最後の位置が失われる可能性があります。この場合、最初 のスナップショットモード (デフォルト) で設定されたコネクターが最終的に再起動されると、MongoDB サーバーには開始点がなくなり、コネクターはエラーによって失敗します。
MongoDB による書き込みの損失
失敗した場合の特定の状況では、MongoDB はコミットを失う可能性があり、その場合には MongoDB コネクターでは、失われた変更をキャプチャーできません。たとえば、プライマリーが変更を適用して、その oplog にその変更を記録した後に、突然クラッシュした場合には、セカンダリーノードがコンテンツを読み取るまでに oplog が利用できなくなる可能性があります。その結果、新しいプライマリーノードとして選択されるセカンダリーノードには、oplog からの最新の変更が含まれていない可能性があります。
現時点では、MongoDB のこの副次的な影響を防ぐ方法はありません。
第5章 MySQL の Debezium コネクター
本リリースの Debezium MySQL コネクターには、他の Debezium コネクターによって使用される一般的なコネクターフレームワークを基にした新しいデフォルトのキャプチャー実装が含まれています。改訂されたキャプチャーの実装はテクノロジープレビューの機能です。テクノロジープレビュー機能は、Red Hat の実稼働環境のサービスレベルアグリーメント (SLA) ではサポートされません。また、機能的に完全ではない可能性があるため、Red Hat はテクノロジープレビュー機能を実稼働環境に実装することは推奨しません。テクノロジープレビュー機能は、最新の技術をいち早く提供し、開発段階で機能のテストやフィードバックの収集を可能にするために提供されます。サポート範囲の詳細は、テクノロジープレビュー機能のサポート範囲 を参照してください。
新しいキャプチャー実装で実行中にコネクターによってエラーや予期しない動作が生成される場合は、以下の設定オプションを指定することで以前の実装に戻すことができます。
internal.implementation=legacy
MySQL には、データベースにコミットされた順序ですべての操作を記録するバイナリーログ (binlog) があります。これには、テーブルスキーマの変更やテーブルのデータの変更が含まれます。MySQL はレプリケーションとリカバリーに binlog を使用します。
Debezium MySQL コネクターは binlog を読み取り、行レベルの INSERT
、UPDATE
、および DELETE
操作の変更イベントを生成し、変更イベントを Kafka トピックに出力します。クライアントアプリケーションはこれらの Kafka トピックを読み取ります。
MySQL は通常、指定期間後に binlogs をパージするように設定されているため、MySQL コネクターは各データベースの最初の整合性スナップショット を実行します。MySQL コネクターは、スナップショットが作成された時点から binlog を読み取ります。
このコネクターと互換性のある MySQL データベースのバージョンについては、Debezium でサポートされる設定ページを参照してください。
Debezium MySQL コネクターの使用に関する情報および手順は、以下のように整理されています。
5.1. Debezium MySQL コネクターの仕組み
コネクターがサポートする MySQL トポロジーの概要は、アプリケーションを計画するときに役立ちます。Debezium MySQL コネクターを最適に設定および実行するには、コネクターによるテーブルの構造の追跡方法、スキーマ変更の公開方法、スナップショットの実行方法、および Kafka トピック名の決定方法を理解しておくと便利です。
詳細は以下を参照してください。
5.1.1. Debezium コネクターでサポートされる MySQL トポロジー
Debezium MySQL コネクターは以下の MySQL トポロジーをサポートします。
- スタンドアロン
- 単一の MySQL サーバーを使用する場合は、Debezium MySQL コネクターがサーバーを監視できるように、binlog を有効 (および任意で GTID を有効) にする必要があります。バイナリーログも増分 バックアップ として使用できるため、これは多くの場合で許容されます。この場合、MySQL コネクターは常にこのスタンドアロン MySQL サーバーインスタンスに接続し、それに従います。
- プライマリーおよびレプリカ
Debezium MySQL コネクターはプライマリーサーバーまたはレプリカの 1 つ (レプリカの binlog が有効になっている場合) に従うことができますが、コネクターはサーバーが認識できるクラスターのみで変更を確認できます。通常、これはマルチプライマリートポロジー以外では問題ではありません。
コネクターは、サーバーの binlog の位置を記録します。この位置は、クラスターの各サーバーごとに異なります。そのため、コネクターは 1 つの MySQL サーバーインスタンスのみに従う必要があります。このサーバーに障害が発生した場合、サーバーを再起動またはリカバリーしないと、コネクターは継続できません。
- 高可用性クラスター
- MySQL にはさまざまな 高可用性 ソリューションが存在し、問題や障害の耐性をつけ、即座に回復することが大変容易になります。ほとんどの HA MySQL クラスターは GTID を使用します。そのため、レプリカはあらゆるプライマリーサーバーの変更をすべて追跡できます。
- マルチプライマリー
ネットワークデータベース (NDB) クラスターのレプリケーション は、複数のプライマリーアーバーからそれぞれをレプリケートする 1 つ以上の MySQL レプリカを使用します。これは、複数の MySQL クラスターのレプリケーションを集約する強力な方法です。このトポロジーには GTID を使用する必要があります。
Debezium MySQL コネクターはこれらのマルチプライマリー MySQL レプリカをソースとして使用することができ、新しいレプリカが古いレプリカに追い付けば、異なるマルチプライマリー MySQL レプリカにフェイルオーバーできます。つまり、新しいレプリカには最初のレプリカで確認されたすべてのトランザクションが含まれます。これは、新しいマルチプライマリー MySQL レプリカへの再接続を試み、binlog で適切な場所を見つけようとする際に、特定の GTID ソースが含まれるまたは除外されるようにコネクターを設定できるため、コネクターがデータベースやテーブルのサブセットのみを使用している場合でも機能します。
- ホステッド
Debezium MySQL コネクターが Amazon RDS や Amazon Aurora などのホステッドオプションを使用するためのサポートがあります。
これらのホステッドオプションではグローバル読み取りロックが許可されないため、テーブルレベルロックを使用して 整合性スナップショット を作成します。
5.1.2. Debezium MySQL コネクターによるデータベーススキーマの変更の処理方法
データベースクライアントがデータベースのクエリーを行うと、クライアントはデータベースの現在のスキーマを使用します。しかし、データベーススキーマはいつでも変更が可能です。そのため、挿入、更新、または削除の操作が記録されるたびに、コネクターはどのスキーマであるかを特定できる必要があります。また、テーブルのスキーマが変更される前に記録された、比較的古いイベントをコネクターが処理するので、コネクターは現在のスキーマだけを使用することはできません。
スキーマの変更後に発生する変更を正しく処理するために、MySQL にはデータの行レベルの変更だけでなく、データベースに適用される DDL ステートメントも含まれます。コネクターは binlog を読み取り、DDL ステートメントを見つけると、それらの DDL ステートメントを解析し、各テーブルのスキーマのインメモリー表現を更新します。コネクターはこのスキーマ表現を使用して、挿入、更新、または削除の操作時にテーブルの構造を特定し、適切な変更イベントを生成します。別のデータベース履歴 Kafka トピックでは、コネクターは各 DDL ステートメントがある binlog の場所とともにすべての DDL ステートメントを記録します。
コネクターが正常にクラッシュまたは停止された後にコネクターが再起動されると、コネクターは特定の場所 (特定の時点) から binlog の読み取りを開始します。コネクターは、データベース履歴の Kafka トピックを読み取り、コネクターが起動する binlog の時点まですべての DDL ステートメントを解析することで、この時点で存在したテーブル構造を再ビルドします。
このデータベース履歴トピックはコネクターのみが使用します。コネクターは任意で、コンシューマーアプリケーション向けの異なるトピックへのスキーマ変更イベントの生成を表示でき ます。
MySQL コネクターが、gh-ost
または pt-online-schema-change
などのスキーマ変更ツールが適用されるテーブルで変更をキャプチャーすると、移行プロセス中にヘルパーテーブルが作成されます。これらのヘルパーテーブルへの変更をキャプチャーするようにコネクターを設定する必要があります。ヘルパーテーブル用に生成されたレコードがコンシューマーに必要ない場合は、メッセージ変換を 1 回適用して、除去できます。
Debezium イベントレコードを受信する トピックのデフォルト名 を参照してください。
5.1.3. Debezium MySQL コネクターによるデータベーススキーマの変更の公開方法
Debezium MySQL コネクターを設定すると、データベースのキャプチャーされたテーブルに適用されるスキーマの変更を記述するスキーマ変更イベントを生成できます。コネクターは、スキーマ変更イベントをすべて <serverName>
という名前の Kafka トピックに書き込みます。serverName
は database.server.name
設定プロパティーに指定されたコネクターの名前になります。コネクターがスキーマ変更トピックに送信するメッセージには、ペイロードと、任意で変更イベントメッセージのスキーマが含まれます。
スキーマ変更イベントメッセージのペイロードには、以下の要素が含まれます。
ddl
-
スキーマの変更につながる SQL
CREATE
、ALTER
、またはDROP
ステートメントを提供します。 databaseName
-
DDL ステートメントが適用されるデータベースの名前。
databaseName
の値は、メッセージキーとして機能します。 pos
- ステートメントが表示される binlog の位置。
tableChanges
-
スキーマの変更後のテーブルスキーマ全体の構造化表現。
tableChanges
フィールドには、テーブルの各列のエントリーなどのアレイが含まれます。構造化された表現は JSON または Avro 形式でデータを表示するため、コンシューマーは DDL パーサーを介して最初にメッセージを処理しなくてもメッセージを簡単に読み取りできます。
キャプチャーモードであるテーブルでは、コネクターはスキーマ変更トピックにスキーマ変更の履歴だけでなく、内部データベース履歴トピックにも格納します。内部データベース履歴トピックはコネクターのみの使用を対象としており、使用するアプリケーションによる直接使用を目的としていません。スキーマ変更に関する通知が必要なアプリケーションが、スキーマ変更トピックからの情報のみを使用するようにしてください。
データベース履歴トピックをパーティションに分割しないでください。データベース履歴トピックが正しく機能するには、コネクターが出力するイベントレコードの一貫したグローバル順序を維持する必要があります。
トピックがパーティション間で分割されないようにするには、以下のいずれかの方法を使用してトピックのパーティション数を設定します。
-
データベース履歴トピックを手動で作成する場合は、パーティション数を
1
に指定します。 -
Apache Kafka ブローカーを使用してデータベース履歴トピックを自動的に作成する場合に、トピックが作成されるので、Kafka
num.partitions
設定オプションの値を1
に設定します。
コネクターがスキーマ変更トピックに出力するメッセージの形式は、初期の状態であり、通知なしに変更される可能性があります。
以下に例を示します。MySQL コネクタースキーマ変更トピックに出力されるメッセージ
以下の例は、JSON 形式の一般的なスキーマ変更メッセージを示しています。メッセージには、テーブルスキーマの論理表現が含まれます。
{ "schema": { ... }, "payload": { "source": { // (1) "version": "1.7.2.Final", "connector": "mysql", "name": "dbserver1", "ts_ms": 0, "snapshot": "false", "db": "inventory", "sequence": null, "table": "customers", "server_id": 0, "gtid": null, "file": "mysql-bin.000003", "pos": 219, "row": 0, "thread": null, "query": null }, "databaseName": "inventory", // (2) "schemaName": null, "ddl": "ALTER TABLE customers ADD COLUMN middle_name VARCHAR(2000)", // (3) "tableChanges": [ // (4) { "type": "ALTER", // (5) "id": "\"inventory\".\"customers\"", // (6) "table": { // (7) "defaultCharsetName": "latin1", "primaryKeyColumnNames": [ // (8) "id" ], "columns": [ // (9) { "name": "id", "jdbcType": 4, "nativeType": null, "typeName": "INT", "typeExpression": "INT", "charsetName": null, "length": 11, "scale": null, "position": 1, "optional": false, "autoIncremented": true, "generated": true }, { "name": "first_name", "jdbcType": 12, "nativeType": null, "typeName": "VARCHAR", "typeExpression": "VARCHAR", "charsetName": "latin1", "length": 255, "scale": null, "position": 2, "optional": false, "autoIncremented": false, "generated": false }, { "name": "last_name", "jdbcType": 12, "nativeType": null, "typeName": "VARCHAR", "typeExpression": "VARCHAR", "charsetName": "latin1", "length": 255, "scale": null, "position": 3, "optional": false, "autoIncremented": false, "generated": false }, { "name": "email", "jdbcType": 12, "nativeType": null, "typeName": "VARCHAR", "typeExpression": "VARCHAR", "charsetName": "latin1", "length": 255, "scale": null, "position": 4, "optional": false, "autoIncremented": false, "generated": false }, { "name": "middle_name", "jdbcType": 12, "nativeType": null, "typeName": "VARCHAR", "typeExpression": "VARCHAR", "charsetName": "latin1", "length": 2000, "scale": null, "position": 5, "optional": true, "autoIncremented": false, "generated": false } ] } } ] }, "payload": { "databaseName": "inventory", "ddl": "CREATE TABLE products ( id INTEGER NOT NULL AUTO_INCREMENT PRIMARY KEY, name VARCHAR(255) NOT NULL, description VARCHAR(512), weight FLOAT ); ALTER TABLE products AUTO_INCREMENT = 101;", "source" : { "version": "1.7.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 } } }
項目 | フィールド名 | 説明 |
---|---|---|
1 |
|
|
2 |
|
変更が含まれるデータベースとスキーマを識別します。 |
3 |
|
このフィールドには、スキーマの変更を行う DDL が含まれます。 |
4 |
| DDL コマンドによって生成されるスキーマの変更が含まれる 1 つ以上の項目の配列。 |
5 |
| 変更の種類を説明します。値は以下のいずれかになります。
|
6 |
|
作成、変更、または破棄されたテーブルの完全な識別子。テーブルの名前が変更されると、この識別子は |
7 |
| 適用された変更後のテーブルメタデータを表します。 |
8 |
| テーブルのプライマリーキーを設定する列のリスト。 |
9 |
| 変更されたテーブルの各列のメタデータ。 |
スキーマ履歴トピック も参照してください。
5.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 コネクターのトラブルシューティングに関する詳細は、behavior when things go wrong を参照してください。
- グローバル読み取りロックが許可されない
一部の環境では、グローバル読み取りロックが許可されません。Debezium MySQL コネクターがグローバル読み取りロックが許可されないことを検出すると、代わりにテーブルレベルロックを使用して、この方法でスナップショットを実行します。これには、Debezium コネクターのデータベースユーザーに
LOCK TABLES
権限が必要になります。表5.3 テーブルレベルロックを使用して最初のスナップショットを実行するためのワークフロー ステップ アクション 1
テーブルレベルロックを取得します。
2
繰り返し可能な読み取りセマンティクス でトランザクションを開始し、トランザクション内の後続の読み取りがすべて 整合性スナップショット に対して実行されるようにします。
3
データベースとテーブルの名前を読み取り、選別します。
4
現在の binlog の位置を読み取ります。
5
コネクターが変更をキャプチャーするように設定されたデータベースとテーブルのスキーマを読み取ります。
6
該当する場合は、DDL の変更をスキーマ変更トピックに書き込みます。これには、必要なすべての
DROP…
およびCREATE…
DDL ステートメント。7
データベーステーブルをスキャンします。コネクターは、行ごとに、
CREATE
イベントを関係するテーブル固有の Kafka トピックに出力します。8
トランザクションをコミットします。
9
テーブルレベルロックを解除します。
10
コネクターオフセットの完了済みスナップショットを記録します。
5.1.4.1. アドホックスナップショット
アドホックスナップショットの使用はテクノロジープレビュー機能です。テクノロジープレビュー機能は、Red Hat の実稼働環境のサービスレベルアグリーメント (SLA) ではサポートされません。また、機能的に完全ではない可能性があるため、Red Hat はテクノロジープレビュー機能を実稼働環境に実装することは推奨しません。テクノロジープレビュー機能は、最新の技術をいち早く提供し、開発段階で機能のテストやフィードバックの収集を可能にするために提供されます。サポート範囲の詳細は、テクノロジープレビュー機能のサポート範囲 を参照してください。
デフォルトでは、コネクターは初回スナップショット操作の開始後にのみ実行されます。通常の状況では、この最初のスナップショットが作成されると、コネクターではスナップショットプロセスは繰り返し処理されません。コネクターがキャプチャーする今後の変更イベントデータはストリーミングプロセス経由でのみ行われます。
ただし、場合によっては、最初のスナップショット中にコネクターを取得したデータが古くなったり、失われたり、または不完全となったり可能性があります。テーブルデータを再キャプチャーするメカニズムを提供するため、Debezium にはアドホックスナップショットを実行するオプションがあります。データベースで以下が変更されたことで、アドホックスナップショットが実行される場合があります。
- コネクター設定は、異なるテーブルセットをキャプチャーするように変更されます。
- Kafka トピックを削除して、再構築する必要があります。
- 設定エラーや他の問題が原因で、データの破損が発生します。
アドホックと呼ばれるスナップショット を開始することで、以前にスナップショットをキャプチャーしたテーブルのスナップショットを再実行できます。アドホックスナップショットには、シグナルテーブル を使用する必要があります。シグナルリクエストを Debezium シグナルテーブルに送信して、アドホックスナップショットを開始します。
既存のテーブルのアドホックスナップショットを開始すると、コネクターはテーブルにすでに存在するトピックにコンテンツを追加します。既存のトピックが削除された場合には、トピックの自動作成 が有効になっているのであれば、Debezium は自動的にトピックを作成できます。
アドホックのスナップショットシグナルは、スナップショットに追加するテーブルを指定します。スナップショットは、データベースの内容全体をキャプチャーしたり、データベース内のテーブルのサブセットのみをキャプチャーしたりできます。
execute-snapshot
メッセージをシグナルテーブルに送信してキャプチャーするテーブルを指定します。以下の表で説明されているように、run-snapshot
シグナルのタイプを incremental
に設定し、スナップショットに追加するテーブルの名前を指定します。
フィールド | デフォルト | 値 |
---|---|---|
|
|
実行するスナップショットのタイプを指定します。 |
| 該当なし |
スナップショットを作成するテーブルの完全修飾名が含まれる配列。 |
アドホックスナップショットのトリガー
execute-snapshot
シグナルタイプのエントリーをシグナルテーブルに追加して、アドホックスナップショットを開始します。コネクターがメッセージを処理した後に、スナップショット操作を開始します。スナップショットプロセスは、最初と最後のプライマリーキーの値を読み取り、これらの値を各テーブルの開始ポイントおよびエンドポイントとして使用します。テーブルのエントリー数と設定されたチャンクサイズに基づいて、Debezium はテーブルをチャンクに分割し、チャンクごとに 1 度に 1 つずつスナップショットを順番に作成していきます。
現在、execute-snapshot
アクションタイプは 増分スナップショット のみをトリガーします。詳細は、スナップショットの増分を参照してください。
5.1.4.2. 増分スナップショット
増分スナップショットの使用はテクノロジープレビュー機能です。テクノロジープレビュー機能は、Red Hat の実稼働環境のサービスレベルアグリーメント (SLA) ではサポートされません。また、機能的に完全ではない可能性があるため、Red Hat はテクノロジープレビュー機能を実稼働環境に実装することは推奨しません。テクノロジープレビュー機能は、最新の技術をいち早く提供し、開発段階で機能のテストやフィードバックの収集を可能にするために提供されます。サポート範囲の詳細は、テクノロジープレビュー機能のサポート範囲 を参照してください。
スナップショットを柔軟に管理するため、Debezium には 増分スナップショット と呼ばれる補助スナップショットメカニズムが含まれています。増分スナップショットは、Debezium コネクターにシグナルを送信するための Debezium メカニズムに依存します。
増分スナップショットでは、最初のスナップショットのように、データベースの完全な状態を一度にすべてキャプチャーする代わりに、一連の設定可能なチャンクで各テーブルを段階的にキャプチャーします。スナップショットがキャプチャーするテーブルと、各チャンクのサイズ を指定できます。チャンクのサイズにより、データベース上の各フェッチ操作中にスナップショットで収集される行数が決まります。増分スナップショットのデフォルトのチャンクサイズは 1 KB です。
増分スナップショットが進むと、Debezium はウォーターマークを使用して進捗を追跡し、キャプチャーする各テーブル行のレコードを管理します。この段階的なアプローチでは、標準の初期スナップショットプロセスと比較して、以下の利点があります。
- スナップショットが完了するまで、ストリーミングストリーミングを延期する代わりに、ストリームしたデータキャプチャーと並行して増分スナップショットを実行できます。コネクターはスナップショットプロセス全体で変更ログからのほぼリアルタイムイベントをキャプチャーし続け、他の操作はブロックしません。
- 増分スナップショットの進捗が中断された場合は、データを失うことなく再開できます。プロセスが再開すると、スナップショットは最初からテーブルをキャプチャーするのではなく、停止した時点から開始します。
-
いつでも増分スナップショットを実行し、必要に応じてプロセスを繰り返してデータベースの更新に適合できます。たとえば、コネクター設定を変更してテーブルを
table.include.list
プロパティーに追加した後にスナップショットを再実行します。
増分スナップショットプロセス
増分スナップショットを実行する場合には、Debezium は各テーブルをプライマリーキー別に分類して、設定されたチャンクサイズ に基づいてテーブルをチャンクに分割します。チャンクごとに作業し、テーブルの行ごとにチャンクでキャプチャーします。キャプチャーする行ごとに、スナップショットは READ
イベントを出力します。そのイベントは、対象となるチャンクのスナップショットを開始する時の行の値を表します。
スナップショットの作成が進むにつれ、他のプロセスがデータベースへのアクセスを継続し、テーブルレコードが変更される可能性があります。このような変更を反映させるように、通常通りに INSERT
、UPDATE
、DELETE
操作がトランザクションログにコミットされます。同様に、継続中の Debezium ストリーミングプロセスは、これらの変更イベントを検出し、対応する変更イベントレコードを Kafka に出力します。
Debezium を使用してプライマリーキーが同じレコード間での競合を解決する方法
場合によっては、ストリーミングプロセスが出力する UPDATE
または DELETE
イベントを順番に受信できます。つまり、ストリーミングプロセスは、スナップショットがその行の READ
イベントが含まれるチャンクをキャプチャーする前に、テーブルの行を変更するイベントを生成する可能性があります。スナップショットが最終的に対象の行にあった READ
イベントを出力すると、その値はすでに置き換えられています。Debezium は、シーケンスが到達する増分スナップショットイベントが正しい論理順序で処理されるように、競合を解決するためにバッファースキームを使用します。スナップショットのイベント間で競合が発生し、ストリームされたイベントが解決されてからでないと、Debezium はイベントのレコードを Kafka に送信しません。
スナップショットウィンドウ
遅れて入ってきた READ
イベントと、同じテーブルの行を変更するストリーミングイベント間の競合の解決を容易にするために、Debezium は スナップショットウィンドウ と呼ばれるものを使用します。スナップショットウィンドウは、増分スナップショットが指定のテーブルチャンクのデータをキャプチャーしている途中に、間隔を決定します。チャンクのスナップショットウィンドウを開く前に、Debezium は通常の動作に従い、トランザクションログから直接ターゲットの Kafka トピックにイベントをダウンストリームに出力します。ただし、特定のチャンクのスナップショットが開放された瞬間から終了するまで、Debezium は重複除去のステップを実行して、プライマリーキーが同じイベント間での競合を解決します。
データコレクションごとに、Debezium は 2 種類のイベントを出力し、それらの両方のレコードを単一の宛先 Kafka トピックに保存します。テーブルから直接キャプチャーするスナップショットレコードは、READ
操作として出力されます。その間、ユーザーはデータコレクションのレコードの更新を続け、各コミットを反映するようにトランザクションログが更新されるので、Debezium は変更ごとに UPDATE
または DELETE
操作を出力します。
スナップショットウィンドウが開放され、Debezium がスナップショットチャンクの処理を開始すると、スナップショットレコードをメモリーバッファーに提供します。スナップショットウィンドウ中に、バッファー内の READ
イベントのプライマリーキーは、受信ストリームイベントのプライマリーキーと比較されます。一致するものが見つからない場合、ストリーミングされたイベントレコードが Kafka に直接送信されます。Debezium が一致を検出すると、バッファーされた READ
イベントを破棄し、ストリーミングされたレコードを宛先トピックに書き込みます。これは、ストリーミングされたイベントが静的スナップショットイベントよりも論理的に優先されるためです。チャンクのスナップショットウィンドウが終了すると、バッファーに含まれるのは、関連するトランザクションログイベントが存在しない READ
イベントのみです。Debezium は、これらの残りの READ
イベントをテーブルの Kafka トピックに出力します。
コネクターは各スナップショットチャンクにプロセスを繰り返します。
増分スナップショットのトリガー
現在、増分スナップショットを開始する唯一の方法は、アドホックスナップショットシグナル をソースデータベースのシグナルテーブルに送信することです。SQL INSERT
クエリーとしてテーブルにシグナルを送信します。Debezium がシグナルテーブルの変更を検出すると、シグナルを読み取り、要求されたスナップショット操作を実行します。
送信するクエリーはスナップショットに追加するテーブルを指定し、必要に応じてスナップショット操作の種類を指定します。現在、スナップショット操作で唯一の有効なオプションはデフォルト値の incremental
だけです。
スナップショットに追加するテーブルを指定するには、テーブルを一覧表示する data-collections
アレイを指定します (例:
{"data-collections": ["public.MyFirstTable", "public.MySecondTable"]}
)。
増分スナップショットシグナルの data-collections
アレイにはデフォルト値がありません。data-collections
アレイが空である場合には、アクションが不要であり、スナップショットを実行しないことが、Debezium で検出されます。
前提条件
- シグナルデータコレクションがソースのデータベースに存在し、コネクターはこれをキャプチャーするように設定されています。
-
シグナルデータコレクションは
signal.data.collection
プロパティーで指定されます。
手順
SQL クエリーを送信し、アドホック増分スナップショット要求をシグナルテーブルに追加します。
INSERT INTO _<signalTable>_ (id, type, data) VALUES (_'<id>'_, _'<snapshotType>'_, '{"data-collections": ["_<tableName>_","_<tableName>_"],"type":"_<snapshotType>_"}');
以下に例を示します。
INSERT INTO myschema.debezium_signal (id, type, data) VALUES('ad-hoc-1', 'execute-snapshot', '{"data-collections": ["schema1.table1", "schema2.table2"],"type":"incremental"}');
コマンドの
id
、type
、およびdata
パラメーターの値は、シグナルテーブルのフィールド に対応します。以下の表では、これらのパラメーターについて説明しています。
表5.5 シグナルテーブルに増分スナップショットシグナルを送信する SQL コマンドのフィールドの説明 値 説明 myschema.debezium_signal
ソースデータベースにあるシグナルテーブルの完全修飾名を指定します。
ad-hoc-1
id
パラメーターは、シグナルリクエストのID
識別子として割り当てられる任意の文字列を指定します。
この文字列を使用して、シグナルテーブルのエントリーへのログメッセージを特定します。Debezium はこの文字列を使用しません。代わりに、スナップショット作成中に、Debezium は独自のID
文字列をウォーターマークシグナルとして生成します。execute-snapshot
type
パラメーターを指定し、シグナルがトリガーする操作を指定します。
data-collections
スナップショットに含めるテーブル名の配列を指定するシグナルの
data
フィールドの必須コンポーネント。
配列は、signal.data.collection
設定プロパティーにコネクターのシグナルテーブルの名前を指定するときに使用する形式で、完全修飾名別にテーブルを一覧表示します。incremental
実行するスナップショット操作の種類指定するシグナルの
data
フィールドの任意のtype
コンポーネント。
現在、唯一の有効なオプションはデフォルト値incremental
だけです。
シグナルテーブルに送信する SQL クエリーでのtype
値の指定は任意です。
値を指定しない場合には、コネクターは増分スナップショットを実行します。
以下の例は、コネクターによってキャプチャーされる増分スナップショットイベントの JSON を示しています。
以下に例を示します。増分スナップショットイベントメッセージ
{ "before":null, "after": { "pk":"1", "value":"New data" }, "source": { ... "snapshot":"incremental" 1 }, "op":"r", 2 "ts_ms":"1620393591654", "transaction":null }
項目 | フィールド名 | 説明 |
---|---|---|
1 |
|
実行するスナップショット操作タイプを指定します。 |
2 |
|
イベントタイプを指定します。 |
5.1.5. Debezium MySQL 変更イベントレコードを受信する Kafka トピックのデフォルト名
デフォルトでは、MySQL コネクターは、テーブルで発生するすべての INSERT
、UPDATE
、DELETE
操作の変更イベントを、そのテーブルに固有の単一の Apache Kafka トピックに書き込みます。
コネクターは以下の規則を使用して変更イベントトピックに名前を付けます。
serverName.databaseName.tableName
fulfillment
はサーバー名、inventory
はデータベース名で、データベースに orders
、customers
、および products
という名前のテーブルが含まれるとします。Debezium MySQL コネクターは、データベースのテーブルごとに 1 つずつ、3 つの Kafka トピックにイベントを出力します。
fulfillment.inventory.orders fulfillment.inventory.customers fulfillment.inventory.products
以下のリストは、デフォルト名のコンポーネントの定義を示しています。
- serverName
-
database.server.name
コネクター設定プロパティーで指定したコネクターの論理名です。 - schemaName
- 操作が発生したスキーマの名前。
- tableName
- 操作が発生したテーブルの名前。
コネクターは同様の命名規則を適用して、内部データベース履歴トピック (スキーマ変更トピック と トランザクションメタデータトピック) にラベルを付けます。
デフォルトのトピック名が要件を満たさない場合は、カスタムトピック名を設定できます。カスタムトピック名を設定するには、論理トピックルーティング SMT に正規表現を指定します。論理トピックルーティング SMT を使用してトピックの命名をカスタマイズする方法は、トピックルーティング を参照してください。
トランザクションメタデータ
Debezium は、トランザクション境界を表し、データ変更イベントメッセージをエンリッチするイベントを生成できます。
Debezium は、コネクターのデプロイ後に発生するトランザクションに対してのみメタデータを登録し、受信します。コネクターをデプロイする前に発生するトランザクションのメタデータは利用できません。
Debezium は、すべてのトランザクションで BEGIN
および END
区切り文字のトランザクション境界イベントを生成します。トランザクション境界イベントには以下のフィールドが含まれます。
status
-
BEGIN
またはEND
id
- 一意のトランザクション識別子の文字列表現。
event_count
(END
イベント用)- トランザクションによって出力されるイベントの合計数。
data_collections
(END
イベント用)-
data_collection
とevent_count
要素のペアの配列。これは、コネクターがデータコレクションから発信された変更に対して出力するイベントの数を示します。
例
{ "status": "BEGIN", "id": "0e4d5dcd-a33b-11ea-80f1-02010a22a99e:10", "event_count": null, "data_collections": null } { "status": "END", "id": "0e4d5dcd-a33b-11ea-80f1-02010a22a99e:10", "event_count": 2, "data_collections": [ { "data_collection": "s1.a", "event_count": 1 }, { "data_collection": "s2.a", "event_count": 1 } ] }
コネクターはトランザクションイベントを <database.server.name>
.transaction
トピックに出力します。
変更データイベントのエンリッチメント
トランザクションメタデータを有効にすると、データメッセージ Envelope
は新しい transaction
フィールドでエンリッチされます。このフィールドは、複合フィールドの形式ですべてのイベントに関する情報を提供します。
-
id
- 一意のトランザクション識別子の文字列表現。 -
total_order
- トランザクションによって生成されたすべてのイベントを対象とするイベントの絶対位置。 -
data_collection_order
- トランザクションによって出力されたすべてのイベントを対象とするイベントのデータコレクションごとの位置。
以下は、メッセージの例になります。
{ "before": null, "after": { "pk": "2", "aa": "1" }, "source": { ... }, "op": "c", "ts_ms": "1580390884335", "transaction": { "id": "0e4d5dcd-a33b-11ea-80f1-02010a22a99e:10", "total_order": "1", "data_collection_order": "1" } }
GTID が有効ではないシステムの場合は、binlog のファイル名と binlog の位置の組み合わせを使用してトランザクション識別子が作成されます。たとえば、トランザクション BEGIN イベントに対応する binlog のファイル名と位置が mysql-bin.000002 および 1913 の場合には、Debezium が構築したトランザクション識別子は file=mysql-bin.000002,pos=1913
になります。
5.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、または _) でなければなりません。無効な文字がある場合は、アンダースコアに置き換えられます。
論理サーバー名、データベース名、またはテーブル名に無効な文字が含まれ、名前を区別する唯一の文字が無効であると、無効な文字はすべてアンダースコアに置き換えられるため、予期せぬ競合が発生する可能性があります。
詳細は以下を参照してください。
5.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 |
|
この変更イベントが生成された行のキーが含まれます。この例では、キーには値が |
5.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" },