MySQLシリーズサブスクリプションデータ操作の消費ガイド
ダウンロード
フォーカスモード
フォントサイズ
操作シナリオ
本操作は、MySQL/TDSQL-C MySQL/MariaDB/Percona/TDSQL MySQLのデータ購読、データ購読Kafka版（現在のKafka ServerバージョンはV2.6.0）に適用されます。0.11バージョン以上のKafka クライアントを使用して購読データを消費できます。本ドキュメントでは、Java、Go、Python言語のクライアント消費Demo例を提供し、データ消費プロセスを迅速にテストし、データ形式の解析方法を理解できるようにします。
注意事項
1. Demo にはデータ消費の使用方法のデモは含まれておらず、データの印刷処理のみが行われています。お客様はこれに基づいてデータ処理ロジックを自身で作成する必要があります。他の言語の Kafka クライアントを使用してデータを消費し、解析することもできます。
2. 現在、外部ネットワーク経由でデータ購読のKafkaに接続して消費することはサポートされておらず、Tencent Cloudの内部ネットワークからのアクセスのみがサポートされています。また、購読するデータベースインスタンスが属する地域とデータを消費する地域は同じである必要があります。
3. データ形式としてAvro、JSONを選択した場合、DTS購読における大きなメッセージの処理には以下の制限があります。
2024年3月25日より前に作成されたタスクでは、DTS購読の単一メッセージ処理には一定の上限があります。ソースデータベース内の単一行データが5MBを超えると、購読タスクでエラーが発生する可能性があります。
2024年3月25日以降に作成されたタスクでは、ソースデータベース内の単一行データが5MBを超えた場合でも、購読タスクでエラーは発生しません。最新の消費Demoを使用することで、大きなメッセージの消費データを取得できます。
4. 指定のデータベース/テーブルオブジェクト（ソースインスタンス全体ではない）を購読し、かつKafkaの単一パーティションを使用するシナリオでは、DTSは増分データを解析した後、購読対象のデータのみをKafka Topicに書き込みます。その他の購読対象外のデータは空のトランザクションに変換されてKafka Topicに書き込まれるため、データを消費する際に空のトランザクションが発生します。空のトランザクションのBegin/CommitメッセージにはトランザクションのGTID情報が保持されており、GTIDの連続性と完全性を保証できます。また、MySQL/TDSQL-C MySQLの消費Demoでは、複数の空のトランザクションも圧縮処理され、メッセージ数が削減されます。
5. データの再入可能性を保証するために、DTS購読はCheckpointメカニズムを導入しています。メッセージがKafka Topicに書き込まれる際、通常10秒ごとにCheckpointが1つ挿入され、データ同期の位置を識別するために使用されます。タスクが中断された後に再起動すると、中断位置を認識してチェックポイントからの再開を実現します。また、コンシューマー側はCheckpointメッセージを受信すると、Kafkaの消費位置のコミットを一度実行し、消費位置をタイムリーに更新します。
6. データ形式としてJSONを選択した場合、オープンソースの購読ツールCanalを使用したことがある、または熟知している場合は、ここで消費されたJSON形式のデータをCanalツール互換のデータ形式に変換してから後続の処理を行うことができます。私たちのDemoではすでに関連サポートが提供されており、Demoを起動する際のパラメータにtrans2canalを追加するだけで実現できます。現在、この機能はJava言語のみでサポートされています。
7. ProtoBufおよびJSONデータ形式を選択した場合、ソース側MySQLの数値型について、DTSがKafkaに書き込む際に文字列型に変換されます。そのうち、FLOATおよびDOUBLE型は文字列に変換される際、絶対値が1e-4未満または1e+6以上の場合、結果は科学的記数法で表されます。科学的記数法は、仮数に10の累乗を掛けた形式で浮動小数点数を表すため、その桁数と正確な値をより明確に示すことができ、同時にストレージ容量を節約できます。コンシューマー側で数値型に変換する必要がある場合は、科学的記数法で表された浮動小数点数の文字列に対して特別な処理を行うように注意してください。
8. 購読タスクのデータ形式としてJSONを選択した場合、ソースデータベースのMySQLパラメータ
binlog_row_imageがminimalに設定されていると、minimalの実装は、影響のある列のみをbinlogに記録し、影響のない列は記録しないため、消費結果のUPDATE文では未更新フィールドの値が欠落します。DTSはNULL値と区別するために、##DTS_NA_VALUE##を使用して未更新フィールドの値を表します。ユーザーが消費結果のUPDATE文に未更新フィールドの値を含めることを希望する場合は、ソース側MySQLパラメータ
binlog_row_imageをfullに設定してから、購読タスクを再作成する必要があります。ProtoBufとAvro形式はバイナリ形式であるため、特殊な文字列を使用して欠落フィールド値とNULL値を区別する必要がなく、この問題は発生しません。
9. DTSのバックエンド巡回検査は、コンシューマグループを作成してユーザーKafkaのオフセット情報を消費し、消費が正常に進行しているかどうかをチェックします。
Demo ダウンロード
購読タスクの設定では、ProtoBuf、Avro、JSONの異なる購読データ形式を選択できます。ProtoBufとAvroはバイナリ形式を採用しており、消費効率がより高くなります。JSONは軽量なテキスト形式を採用しており、より簡単に使用できます。購読タスクで選択した形式に応じて、対応する形式のDemoをダウンロードする必要があります。
以下のDemo例には、対応するProtobuf/Avro/JSONプロトコルファイルがすでに含まれているため、別途ダウンロードする必要はありません。もしProtobufプロトコルファイルを自分でダウンロードすることを選択した場合は、データ構造が正しく互換性を持つように、Protobuf 3.Xバージョンを使用してコードを生成してください。
Demoにおけるロジックの説明および主要パラメータの説明については、Demo説明を参照してください。
Java Demo 操作手順
コンパイル環境：MavenまたはGradleパッケージ管理ツール、JDK8。
実行環境：Tencent Cloud CVM（購読インスタンスと同じ地域である必要があり、Kafkaサーバーの内部ネットワークアドレスにアクセスできます）、JRE8をインストールします。
操作手順は以下の通りです：
1. 新版データ購読タスクを作成します。詳細についてはデータ購読Kafka版を参照してください。
2. 1つまたは複数のコンシューマグループを作成します。詳細についてはコンシューマグループの追加を参照してください。
3. Java Demoをダウンロードしてから、そのファイルを解凍してください。
4. 解凍後のディレクトリに入ります。利用しやすいように、ディレクトリにはMavenモデルファイル、pom.xmlファイル、Gradle関連設定ファイルがそれぞれ配置されており、ユーザーは必要に応じて選択できます。
Mavenを使用してパッケージ化します：mvn clean package。
Gradleを使用してパッケージ化します：gradle fatJarでパッケージ化してすべての依存関係を含めるか、またはgradle jarでパッケージ化します。
5. Demoを実行します。
Mavenでパッケージ化した後、ターゲットフォルダtargetに入り、
java -jar sub_demo-1.0-SNAPSHOT-jar-with-dependencies.jar --brokers=xxx --topic=xxx --group=xxx --user=xxx --password=xxx --trans2sqlを実行します。Gradleでパッケージ化した後、フォルダbuild/libsに入り、
java -jar sub_demo-with-dependencies-1.0-SNAPSHOT.jar --brokers=xxx --topic=xxx --group=xxx --user=xxx --password=xxx --trans2sqlを実行します。各パラメータの詳細な説明は以下の通りです：
パラメータ | 説明 |
brokers | |
topic | |
group | |
user/password | |
trans2sql | SQL文に変換するかどうかを示します。このパラメータを指定するとSQL文に変換され、指定しない場合は変換されません。 |
trans2canal | オプションです。JSONデータ形式のみをサポートし、Java言語に限定されます。 データ形式としてJSONを選択した場合、JSON形式のデータをCanalツール互換のデータ形式に変換できます。このパラメータを指定するとCanal形式に変換され、指定しない場合は変換されません。 |
6. 消費状況を観察します。
ユーザーはIDEを使用してコンパイルとパッケージ化を行うこともできます。パッケージ化が完了すると、プロジェクトルートディレクトリのtargetフォルダにあるsub_demo-1.0-SNAPSHOT-jar-with-dependenciesが、必要な依存関係を含む実行可能なjarパッケージとなります。
Golang Demo 操作手順
コンパイル環境：Golang 1.12以降のバージョン、Go Module環境を設定します。
実行環境：Tencent Cloud CVM（サブスクリプションインスタンスと同じ地域にある必要があり、Kafkaサーバーの内部ネットワークアドレスにアクセスできます）。
操作手順は以下の通りです：
1. 新版データ購読タスクを作成します。詳細についてはデータ購読Kafka版を参照してください。
2. 1つまたは複数のコンシューマグループを作成します。詳細についてはコンシューマグループの追加を参照してください。
3. Golang Demoをダウンロードしてから、そのファイルを解凍してください。
4. 解凍後のディレクトリに入り、
go build -o subscribe ./mainを実行して、実行可能ファイルsubscribeを生成します。5. 以下のコードを実行します。
/subscribe --brokers=xxx --topic=xxx --group=xxx --user=xxx --password=xxx --trans2sql=true
ここで、brokersはデータ購読Kafkaの内部ネットワークアクセスアドレス、topicはデータ購読タスクの購読topicです。これらは購読詳細ページで確認できます。group、user、passwordはそれぞれコンシューマグループの名前、アカウント、パスワードであり、消費管理ページで確認できます。trans2sqlはSQL文に変換するかどうかを示します。6. 消費状況を観察します。
Python3 Demo 操作手順
コンパイル実行環境：Tencent Cloud CVM（購読インスタンスと同じ地域にある必要があり、Kafkaサーバーの内部ネットワークアドレスにアクセスできます）、Python3、pip3（依存パッケージのインストールに使用）をインストールします。
pip3を使用して依存パッケージをインストールします：
pip install flagpip install kafka-pythonpip install protobuf
操作手順は次のとおりです。
1. 新版データ購読タスクを作成します。詳細についてはデータ購読Kafka版を参照してください。
2. 1つまたは複数のコンシューマグループを作成します。詳細についてはコンシューマグループの追加を参照してください。
3. Python3 Demoをダウンロードしてから、そのファイルを解凍してください。
4. 以下のコードを実行します：
python main.py --brokers=xxx --topic=xxx --group=xxx --user=xxx --password=xxx --trans2sql=1
ここで、brokersはデータ購読Kafkaの内部ネットワークアクセスアドレス、topicはデータ購読タスクの購読topicです。これらは購読詳細ページで確認できます。group、user、passwordはそれぞれコンシューマグループの名前、アカウント、パスワードであり、消費管理ページで確認できます。trans2sqlはSQL文に変換するかどうかを示します。5. 消費状況を観察します。
フィードバック