操作シナリオ
データ購読Kafka版（現在のKafka ServerバージョンはV2.6.0）では、Protobuf形式の購読データに対して、Flinkクライアント（クライアントタイプはDataStream APIのみサポート）を使用して消費できます。本シナリオでは、consumer-demo-tdsql-pb-flink を使用してデータを消費するDemoサンプルを提供します。これにより、データ消費のプロセスを迅速にテストし、データ形式の解析方法を理解できます。
前提条件
3. Flink実行環境がインストール済みで、Flinkタスクを正常に実行できます。 注意事項
Demoにはデータ消費の使用方法のデモは含まれておらず、データの印刷処理のみが行われています。お客様はこれに基づいてデータ処理ロジックを自身で作成する必要があります。
現在、外部ネットワーク経由でデータ購読のKafkaに接続して消費することはサポートされておらず、Tencent Cloudの内部ネットワークからのアクセスのみがサポートされています。また、購読するデータベースインスタンスが属する地域とデータ消費の地域は同じである必要があります。
指定のデータベース/テーブルオブジェクト（ソースインスタンス全体ではない）を購読し、かつKafka単一パーティションを使用するシナリオでは、DTSが増分データを解析した後、購読対象のデータのみをKafka Topicに書き込みます。その他の購読対象外のデータは空のトランザクションに変換されてKafka Topicに書き込まれるため、データを消費する際に空のトランザクションが発生します。空のトランザクションのBegin/CommitメッセージにはトランザクションのGTID情報が保持されており、GTIDの連続性と完全性を保証できます。
データの再入可能性を保証するために、DTS購読はCheckpointメカニズムを導入しています。メッセージがKafka Topicに書き込まれる際、通常10秒ごとにCheckpointが1つ挿入され、データ同期の位置を識別するために使用されます。タスクが中断された後に再起動すると、中断位置を認識して再開できます。また、コンシューマー側はCheckpointメッセージを受信すると、Kafkaの消費位置のコミットを一度実行し、消費位置をタイムリーに更新します。
消費Demoのダウンロード
購読タスクの設定において、TDSQL MySQLがサポートする購読データ形式はProtobufです。Protobufはバイナリ形式を採用しており、消費効率がより高くなります。以下のDemoにはProtobufプロトコルファイルが既に含まれているため、別途ダウンロードする必要はありません。お客様がProtobufプロトコルファイルを自身でダウンロードすることを選択した場合は、データ構造が正しく互換性を持つように、Protobuf 3.Xバージョンを使用してコードを生成してください。 Java Flink Demoの操作手順
コンパイル環境：Mavenパッケージ管理ツール、JDK8。ユーザーはパッケージツールを自身で選択できます。以下ではMavenを例として説明します。
実行環境：Tencent Cloud CVM（購読インスタンスと同じ地域である必要があり、Kafkaサーバーの内部ネットワークアドレスにアクセスできます）、JRE8をインストールします。
操作手順：
1. consumer-demo-tdsql-pb-flink.zipをダウンロードして、そのファイルを解凍してください。
2. 解凍後のディレクトリに入ります。利便性のため、ディレクトリには既にpom.xmlファイルが配置されています。ユーザーはFlinkのバージョンを変更し、Flinkクラスターのバージョンとpom.xmlが依存するFlinkのバージョンが一致するようにする必要があります。
3. 以下のコード内の${flink.version}は、クラスターのFlinkバージョンと一致させる必要があります。
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-streaming-java_${scala.binary.version}</artifactId>
<version>${flink.version}</version>
<scope>provided</scope>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-connector-kafka_${scala.binary.version}</artifactId>
<version>${flink.version}</version>
<exclusions>
<exclusion>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka-clients</artifactId>
</exclusion>
</exclusions>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-clients_${scala.binary.version}</artifactId>
<version>${flink.version}</version>
</dependency>
4. pomファイルがあるディレクトリに入り、Mavenを使用してパッケージングするか、直接IDEAを使用してパッケージングします。
Mavenを使用してパッケージングします：mvn clean package。
5. FlinkクライアントタイプがDataStream APIのシナリオでは、Flinkクライアントコマンドを使用してジョブをFlinkクラスターに送信し、消費を開始します。
./bin/flink run consumer-demo-tdsql-pb-flink.jar --brokers xxx --topic xxx --group xxx --user xxx --password xxx --trans2sql
brokersはデータ購読Kafkaの内部ネットワークアクセスアドレスです。topicはデータ購読タスクの購読topicです。この2つは購読詳細ページで確認できます。group、user、passwordはそれぞれコンシューマグループの名前、アカウント、パスワードです。消費管理ページで確認できます。trans2sqlはSQL文に変換するかどうかを示します。javaコードでは、このパラメータを指定するとSQL文に変換され、指定しない場合は変換されません。
6. ソースデータベースでDML文を実行し、Flinkに送信されたジョブの消費状況を観察してください。
CREATE TABLE `flink_test` (
`id` varchar(32) CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci NOT NULL,
`parent_id` varchar(32) CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci DEFAULT NULL,
`user_id` bigint NOT NULL,
`type` int NOT NULL COMMENT '1:支出 2：收入',
`create_time` timestamp(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3),
PRIMARY KEY (`id`) USING BTREE
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_general_ci ROW_FORMAT=DYNAMIC
7. Flinkに送信されたジョブの消費状況を観察してください。
Task Managersで特定のtaskのログを確認します。
Task Managersで具体的なStdout情報を確認します。