Kafka にデータを複製する

このドキュメントでは、TiCDC を使用して増分データを Apache Kafka に複製する変更フィードを作成する方法について説明します。

レプリケーションタスクを作成する

次のコマンドを実行してレプリケーション タスクを作成します。

cdc cli changefeed create \
    --server=http://10.0.10.25:8300 \
    --sink-uri="kafka://127.0.0.1:9092,127.0.0.1:9093,127.0.0.1:9094/topic-name?protocol=canal-json&kafka-version=2.4.0&partition-num=6&max-message-bytes=67108864&replication-factor=1" \
    --changefeed-id="simple-replication-task"

Create changefeed successfully!
ID: simple-replication-task
Info: {"sink-uri":"kafka://127.0.0.1:9092,127.0.0.1:9093,127.0.0.1:9094/topic-name?protocol=canal-json&kafka-version=2.4.0&partition-num=6&max-message-bytes=67108864&replication-factor=1","opts":{},"create-time":"2023-11-28T22:04:08.103600025+08:00","start-ts":415241823337054209,"target-ts":0,"admin-job-type":0,"sort-engine":"unified","sort-dir":".","config":{"case-sensitive":false,"filter":{"rules":["*.*"],"ignore-txn-start-ts":null,"ddl-allow-list":null},"mounter":{"worker-num":16},"sink":{"dispatchers":null},"scheduler":{"type":"table-number","polling-time":-1}},"state":"normal","history":null,"error":null}

• --server : TiCDC クラスター内の任意の TiCDCサーバーのアドレス。
• --changefeed-id : レプリケーションタスクのID。形式は正規表現^[a-zA-Z0-9]+(\-[a-zA-Z0-9]+)*$一致する必要があります。このIDが指定されていない場合、TiCDCは自動的にUUID（バージョン4形式）をIDとして生成します。
• --sink-uri : レプリケーションタスクのダウンストリームアドレス。詳細はkafkaでシンクURIを設定する参照してください。
• --start-ts : チェンジフィードの開始TSOを指定します。このTSOから、TiCDCクラスターはデータのプルを開始します。デフォルト値は現在時刻です。
• --target-ts : チェンジフィードの終了TSOを指定します。このTSOまで、TiCDCクラスターはデータのプルを停止します。デフォルト値は空で、TiCDCはデータのプルを自動的に停止しません。
• --config : changefeed設定ファイルを指定します。詳細はTiCDC Changefeedコンフィグレーションパラメータ参照してください。

サポートされているKafkaのバージョン

次の表は、各 TiCDC バージョンでサポートされる最小 Kafka バージョンを示しています。

Kafka のシンク URI を構成する

シンクURIは、TiCDCターゲットシステムの接続情報を指定するために使用されます。形式は次のとおりです。

[scheme]://[host]:[port][/path]?[query_parameters]

ヒント：

下流のKafkaに複数のホストまたはポートがある場合は、シンクURIに複数の[host]:[port]設定できます。例：

[scheme]://[host]:[port],[host]:[port],[host]:[port][/path]?[query_parameters]

サンプル構成:

--sink-uri="kafka://127.0.0.1:9092/topic-name?protocol=canal-json&kafka-version=2.4.0&partition-num=6&max-message-bytes=67108864&replication-factor=1"

以下は、Kafka に対して構成できるシンク URI パラメータと値の説明です。

ベストプラクティス

• 独自のKafkaトピックを作成することをお勧めします。少なくとも、トピックがKafkaブローカーに送信できる各メッセージの最大データ量と、下流のKafkaパーティションの数を設定する必要があります。チェンジフィードを作成する場合、これらの2つの設定はそれぞれmax-message-bytesとpartition-numに対応します。
• まだ存在しないトピックでチェンジフィードを作成した場合、TiCDCはpartition-numとreplication-factorパラメータを使用してトピックを作成しようとします。これらのパラメータは明示的に指定することをお勧めします。
• ほとんどの場合、 canal-jsonプロトコルを使用することをお勧めします。

注記：

