Apache Kafkaへのシンク

このドキュメントでは、TiDB CloudからApache Kafkaへデータをストリーミングするためのチェンジフィードを作成する方法について説明します。

注記：

変更フィード機能を使用するには、 TiDB Cloud Dedicatedクラスタのバージョンがv6.1.3以降であることを確認してください。

制限

• TiDB Cloud Dedicatedクラスターごとに、最大 100 個の変更フィードを作成できます。
• 現在、 TiDB Cloudは、Kafkaブローカーに接続するための自己署名TLS証明書のアップロードをサポートしていません。
• TiDB Cloud はTiCDC を使用して変更フィードを確立するため、同じTiCDCの制限があります。
• 複製対象のテーブルに主キーまたはNULLを許容しない一意インデックスがない場合、複製中に一意制約が存在しないことで、一部の再試行シナリオにおいて、下流で重複データが挿入される可能性があります。

• ネットワーク接続方法としてプライベートリンクまたはプライベートサービスコネクトを選択する場合は、 TiDB Cloud Dedicatedクラスタのバージョンが以下の要件を満たしていることを確認してください。

  • v6.5.xの場合：バージョンv6.5.9以降
  • v7.1.xの場合：バージョンv7.1.4以降
  • v7.5.xの場合：バージョンv7.5.1以降
  • v8.1.xの場合：v8.1.x以降のすべてのバージョンがサポートされています。

• データ形式としてDebeziumを使用する場合は、 TiDB Cloud Dedicatedクラスターのバージョンがv8.1.0以降であることを確認してください。

• Kafkaメッセージのパーティション分散については、以下の点に注意してください。

  • 変更ログをプライマリキーまたはインデックス値に基づいて、指定したインデックス名を持つ Kafka パーティションに分散する場合は、 TiDB Cloud Dedicatedクラスターのバージョンが v7.5.0 以降であることを確認してください。
  • 変更ログを列値ごとにKafkaパーティションに分散する場合は、 TiDB Cloud Dedicatedクラスタのバージョンがv7.5.0以降であることを確認してください。

前提条件

Apache Kafkaにデータをストリーミングするためのチェンジフィードを作成する前に、以下の前提条件を満たす必要があります。

• ネットワーク接続を設定する
• Kafka ACL認証の権限を追加する

ネットワーク

TiDB Cloud DedicatedクラスターApache Kafka サービスに接続できることを確認します。次の接続方法のいずれかを選択できます。

• プライベート接続: VPC CIDR の競合を回避し、セキュリティ コンプライアンスを満たすのに最適ですが、 プライベートデータリンクのコストが発生します。
• VPCピアリング：費用対効果の高い選択肢として適していますが、潜在的なVPC CIDRの競合やセキュリティ上の考慮事項を管理する必要があります。
• パブリックIP：迅速なセットアップに適しています。

Kafka ACL認証

TiDB Cloudの変更フィードがデータをApache Kafkaにストリーミングし、Kafkaトピックを自動的に作成できるようにするには、Kafkaに以下の権限が追加されていることを確認してください。

• Kafka のトピック リソース タイプにCreateおよびWrite権限が追加されます。
• Kafka のクラスタ リソース タイプにDescribeConfigs権限が追加されます。

たとえば、Kafka クラスターが Confluent Cloud にある場合、詳細については Confluent ドキュメントのリソースとACLの追加参照してください。

ステップ1. Apache KafkaのChangefeedページを開きます。

1. TiDB Cloudコンソールにログインします。
2. ターゲットのTiDB Cloud Dedicatedクラスターの概要ページに移動し、左側のナビゲーション ペインで[データ] > [変更フィード]をクリックします。
3. 「変更フィードの作成」をクリックし、宛先として「Kafka」を選択します。

ステップ2. changefeedターゲットを設定する

手順は、選択した接続方法によって異なります。

ステップ3．チェンジフィードを設定する

1. テーブル フィルターをカスタマイズして、複製するテーブルをフィルターします。ルールの構文については、テーブルフィルタルールを参照してください。

   • 大文字小文字の区別：フィルタルールにおけるデータベース名とテーブル名の照合において、大文字小文字を区別するかどうかを設定できます。デフォルトでは、大文字小文字は区別されません。
   • フィルタルール：この列でフィルタルールを設定できます。デフォルトでは、すべてのテーブルを複製するルール*.*が設定されています。新しいルールを追加すると、 TiDB Cloud はTiDB 内のすべてのテーブルをクエリし、右側のボックスにルールに一致するテーブルのみを表示します。フィルタルールは最大 100 個まで追加できます。
   • 有効なキーを持つテーブル：この列には、主キーや一意インデックスなど、有効なキーを持つテーブルが表示されます。
   • 有効なキーのないテーブル: この列には、主キーまたは一意キーがないテーブルが表示されます。一意の識別子がないと、ダウンストリームが重複イベントを処理する際にデータの一貫性が失われる可能性があるため、これらのテーブルはレプリケーション中に問題となります。データの一貫性を確保するには、レプリケーションを開始する前に、これらのテーブルに一意キーまたは主キーを追加することをお勧めします。または、フィルタルールを追加してこれらのテーブルを除外することもできます。たとえば、ルールtest.tbl1を使用して、テーブル"!test.tbl1"除外できます。

