📣
TiDB Cloud Premium はパブリックプレビュー中です。エンタープライズワークロード向けの無制限のスケーリング、即時の弾力性、高度なセキュリティを提供します。このページは自動翻訳されたものです。原文はこちらからご覧ください。
AI
開発者
ベストプラクティス
API

単一行データの TiCDC データ整合性検証

v7.1.0以降、TiCDCはデータ整合性検証機能を導入しました。この機能は、 チェックサムアルゴリズムを使用して単一行データの整合性を検証します。この機能は、TiDBからデータを書き込み、TiCDCを介して複製し、Kafkaクラスターに書き込むプロセスでエラーが発生していないかどうかを検証するのに役立ちます。現在、この機能は、ダウンストリームとしてKafkaを使用し、プロトコルとしてSimpleまたはAvroを使用するチェンジフィードのみでサポートされています。チェックサムアルゴリズムの詳細については、 チェックサム計算アルゴリズムを参照してください。

機能を有効にする

TiCDCはデフォルトでデータ整合性検証を無効にしています。有効にするには、以下の手順を実行してください。

  1. tidb_enable_row_level_checksumシステム変数を設定して、アップストリーム TiDB クラスター内の単一行データのチェックサム整合性検証機能を有効にします。

    SET GLOBAL tidb_enable_row_level_checksum = ON;

    この構成は新しく作成されたセッションに対してのみ有効になるため、TiDB に再接続する必要があります。

  2. changefeed の作成時に--configパラメータで指定する設定ファイルに、次の設定を追加します。

    [integrity]
integrity-check-level = "correctness"
corruption-handle-level = "warn"

  3. データエンコード形式としてAvroを使用する場合は、 sink-uriにenable-tidb-extension=trueを設定する必要があります。ネットワーク転送中に数値精度が失われ、チェックサム検証エラーが発生するのを防ぐため、 avro-decimal-handling-mode=stringとavro-bigint-unsigned-handling-mode=stringを設定する必要があります。以下に例を示します。

    cdc cli changefeed create --server=http://127.0.0.1:8300 --changefeed-id="kafka-avro-checksum" --sink-uri="kafka://127.0.0.1:9092/topic-name?protocol=avro&enable-tidb-extension=true&avro-decimal-handling-mode=string&avro-bigint-unsigned-handling-mode=string" --schema-registry=http://127.0.0.1:8081 --config changefeed_config.toml

    上記の設定により、changefeed によって Kafka に書き込まれる各メッセージには、対応するデータのチェックサムが含まれます。これらのチェックサム値に基づいてデータの整合性を検証できます。

機能を無効にする

TiCDC ではデータ整合性の検証はデフォルトで無効になっています。この機能を有効にした後に無効にするには、次の手順を実行します。

  1. タスク構成の更新で説明したPause Task -> Modify Configuration -> Resume Taskプロセスに従い、changefeed の--configパラメータで指定された設定ファイル内の[integrity]構成をすべて削除します。

    [integrity]
integrity-check-level = "none"
corruption-handle-level = "warn"

  2. チェックサム整合性検証機能を無効にするには、上流のTiDBで次のSQL文を実行します（ tidb_enable_row_level_checksum ）。

    SET GLOBAL tidb_enable_row_level_checksum = OFF;

    上記の設定は、新しく作成されたセッションにのみ適用されます。TiDBに書き込みを行っているすべてのクライアントが再接続すると、changefeedによってKafkaに書き込まれるメッセージには、対応するデータのチェックサムが含まれなくなります。

チェックサムアルゴリズム

チェックサムV1

v8.4.0 より前では、TiDB および TiCDC はチェックサムの計算と検証に Checksum V1 を使用します。

単一行データのチェックサム整合性検証機能を有効にすると、TiDBはCRC32アルゴリズムを使用して各行のチェックサムを計算し、データと共にTiKVに書き込みます。TiCDCはTiKVからデータを読み取り、同じアルゴリズムを使用してチェックサムを再計算します。2つのチェックサムが等しい場合、TiDBからTiCDCへの転送中にデータの整合性が保たれていることを示します。

TiCDCはデータを特定の形式にエンコードし、Kafkaに送信します。Kafkaコンシューマーがデータを読み取ると、TiDBと同じCRC32アルゴリズムを使用して新しいチェックサムを計算します。新しいチェックサムがデータ内のチェックサムと一致する場合、TiCDCからKafkaコンシューマーへの送信中にデータの一貫性が保たれていることを示します。

チェックサムV2

v8.4.0 以降、TiDB および TiCDC では、 ADD COLUMNまたはDROP COLUMN操作後に更新イベントまたは削除イベントで古い値を検証する際の Checksum V1 の問題に対処するために Checksum V2 が導入されています。

v8.4.0 以降で作成されたクラスター、または v8.4.0 以降にアップグレードされたクラスターでは、単一行データのチェックサム検証が有効な場合、TiDB はデフォルトで Checksum V2 を使用します。TiCDC は Checksum V1 と V2 の両方を処理できます。この変更は TiDB と TiCDC の内部実装にのみ影響し、下流の Kafka コンシューマーにおけるチェックサムの計算方法には影響しません。

チェックサム計算アルゴリズム

チェックサム計算アルゴリズムの疑似コードは次のとおりです。

fn checksum(columns) {
    let result = 0
    for column in sort_by_schema_order(columns) {
        result = crc32.update(result, encode(column))
    }
    return result
}

  • columnsは列IDでソートする必要があります。Avroスキーマでは、フィールドは既に列IDでソートされているため、 columnsの順序をそのまま使用できます。

  • encode(column)関数は列の値をバイト列にエンコードします。エンコードのルールは列のデータ型によって異なります。具体的なルールは以下のとおりです。

    • TINYINT、SMALLINT、INT、BIGINT、MEDIUMINT、YEAR 型は UINT64 に変換され、リトルエンディアンでエンコードされます。たとえば、数値0x0123456789abcdefはhex'0x0123456789abcdef'としてエンコードされます。

    • FLOAT および DOUBLE 型は DOUBLE に変換され、その後 IEEE754 形式の UINT64 としてエンコードされます。

    • BIT、ENUM、SET 型は UINT64 に変換されます。

      • BIT 型はバイナリ形式の UINT64 に変換されます。
      • ENUM型とSET型は、UINT64の対応するINT値に変換されます。例えば、 SET('a','b','c')型の列のデータ値が'a,c'の場合、その値は0b101 （10進数では5）としてエンコードされます。

    • TIMESTAMP、DATE、DURATION、DATETIME、JSON、および DECIMAL 型は、最初に STRING に変換され、次にバイトに変換されます。

    • CHAR、VARCHAR、VARSTRING、STRING、 TEXT、および BLOB 型 (TINY、MEDIUM、および LONG を含む) は、直接バイトに変換されます。

    • NULL および GEOMETRY 型はチェックサム計算から除外され、この関数は空のバイトを返します。

Golangを使用したデータ消費とチェックサム検証の実装の詳細については、 TiCDC 行データチェックサム検証を参照してください。

このページは役に立ちましたか？