protocolがopen-protocol場合、TiCDC は複数のイベントを 1 つの Kafka メッセージにエンコードし、 max-message-bytesで指定された長さを超えるメッセージの生成を回避します。1 行の変更イベントのエンコード結果がmax-message-bytesを超える場合、変更フィードはエラーを報告し、ログを出力。

TiCDCはKafkaの認証と認可を使用します

以下は、Kafka SASL 認証を使用する場合の例です。

• SASL/プレーン

  --sink-uri="kafka://127.0.0.1:9092/topic-name?kafka-version=2.4.0&sasl-user=alice-user&sasl-password=alice-secret&sasl-mechanism=plain"
  

• SASL/SCRAM

  SCRAM-SHA-256とSCRAM-SHA-512はPLAIN方式に似ています。対応する認証方式としてsasl-mechanism指定するだけです。

• SASL/GSSAPI

  SASL/GSSAPI user認証:

  --sink-uri="kafka://127.0.0.1:9092/topic-name?kafka-version=2.4.0&sasl-mechanism=gssapi&sasl-gssapi-auth-type=user&sasl-gssapi-kerberos-config-path=/etc/krb5.conf&sasl-gssapi-service-name=kafka&sasl-gssapi-user=alice/for-kafka&sasl-gssapi-password=alice-secret&sasl-gssapi-realm=example.com"
  

  sasl-gssapi-userとsasl-gssapi-realmの値は、Kerberos で指定されている原理と関連しています。例えば、プリンシパルがalice/for-kafka@example.comに設定されている場合、 sasl-gssapi-userとsasl-gssapi-realmそれぞれalice/for-kafkaとexample.comとして指定されます。

  SASL/GSSAPI keytab認証:

  --sink-uri="kafka://127.0.0.1:9092/topic-name?kafka-version=2.4.0&sasl-mechanism=gssapi&sasl-gssapi-auth-type=keytab&sasl-gssapi-kerberos-config-path=/etc/krb5.conf&sasl-gssapi-service-name=kafka&sasl-gssapi-user=alice/for-kafka&sasl-gssapi-keytab-path=/var/lib/secret/alice.key&sasl-gssapi-realm=example.com"
  

  SASL/GSSAPI 認証方式の詳細については、 GSSAPIの設定参照してください。

• TLS/SSL暗号化

  KafkaブローカーでTLS/SSL暗号化が有効になっている場合は、 --sink-uriに-enable-tls=trueパラメータを追加する必要があります。自己署名証明書を使用する場合は、 --sink-uriにca 、 cert 、 keyも指定する必要があります。

• ACL認証

  TiCDC が適切に機能するために必要な最小限の権限セットは次のとおりです。

  • トピックリソースタイプのCreate 、 Write 、およびDescribe権限。
  • クラスタリソース タイプに対するDescribeConfig権限。

  各権限の使用シナリオは次のとおりです。

  リソースタイプ	操作の種類	シナリオ
  クラスタ	DescribeConfig	変更フィードの実行中にクラスターのメタデータを取得します
  トピック	Describe	チェンジフィードの開始時にトピックを作成しようとします
  トピック	Create	チェンジフィードの開始時にトピックを作成しようとします
  トピック	Write	トピックにデータを送信します

  変更フィードを作成または開始するときに、指定された Kafka トピックがすでに存在する場合は、 DescribeおよびCreate権限を無効にすることができます。

TiCDC を Kafka Connect (Confluent Platform) と統合する

Confluent が提供するデータコネクタ使用して、データをリレーショナル データベースまたは非リレーショナル データベースにストリーミングするには、 avroプロトコル使用し、 schema-registryでConfluent スキーマレジストリ URL を指定する必要があります。

サンプル構成:

--sink-uri="kafka://127.0.0.1:9092/topic-name?&protocol=avro&replication-factor=3" --schema-registry="http://127.0.0.1:8081" --config changefeed_config.toml

[sink]
dispatchers = [
 {matcher = ['*.*'], topic = "tidb_{schema}_{table}"},
]

詳細な統合ガイドについては、 TiDB と Confluent Platform の統合に関するクイック スタート ガイド参照してください。

TiCDC を AWS Glue スキーマレジストリと統合する