2. イベントフィルターをカスタマイズして、複製したいイベントを絞り込みます。

   • 一致するテーブル：この列では、イベントフィルターを適用するテーブルを設定できます。ルールの構文は、前のテーブルフィルター領域で使用されているものと同じです。変更フィードごとに最大10個のイベントフィルタールールを追加できます。
   • イベントフィルター：以下のイベントフィルターを使用して、変更フィードから特定のイベントを除外できます。
     • イベントを無視する：指定されたイベントタイプを除外します。
     • SQL を無視: 指定された式に一致する DDL イベントを除外します。たとえば、 ^drop DROPで始まるステートメントを除外し、 add columnはADD COLUMNを含むステートメントを除外します。
     • 挿入値の式を無視する: 特定の条件を満たすINSERTステートメントを除外します。たとえば、 id >= 100は、 INSERTが 100 以上であるidステートメントを除外します。
     • 新しい値の更新式を無視する: 新しい値が指定された条件に一致するUPDATEステートメントを除外します。たとえば、 gender = 'male'はgenderがmaleになるような更新を除外します。
     • 古い値の更新を無視する式: 古い値が指定された条件に一致するUPDATEステートメントを除外します。たとえば、 age < 18 ageの古い値が 18 未満である場合の更新を除外します。
     • 削除値式を無視する: 指定された条件を満たすDELETEステートメントを除外します。たとえば、 name = 'john'はDELETEがnameである'john'ステートメントを除外します。

3. カラムセレクタをカスタマイズして、イベントから列を選択し、選択した列に関連するデータ変更のみを下流に送信します。

   • 一致するテーブル：列セレクターを適用するテーブルを指定します。どのルールにも一致しないテーブルの場合、すべての列が送信されます。
   • カラムセレクター：一致したテーブルのどの列をダウンストリームに送信するかを指定します。

   マッチングルールの詳細については、 カラムセレクターを参照してください。

4. データフォーマット領域で、希望するKafkaメッセージのフォーマットを選択してください。

   • Avroは、コンパクトで高速なバイナリデータフォーマットであり、豊富なデータ構造を備え、様々なフローシステムで広く利用されています。詳細については、 Avroデータ形式参照してください。
   • Canal-JSONは、解析が容易なプレーンなJSONテキスト形式です。詳細については、 Canal-JSONデータ形式参照してください。
   • オープン プロトコルは、監視、キャッシュ、全文インデックス作成、分析エンジン、および異なるデータベース間のプライマリとセカンダリのレプリケーションのためのデータ ソースを提供する行レベルのデータ変更通知プロトコルです。詳細については、 オープンプロトコルデータフォーマットを参照してください。
   • Debeziumは、データベースの変更をキャプチャするためのツールです。キャプチャされた各データベース変更を「イベント」と呼ばれるメッセージに変換し、これらのイベントをKafkaに送信します。詳細については、 Debeziumデータ形式参照してください。

5. TiDB拡張フィールドをKafkaメッセージ本文に追加する場合は、 TiDB拡張オプションを有効にしてください。

   TiDB 拡張フィールドの詳細については、 Avroデータ形式のTiDB拡張フィールドフィールド」およびCanal-JSONデータ形式のTiDB拡張フィールド参照してください。

6. データ形式としてAvroを選択すると、ページにAvro固有の設定項目が表示されます。これらの設定項目は、以下のように入力できます。

   • 「Decimal」および「Unsigned BigInt」の設定では、 TiDB CloudがKafkaメッセージ内のdecimal型およびunsigned bigint型データ型をどのように処理するかを指定します。
   • スキーマレジストリ領域で、スキーマレジストリエンドポイントを入力します。HTTP認証を有効にすると、ユーザー名とパスワードのフィールドが表示され、 TiDB Cloud Dedicatedクラスターのエンドポイントとパスワードが自動的に入力されます。TiDB 。