v7.4.0以降、TiCDCは、ユーザーがデータレプリケーションにアブロプロトコル選択した場合、スキーマレジストリとしてAWS Glue スキーマレジストリ使用をサポートします。設定例は次のとおりです。

./cdc cli changefeed create --server=127.0.0.1:8300 --changefeed-id="kafka-glue-test" --sink-uri="kafka://127.0.0.1:9092/topic-name?&protocol=avro&replication-factor=3" --config changefeed_glue.toml

[sink]
[sink.kafka-config.glue-schema-registry-config]
region="us-west-1"  
registry-name="ticdc-test"
access-key="xxxx"
secret-access-key="xxxx"
token="xxxx"

上記の設定では、 regionとregistry-name必須フィールドですが、 access-key 、 secret-access-key 、 tokenオプションフィールドです。AWS認証情報は、changefeed設定ファイルではなく、環境変数として設定するか、 ~/.aws/credentialsファイルに保存するのがベストプラクティスです。

詳細については、 Go V2 向け公式 AWS SDK ドキュメントを参照してください。

Kafka シンクのトピックおよびパーティションディスパッチャーのルールをカスタマイズする

マッチャールール

dispatchersの次の構成を例に挙げます。

[sink]
dispatchers = [
  {matcher = ['test1.*', 'test2.*'], topic = "Topic expression 1", partition = "ts" },
  {matcher = ['test3.*', 'test4.*'], topic = "Topic expression 2", partition = "index-value" },
  {matcher = ['test1.*', 'test5.*'], topic = "Topic expression 3", partition = "table"},
  {matcher = ['test6.*'], partition = "ts"}
]

• マッチャールールに一致するテーブルは、対応するトピック式で指定されたポリシーに従ってディスパッチされます。例えば、テーブルtest3.aaは「トピック式2」に従ってディスパッチされ、テーブルtest5.aaは「トピック式3」に従ってディスパッチされます。
• 複数のマッチャールールに一致するテーブルの場合、最初に一致するトピック式に従ってディスパッチされます。例えば、 test1.aaテーブルは「トピック式 1」に従ってディスパッチされます。
• どのマッチャールールにも一致しないテーブルの場合、対応するデータ変更イベントは--sink-uriで指定されたデフォルトトピックに送信されます。例えば、 test10.aaテーブルはデフォルトトピックに送信されます。
• マッチャールールに一致するもののトピックディスパッチャーを指定していないテーブルの場合、対応するデータ変更は--sink-uriで指定されたデフォルトトピックに送信されます。例えば、 test6.aaテーブルはデフォルトトピックに送信されます。

トピックディスパッチャ

topic = "xxx" を使用してトピックディスパッチャを指定し、トピック式を使用して柔軟なトピックディスパッチポリシーを実装できます。トピックの総数は1000未満にすることをお勧めします。

Topic 式の形式は[prefix]{schema}[middle][{table}][suffix]です。

• prefix : オプション。トピック名のプレフィックスを示します。
• {schema} : 必須。スキーマ名を一致させるために使用されます。v7.1.4以降、このパラメータはオプションです。
• middle : オプション。スキーマ名とテーブル名の間の区切り文字を示します。
• {table} : オプション。テーブル名と一致させるために使用されます。
• suffix : オプション。トピック名の接尾辞を示します。

prefix 、 middle 、 suffix 、 a-z 、 A-Z 、 0-9 、 . 、 _ 、 -の文字のみを使用できます。 {schema}と{table}どちらも小文字です。 {Schema}や{TABLE}などのプレースホルダーは無効です。

例:

• matcher = ['test1.table1', 'test2.table2'], topic = "hello_{schema}_{table}"
  • test1.table1に対応するデータ変更イベントは、 hello_test1_table1という名前のトピックに送信されます。
  • test2.table2に対応するデータ変更イベントは、 hello_test2_table2という名前のトピックに送信されます。
• matcher = ['test3.*', 'test4.*'], topic = "hello_{schema}_world"
  • test3内のすべてのテーブルに対応するデータ変更イベントは、 hello_test3_worldという名前のトピックに送信されます。
  • test4内のすべてのテーブルに対応するデータ変更イベントは、 hello_test4_worldという名前のトピックに送信されます。