7. トピック配信エリアで配信モードを選択し、選択したモードに応じてトピック名の設定を入力します。

   データ形式としてAvroを選択した場合、 [配信モード]ドロップダウンリストでは、[テーブルごとの変更ログをKafkaトピックに配信する]モードのみを選択できます。

   配信モードは、変更フィードがKafkaトピックを作成する方法を制御します。テーブルごと、データベースごと、またはすべての変更ログに対して1つのトピックを作成するかを選択できます。

   • テーブルごとに変更ログをKafkaトピックに配信する

     変更フィードでテーブルごとに専用の Kafka トピックを作成する場合は、このモードを選択します。そうすると、テーブルのすべての Kafka メッセージが専用の Kafka トピックに送信されます。トピックのプレフィックス、データベース名とテーブル名の間の区切り文字、およびサフィックスを設定することで、テーブルのトピック名をカスタマイズできます。たとえば、区切り文字を_に設定すると、トピック名は<Prefix><DatabaseName>_<TableName><Suffix>の形式になります。

     スキーマ作成イベントなど、行以外のイベントの変更ログについては、 「デフォルトトピック名」フィールドにトピック名を指定できます。変更フィードは、指定されたトピックに基づいて、これらの変更ログを収集するためのトピックを作成します。

   • データベースごとに変更ログをKafkaトピックに配信する

     変更フィードでデータベースごとに専用のKafkaトピックを作成する場合は、このモードを選択してください。そうすると、データベースのすべてのKafkaメッセージが専用のKafkaトピックに送信されます。トピックのプレフィックスとサフィックスを設定することで、データベースのトピック名をカスタマイズできます。

     解決済みTsイベントなど、行以外のイベントの変更ログについては、 「デフォルトトピック名」フィールドにトピック名を指定できます。変更フィードは、指定されたトピックに基づいて、これらの変更ログを収集するためのトピックを作成します。

   • すべての変更ログを、指定された1つのKafkaトピックに送信する

     変更フィードで全ての変更ログに対して1つのKafkaトピックを作成する場合は、このモードを選択してください。そうすると、変更フィード内のすべてのKafkaメッセージが1つのKafkaトピックに送信されます。トピック名は「トピック名」フィールドで指定できます。

8. パーティション分散領域では、Kafka メッセージの送信先パーティションを決定できます。すべてのテーブルに対して単一のパーティションディスパッチャを定義することも、テーブルごとに異なるパーティションディスパッチャを定義することもできます。TiDB Cloud、次の 4 種類のディスパッチャが提供されています。

   • プライマリキーまたはインデックス値に基づいて変更ログをKafkaパーティションに分散します。

     変更フィードを使用してテーブルの Kafka メッセージを異なるパーティションに送信する場合は、この分散方法を選択してください。行の変更ログの主キーまたはインデックス値によって、変更ログが送信されるパーティションが決まります。この分散方法は、パーティションのバランスを改善し、行レベルの順序性を確保します。

   • 変更ログをテーブルごとにKafkaパーティションに分散する

     変更フィードによってテーブルの Kafka メッセージを 1 つの Kafka パーティションに送信する場合は、この分散方法を選択してください。行の変更ログのテーブル名によって、変更ログが送信されるパーティションが決まります。この分散方法はテーブルの順序性を保証しますが、パーティションのバランスが崩れる可能性があります。

   • 変更ログをタイムスタンプに基づいてKafkaパーティションに分散する

     変更フィードが Kafka メッセージをランダムに異なる Kafka パーティションに送信するようにするには、この分散方法を選択してください。行の変更ログの commitTs によって、変更ログが送信されるパーティションが決まります。この分散方法は、パーティションのバランスを改善し、各パーティションの順序性を確保します。ただし、データ項目の複数の変更が異なるパーティションに送信され、異なるコンシューマーのコンシューマー処理の進行状況が異なる場合があり、データの不整合が発生する可能性があります。そのため、コンシューマーは、複数のパーティションからのデータを消費する前に commitTs でソートする必要があります。

   • 変更ログを列の値に基づいてKafkaパーティションに分散する

     テーブルのKafkaメッセージを複数のパーティションに送信するように変更フィードを設定したい場合は、この配信方法を選択してください。行の変更ログで指定された列の値によって、変更ログの送信先パーティションが決まります。この配信方法では、各パーティション内の順序が確保され、同じ列の値を持つ変更ログが同じパーティションに送信されることが保証されます。

9. トピックコンフィグレーション領域で、以下の数値を設定してください。changefeedは、これらの数値に基づいてKafkaトピックを自動的に作成します。

   • レプリケーション係数：各KafkaメッセージがレプリケートされるKafkaサーバーの数を制御します。有効な値の範囲は、 min.insync.replicasからKafkaブローカーの数までです。
   • パーティション番号：トピックに存在するパーティションの数を制御します。有効な値の範囲は[1, 10 * the number of Kafka brokers]です。

10. [イベントの分割]エリアで、 UPDATEイベントを別々のDELETEとINSERTイベントに分割するか、生のUPDATEイベントとして保持するかを選択します。詳細については、 MySQL以外のシンクにおける、主キーまたは一意キーを分割したUPDATEイベント参照してください。

11. 「次へ」をクリックしてください。

ステップ4. 変更フィード仕様を設定します

1. 「チェンジフィードの仕様」領域で、チェンジフィードで使用する複製容量単位（RCU）チェンジフィードの数を指定します。
2. 変更フィード名欄に、変更フィードの名前を指定します。
3. 「次へ」をクリックして、設定した内容を確認し、次のページへ進んでください。

ステップ5．設定内容を確認する

このページでは、設定したすべての変更フィード設定を確認できます。

エラーが見つかった場合は、戻って修正できます。エラーがない場合は、下部のチェックボックスをクリックし、 「作成」をクリックして変更フィードを作成します。