• matcher = ['test5.*, 'test6.*'], topic = "hard_code_topic_name"
  • test5とtest6のすべてのテーブルに対応するデータ変更イベントは、トピックhard_code_topic_nameに送信されます。トピック名は直接指定できます。
• matcher = ['*.*'], topic = "{schema}_{table}"
  • TiCDC が監視するすべてのテーブルは、「schema_table」ルールに従って個別のトピックにディスパッチされます。例えば、テーブルtest.account場合、TiCDC はデータ変更ログをtest_accountという名前のトピックにディスパッチします。

DDLイベントをディスパッチする

スキーマレベルのDDL

特定のテーブルに関連しないDDLは、スキーマレベルDDLと呼ばれます（ create databaseやdrop databaseなど）。スキーマレベルDDLに対応するイベントは、 --sink-uriで指定されたデフォルトトピックに送信されます。

テーブルレベルのDDL

特定のテーブルに関連するDDLは、テーブルレベルDDLと呼ばれます（例： alter table create table 。テーブルレベルDDLに対応するイベントは、ディスパッチャの設定に従って対応するトピックに送信されます。

たとえば、 matcher = ['test.*'], topic = {schema}_{table}ようなディスパッチャーの場合、DDL イベントは次のようにディスパッチされます。

• DDLイベントに単一のテーブルが関係している場合、DDLイベントは対応するトピックにそのまま送信されます。例えば、DDLイベントdrop table test.table1場合、イベントはtest_table1という名前のトピックに送信されます。
• DDLイベントに複数のテーブルが関係する場合（ rename table / drop table / drop view複数のテーブルに関係する可能性があります）、DDLイベントは複数のイベントに分割され、対応するトピックに送信されます。例えば、DDLイベントrename table test.table1 to test.table10, test.table2 to test.table20場合、イベントrename table test.table1 to test.table10トピックtest_table1に送信され、イベントrename table test.table2 to test.table20トピックtest.table2に送信されます。

パーティションディスパッチャ

partition = "xxx"パーティションディスパッチャを指定するために使用できます。3、5、7、9、11 default index-value tsのディスパッチャcolumnsサポートされてtableます。ディスパッチャのルールは次のとおりです。

• default : デフォルトでtableディスパッチャルールを使用します。スキーマ名とテーブル名に基づいてパーティション番号が計算され、テーブルからのデータが必ず同じパーティションに送信されます。その結果、1つのテーブルからのデータは1つのパーティションにのみ存在し、順序付けが保証されます。ただし、このディスパッチャルールは送信スループットを制限し、コンシューマーを追加しても消費速度を向上させることはできません。
• index-value : 主キー、一意のインデックス、またはindexで明示的に指定されたインデックスのいずれかを使用してパーティション番号を計算し、テーブルデータを複数のパーティションに分散します。単一のテーブルのデータは複数のパーティションに送信され、各パーティションのデータは順序付けされます。コンシューマーを追加することで、消費速度を向上させることができます。
• columns : 明示的に指定された列の値に基づいてパーティション番号を計算し、テーブルデータを複数のパーティションに分散します。単一のテーブルのデータは複数のパーティションに送信され、各パーティションのデータは順序付けされます。コンシューマーを追加することで、消費速度を向上させることができます。
• table : スキーマ名とテーブル名を使用してパーティション番号を計算します。
• ts : 行変更のコミットタイムを用いてパーティション番号を計算し、テーブルデータを複数のパーティションに分散します。単一テーブルのデータは複数のパーティションに送信され、各パーティションのデータは順序付けされます。コンシューマーを追加することで、消費速度を向上させることができます。ただし、データ項目に対する複数の変更が異なるパーティションに送信され、各コンシューマーのコンシューマー進行状況が異なる場合があり、データの不整合が発生する可能性があります。そのため、コンシューマーは複数のパーティションからデータを消費する前に、コミットタイムでソートする必要があります。

dispatchersの次の構成を例に挙げます。

[sink]
dispatchers = [
    {matcher = ['test.*'], partition = "index-value"},
    {matcher = ['test1.*'], partition = "index-value", index = "index1"},
    {matcher = ['test2.*'], partition = "columns", columns = ["id", "a"]},
    {matcher = ['test3.*'], partition = "table"},
]

• testデータベース内のテーブルはindex-valueディスパッチャを使用し、主キーまたは一意のインデックスの値を使用してパーティション番号を計算します。主キーが存在する場合は主キーが使用され、存在しない場合は最短の一意のインデックスが使用されます。
• test1テーブル内のテーブルはindex-valueディスパッチャを使用し、 index1という名前のインデックスに含まれるすべての列の値を使用してパーティション番号を計算します。指定されたインデックスが存在しない場合はエラーが報告されますindexで指定されたインデックスは一意のインデックスである必要があります。
• test2データベース内のテーブルはcolumnsディスパッチャを使用し、列idとaの値を使用してパーティション番号を計算します。いずれかの列が存在しない場合は、エラーが報告されます。
• test3データベース内のテーブルはtableディスパッチャーを使用します。
• test4データベース内のテーブルは、前述のルールのいずれにも一致しないため、 defaultディスパッチャ、つまりtableディスパッチャを使用します。

テーブルが複数のディスパッチャー ルールに一致する場合、最初に一致するルールが優先されます。

注記：

バージョン6.1.0以降、設定の意味を明確にするため、パーティションディスパッチャーを指定するための設定がdispatcherからpartitionに変更されました。5 partition dispatcherの別名です。例えば、次の2つのルールは全く同じ意味です。

[sink]
dispatchers = [
   {matcher = ['*.*'], dispatcher = "index-value"},
   {matcher = ['*.*'], partition = "index-value"},
]

ただし、 dispatcherとpartition同じルール内に出現させることはできません。例えば、次のルールは無効です。

{matcher = ['*.*'], dispatcher = "index-value", partition = "table"},

カラムセレクター

列セレクター機能は、イベントから列を選択し、それらの列に関連するデータの変更のみをダウンストリームに送信することをサポートします。

column-selectorsの次の構成を例に挙げます。

[sink]
column-selectors = [
    {matcher = ['test.t1'], columns = ['a', 'b']},
    {matcher = ['test.*'], columns = ["*", "!b"]},
    {matcher = ['test1.t1'], columns = ['column*', '!column1']},
    {matcher = ['test3.t'], columns = ["column?", "!column1"]},
]

• 表test.t1場合、列aとbのみが送信されます。
• testデータベース内のテーブル ( t1テーブルを除く) の場合、 bを除くすべての列が送信されます。
• 表test1.t1場合、 column1を除く、 columnで始まるすべての列が送信されます。
• 表test3.t場合、 column1を除く、 columnで始まる 7 文字の列が送信されます。
• どのルールにも一致しないテーブルの場合、すべての列が送信されます。

注記：

column-selectorsルールでフィルタリングされた後、テーブル内のデータは、複製されるために主キーまたは一意キーを持っている必要があります。そうでない場合、変更フィードは作成時または実行時にエラーを報告します。

単一の大きなテーブルの負荷を複数の TiCDC ノードにスケールアウトする

この機能は、単一の大規模テーブルのデータレプリケーション範囲を、データ量と1分あたりの変更行数に応じて複数の範囲に分割し、各範囲でレプリケーションされるデータ量と変更行数をほぼ一定にします。この機能は、これらの範囲を複数のTiCDCノードに分散してレプリケーションすることで、複数のTiCDCノードが同時に単一の大規模テーブルをレプリケーションできるようにします。この機能により、以下の2つの問題が解決されます。

• 単一の TiCDC ノードでは、単一の大きなテーブルを時間内に複製することはできません。
• TiCDC ノードによって消費されるリソース (CPU やメモリなど) は均等に分散されません。

サンプル構成:

[scheduler]
# The default value is "false". You can set it to "true" to enable this feature.
enable-table-across-nodes = true
# When you enable this feature, it only takes effect for tables with the number of regions greater than the `region-threshold` value.
region-threshold = 100000
# When you enable this feature, it takes effect for tables with the number of rows modified per minute greater than the `write-key-threshold` value.
# Note:
# * The default value of `write-key-threshold` is 0, which means that the feature does not split the table replication range according to the number of rows modified in a table by default.
# * You can configure this parameter according to your cluster workload. For example, if it is configured as 30000, it means that the feature will split the replication range of a table when the number of modified rows per minute in the table exceeds 30000.
# * When `region-threshold` and `write-key-threshold` are configured at the same time:
#   TiCDC will check whether the number of modified rows is greater than `write-key-threshold` first.
#   If not, next check whether the number of Regions is greater than `region-threshold`.
write-key-threshold = 30000

次の SQL ステートメントを使用して、テーブルに含まれる地域の数を照会できます。

SELECT COUNT(*) FROM INFORMATION_SCHEMA.TIKV_REGION_STATUS WHERE DB_NAME="database1" AND TABLE_NAME="table1" AND IS_INDEX=0;

Kafka トピックの制限を超えるメッセージを処理する

Kafkaトピックは受信できるメッセージのサイズに制限を設定します。この制限はパラメータmax.message.bytesによって制御されます。TiCDC Kafkaシンクがこの制限を超えるデータを送信した場合、変更フィードはエラーを報告し、データのレプリケーションを続行できません。この問題を解決するために、TiCDCは新しい設定large-message-handle-optionを追加し、以下のソリューションを提供します。

現在、この機能はCanal-JSONとOpen Protocolの2つのエンコーディングプロトコルをサポートしています。Canal-JSONプロトコルを使用する場合は、 sink-uriのうちenable-tidb-extension=true指定する必要があります。

TiCDCデータ圧縮

バージョン7.4.0以降、TiCDC Kafkaシンクは、エンコード直後にデータを圧縮し、圧縮データのサイズとメッセージサイズ制限を比較する機能をサポートしています。この機能により、サイズ制限を超えるメッセージの発生を効果的に削減できます。

構成例は次のとおりです。

[sink.kafka-config.large-message-handle]
# This configuration is introduced in v7.4.0.
# "none" by default, which means that the compression feature is disabled.
# Possible values are "none", "lz4", and "snappy". The default value is "none".
large-message-handle-compression = "none"

large-message-handle-compression有効になっている場合、コンシューマーが受信したメッセージは特定の圧縮プロトコルを使用してエンコードされ、コンシューマー アプリケーションは指定された圧縮プロトコルを使用してデータをデコードする必要があります。

この機能は、Kafka プロデューサーの圧縮機能とは異なります。

• large-message-handle-compressionで指定された圧縮アルゴリズムは、単一のKafkaメッセージを圧縮します。圧縮は、メッセージサイズの制限と比較する前に実行されます。
• 同時に、 sink-uriのcompressionパラメータを使用して圧縮アルゴリズムを設定することもできます。この圧縮アルゴリズムは、複数のKafkaメッセージを含むデータ送信リクエスト全体に適用されます。

large-message-handle-compression設定した場合、TiCDC はメッセージを受信すると、まずメッセージサイズ制限パラメータの値と比較し、サイズ制限を超えるメッセージを圧縮します。5 にcompression sink-uri設定した場合、TiCDC はsink-uri設定に基づいて、送信データ要求全体をシンクレベルで再度圧縮します。

前述の 2 つの圧縮方法の圧縮率は次のように計算されますcompression ratio = size before compression / size after compression * 100 。

ハンドルキーのみ送信

v7.3.0以降、TiCDC Kafkaシンクは、メッセージサイズが制限を超えた場合にハンドルキーのみを送信することをサポートします。これにより、メッセージサイズが大幅に削減され、Kafkaトピックの制限を超えたメッセージサイズに起因するチェンジフィードエラーやタスクの失敗を回避できます。ハンドルキーとは、以下のものを指します。

• 複製するテーブルに主キーがある場合、主キーはハンドル キーになります。
• テーブルに主キーがなく、NOT NULL 一意キーがある場合、NOT NULL 一意キーがハンドル キーになります。

サンプル構成は次のとおりです。

[sink.kafka-config.large-message-handle]
# large-message-handle-option is introduced in v7.3.0.
# Defaults to "none". When the message size exceeds the limit, the changefeed fails.
# When set to "handle-key-only", if the message size exceeds the limit, only the handle key is sent in the data field. If the message size still exceeds the limit, the changefeed fails.
large-message-handle-option = "claim-check"

ハンドルキーのみでメッセージを消費する

ハンドル キーのみのメッセージ形式は次のとおりです。

{
    "id": 0,
    "database": "test",
    "table": "tp_int",
    "pkNames": [
        "id"
    ],
    "isDdl": false,
    "type": "INSERT",
    "es": 1639633141221,
    "ts": 1639633142960,
    "sql": "",
    "sqlType": {
        "id": 4
    },
    "mysqlType": {
        "id": "int"
    },
    "data": [
        {
          "id": "2"
        }
    ],
    "old": null,
    "_tidb": {     // TiDB extension fields
        "commitTs": 429918007904436226,  // A TiDB TSO timestamp
        "onlyHandleKey": true
    }
}

Kafkaコンシューマーはメッセージを受信すると、まずonlyHandleKeyフィールドをチェックします。このフィールドが存在し、値がtrue場合、メッセージには完全なデータのハンドルキーのみが含まれていることを意味します。この場合、完全なデータを取得するには、上流のTiDBにクエリを実行し、 履歴データを読み取るためのtidb_snapshot使用する必要があります。

大きなメッセージを外部storageに送信する

バージョン7.4.0以降、TiCDC Kafkaシンクは、メッセージサイズが制限を超えた場合に、大容量メッセージを外部storageに送信できるようになりました。同時に、TiCDCは外部storage内の大容量メッセージのアドレスを含むメッセージをKafkaに送信します。これにより、メッセージサイズがKafkaトピックの制限を超えたことによる変更フィードの失敗を回避できます。

構成例は次のとおりです。

[sink.kafka-config.large-message-handle]
# large-message-handle-option is introduced in v7.3.0.
# Defaults to "none". When the message size exceeds the limit, the changefeed fails.
# When set to "handle-key-only", if the message size exceeds the limit, only the handle key is sent in the data field. If the message size still exceeds the limit, the changefeed fails.
# When set to "claim-check", if the message size exceeds the limit, the message is sent to external storage.
large-message-handle-option = "claim-check"
claim-check-storage-uri = "s3://claim-check-bucket"

large-message-handle-option "claim-check"に設定する場合、 claim-check-storage-uri有効な外部storageアドレスに設定する必要があります。そうでない場合、変更フィードの作成は失敗します。

ヒント

TiCDC における Amazon S3、GCS、Azure Blob Storage の URI パラメータの詳細については、 外部ストレージサービスのURI形式参照してください。

TiCDCは外部storageサービス上のメッセージをクリーンアップしません。データ利用者は外部storageサービスを独自に管理する必要があります。

外部storageから大きなメッセージを消費する

Kafkaコンシューマーは、外部storage内の大きなメッセージのアドレスを含むメッセージを受信します。メッセージの形式は次のとおりです。

{
    "id": 0,
    "database": "test",
    "table": "tp_int",
    "pkNames": [
        "id"
    ],
    "isDdl": false,
    "type": "INSERT",
    "es": 1639633141221,
    "ts": 1639633142960,
    "sql": "",
    "sqlType": {
        "id": 4
    },
    "mysqlType": {
        "id": "int"
    },
    "data": [
        {
          "id": "2"
        }
    ],
    "old": null,
    "_tidb": {     // TiDB extension fields
        "commitTs": 429918007904436226,  // A TiDB TSO timestamp
        "claimCheckLocation": "s3:/claim-check-bucket/${uuid}.json"
    }
}

メッセージにclaimCheckLocationフィールドが含まれている場合、Kafka コンシューマーは、フィールドで指定されたアドレスに従って、JSON 形式で保存された大容量メッセージデータを読み取ります。メッセージ形式は次のとおりです。

{
  key: "xxx",
  value: "xxx",
}

keyとvalueフィールドには、エンコードされた大きなメッセージが含まれています。これは、Kafka メッセージの対応するフィールドに送信されるはずでした。コンシューマーは、これらの 2 つの部分のデータを解析することで、大きなメッセージの内容を復元できます。