- Release Notes and Announcements
- Product Introduction
- Purchase Guide
- Getting Started
- Access Management
- Preparations
- Data Migration
- Databases Supported by Data Migration
- Cross-Account TencentDB Instance Migration
- Migrating to MySQL
- Migrating to MariaDB
- Migrating to TDSQL-C for MySQL
- Migrating to TDSQL MySQL
- Migrating to PostgreSQL
- Migrating to MongoDB
- Migrating to SQL Server
- Migrating to TencentDB for Redis
- Offline Migration of MySQL Data
- Advanced Migration Operations
- Task Management
- Data Sync
- Databases Supported by Data Sync
- Cross-Account TencentDB Instance Sync
- Synchronize to PostgreSQL
- Sync into Tencent Cloud
- Sync from Tencent Cloud
- Advanced Sync Operations
- Task Management
- Task Status Description
- Viewing Task
- Starting Task
- Retrying Task (MySQL)
- Verifying Task
- Stopping Task
- Pausing Task
- Resuming Task
- Resetting Task
- Deleting Task
- Changing Configuration
- Transition from Pay-as-You-Go to Monthly Subscription
- Terminating/Returning Instance
- Renewing Task
- Deisolating Task
- Deleting Instance
- Viewing Log
- Deleting Task
- Data Subscription (Kafka Edition)
- Databases Supported by Data Subscription
- Creating Data Subscription Task
- Data Consumption
- Creating Consumer Group
- Managing Consumer Group
- Consuming MySQL Data
- Consuming MongoDB Data
- Consuming TDSQL for PostgreSQL Data
- Consuming Subscribed Data with Kafka Client (Protobuf)
- Consuming Subscribed Data with Kafka Client (Avro)
- Consuming Subscribed Data with Kafka Client (JSON)
- Consuming Subscribed Data with Flink Client (Avro)
- Consuming Data with Flink
- Advanced Subscription Operations
- Task Management
- Data Migration (OldDTS)
- Data Subscription (OldDTS)
- Fix for Verification Failure
- Check Item Overview
- Check Items of MySQL/MariaDB/Percona/TDSQL-C for MySQL/TDSQL for MySQL
- MongoDB Check Items
- PostgreSQL Check Items
- TDSQL for PostgreSQL
- Redis/Tendis
- SQL Server
- Database Connection Check
- Source Database Existence Check
- Target Database Existence Check
- Peripheral Check
- Migration Network Check
- Version Check
- Source Instance Permission Check
- Account Conflict Check
- Partial Database Parameter Check
- Source Instance Parameter Check
- Source Instance Type (Master or Replica)
- Parameter Configuration Conflict Check
- Target Database Content Conflict Check
- Target Database Space Check
- Target Database Load Check
- Local Disk Space Check
- Binlog Parameter Check
- Oplog Check
- Foreign Key Dependency Check
- View Check
- Advanced Object Check
- Incremental Migration Precondition Check
- Extension/Plugin Compatibility Check
- Source Database Balancer Check
- Source Database Node Role Check
- ShardKey Check
- Warning Item Check
- DLL Ring Sync Check for Single Database/Table Object
- DLL Sync Conflict Check for Single Database/Table Object in the Same Target
- Level-2 Subpartitioned Table Check
- Primary Key Check
- DDL Check for Tables to Be Migrated
- System Database Conflict Check
- Table Structure Check for Source and Target Instances
- InnoDB Table Check
- Migration Object Dependency Check
- Constraint Check
- Cutover Description
- Monitoring and Alarms
- Ops Management
- DTS-DBbridge
- Best Practices
- API Documentation
- History
- Introduction
- API Category
- Making API Requests
- (NewDTS) Data Migration APIs
- DescribeMigrationJobs
- DescribeMigrationDetail
- CreateMigrationService
- ModifyMigrationJob
- CreateMigrateCheckJob
- DescribeMigrationCheckJob
- SkipCheckItem
- StartMigrateJob
- StopMigrateJob
- ResumeMigrateJob
- RecoverMigrateJob
- ModifyMigrateName
- ModifyMigrateJobSpec
- IsolateMigrateJob
- DestroyMigrateJob
- DescribeMigrateDBInstances
- CompleteMigrateJob
- PauseMigrateJob
- ContinueMigrateJob
- Data Sync APIs
- ModifySyncJobConfig
- CreateModifyCheckSyncJob
- DescribeModifyCheckSyncJobResult
- StartModifySyncJob
- DescribeSyncJobs
- CreateSyncJob
- ConfigureSyncJob
- CreateCheckSyncJob
- SkipSyncCheckItem
- DescribeCheckSyncJobResult
- StartSyncJob
- StopSyncJob
- ResumeSyncJob
- ResizeSyncJob
- RecoverSyncJob
- IsolateSyncJob
- DestroySyncJob
- PauseSyncJob
- ContinueSyncJob
- Data Consistency Check APIs
- Data Types
- Error Codes
- DTS API 2018-03-30
- SDK Documentation
- FAQs
- Error Handling (NewDTS)
- Error Handling (OldDTS)
- Service Agreement
- Release Notes and Announcements
- Product Introduction
- Purchase Guide
- Getting Started
- Access Management
- Preparations
- Data Migration
- Databases Supported by Data Migration
- Cross-Account TencentDB Instance Migration
- Migrating to MySQL
- Migrating to MariaDB
- Migrating to TDSQL-C for MySQL
- Migrating to TDSQL MySQL
- Migrating to PostgreSQL
- Migrating to MongoDB
- Migrating to SQL Server
- Migrating to TencentDB for Redis
- Offline Migration of MySQL Data
- Advanced Migration Operations
- Task Management
- Data Sync
- Databases Supported by Data Sync
- Cross-Account TencentDB Instance Sync
- Synchronize to PostgreSQL
- Sync into Tencent Cloud
- Sync from Tencent Cloud
- Advanced Sync Operations
- Task Management
- Task Status Description
- Viewing Task
- Starting Task
- Retrying Task (MySQL)
- Verifying Task
- Stopping Task
- Pausing Task
- Resuming Task
- Resetting Task
- Deleting Task
- Changing Configuration
- Transition from Pay-as-You-Go to Monthly Subscription
- Terminating/Returning Instance
- Renewing Task
- Deisolating Task
- Deleting Instance
- Viewing Log
- Deleting Task
- Data Subscription (Kafka Edition)
- Databases Supported by Data Subscription
- Creating Data Subscription Task
- Data Consumption
- Creating Consumer Group
- Managing Consumer Group
- Consuming MySQL Data
- Consuming MongoDB Data
- Consuming TDSQL for PostgreSQL Data
- Consuming Subscribed Data with Kafka Client (Protobuf)
- Consuming Subscribed Data with Kafka Client (Avro)
- Consuming Subscribed Data with Kafka Client (JSON)
- Consuming Subscribed Data with Flink Client (Avro)
- Consuming Data with Flink
- Advanced Subscription Operations
- Task Management
- Data Migration (OldDTS)
- Data Subscription (OldDTS)
- Fix for Verification Failure
- Check Item Overview
- Check Items of MySQL/MariaDB/Percona/TDSQL-C for MySQL/TDSQL for MySQL
- MongoDB Check Items
- PostgreSQL Check Items
- TDSQL for PostgreSQL
- Redis/Tendis
- SQL Server
- Database Connection Check
- Source Database Existence Check
- Target Database Existence Check
- Peripheral Check
- Migration Network Check
- Version Check
- Source Instance Permission Check
- Account Conflict Check
- Partial Database Parameter Check
- Source Instance Parameter Check
- Source Instance Type (Master or Replica)
- Parameter Configuration Conflict Check
- Target Database Content Conflict Check
- Target Database Space Check
- Target Database Load Check
- Local Disk Space Check
- Binlog Parameter Check
- Oplog Check
- Foreign Key Dependency Check
- View Check
- Advanced Object Check
- Incremental Migration Precondition Check
- Extension/Plugin Compatibility Check
- Source Database Balancer Check
- Source Database Node Role Check
- ShardKey Check
- Warning Item Check
- DLL Ring Sync Check for Single Database/Table Object
- DLL Sync Conflict Check for Single Database/Table Object in the Same Target
- Level-2 Subpartitioned Table Check
- Primary Key Check
- DDL Check for Tables to Be Migrated
- System Database Conflict Check
- Table Structure Check for Source and Target Instances
- InnoDB Table Check
- Migration Object Dependency Check
- Constraint Check
- Cutover Description
- Monitoring and Alarms
- Ops Management
- DTS-DBbridge
- Best Practices
- API Documentation
- History
- Introduction
- API Category
- Making API Requests
- (NewDTS) Data Migration APIs
- DescribeMigrationJobs
- DescribeMigrationDetail
- CreateMigrationService
- ModifyMigrationJob
- CreateMigrateCheckJob
- DescribeMigrationCheckJob
- SkipCheckItem
- StartMigrateJob
- StopMigrateJob
- ResumeMigrateJob
- RecoverMigrateJob
- ModifyMigrateName
- ModifyMigrateJobSpec
- IsolateMigrateJob
- DestroyMigrateJob
- DescribeMigrateDBInstances
- CompleteMigrateJob
- PauseMigrateJob
- ContinueMigrateJob
- Data Sync APIs
- ModifySyncJobConfig
- CreateModifyCheckSyncJob
- DescribeModifyCheckSyncJobResult
- StartModifySyncJob
- DescribeSyncJobs
- CreateSyncJob
- ConfigureSyncJob
- CreateCheckSyncJob
- SkipSyncCheckItem
- DescribeCheckSyncJobResult
- StartSyncJob
- StopSyncJob
- ResumeSyncJob
- ResizeSyncJob
- RecoverSyncJob
- IsolateSyncJob
- DestroySyncJob
- PauseSyncJob
- ContinueSyncJob
- Data Consistency Check APIs
- Data Types
- Error Codes
- DTS API 2018-03-30
- SDK Documentation
- FAQs
- Error Handling (NewDTS)
- Error Handling (OldDTS)
- Service Agreement
Overview
In data subscription (Kafka Edition, where the current Kafka Server version is 2.6.0), subscribed data in Protobuf format can be consumed by using a Flink client (only the DataStream API type). This document provides a demo for data consumption with consumer-demo-tdsql-pb-flink for you to quickly test the process of data consumption and understand the method of data parsing.
Prerequisites
1. You have created a data consumption task as instructed in Creating Data Subscription Task.
2. You have created a consumer group as instructed in Creating Consumer Group.
3. You have installed Flink, and it can execute tasks normally.
Note
The demo only prints out the consumed data and contains no instructions. You need to write your own data processing logic based on the demo.
Currently, data subscription to Kafka for consumption can only be implemented over the Tencent Cloud private network but not the public network. In addition, the subscribed database instance and the data consumer must be in the same region.
In scenarios where only the specified databases and tables (part of the source instance objects) are subscribed and single-partition Kafka topics are used, only the data of the subscribed objects will be written to Kafka topics after DTS parses the incremental data. The data of non-subscription objects will be converted into empty transactions and then written to Kafka topics. Therefore, there are empty transactions during message consumption. The BEGIN and COMMIT messages in the empty transactions contain the GTID information, which can ensure the GTID continuity and integrity.
To ensure that data can be rewritten from where the task is paused, DTS adopts the checkpoint mechanism for data subscription. Specifically, when messages are written to Kafka topics, a checkpoint message is inserted every 10 seconds to mark the data sync offset. When the task is resumed after being interrupted, data can be rewritten from the checkpointed offeset. The consumer commits a consumption offset every time it encounters a checkpoint message so that the consumption offset can be updated timely.
Downloading Consumption Demo
For the description of the logic and key parameters in the demo, see Protobuf Demo Description (Flink).
When configuring the subscription task, you can only select Protobuf as the format of the subscribed data for TDSQL for MySQL. As Protobuf adopts the binary format, the consumption efficiency is high. The following demo already contains the Protobuf protocol file, so you don't need to download it separately. If you download it on your own, use Protobuf 3.X for code generation to ensure that data structures are compatible.
Demo Language | Protobuf (TDSQL for MySQL) |
Java |
Directions for the Java Flink Demo
Compiling environment: Maven and JDK8. You can choose a desired package management tool. The following takes Maven as an example.
Runtime environment: Tencent Cloud CVM (which can access the private network address of the Kafka server only if it is in the same region as the subscribed instance). Install JRE8.
Directions:
1. Download the consumer-demo-tdsql-pb-flink.zip file and unzip it.
2. Access the directory of the unzipped file. The
pom.xml file has been placed under the directory for your convenience. You need to modify the Flink cluster version to the same as that specified in the pom.xml dependency.3. The value of
${flink.version} in the code below must be the same as the Flink cluster version.<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. Go to the directory where the
pom.xml file is located and package it with Maven or IEDA.
Package with Maven by running mvn clean package.5. If the Flink client type is DataStream API, use Flink client commands to submit the job to the Flink cluster and start consumption.
./bin/flink run consumer-demo-tdsql-pb-flink.jar --brokers xxx --topic xxx --group xxx --user xxx --password xxx --trans2sqlbrokers is the private network access address for data subscription to Kafka, and topic is the subscription topic, which can be viewed on the Subscription details page as instructed in Viewing Subscription Details.group, user, and password are the name, account, and password of the consumer group, which can be viewed on the Consumption Management page as instructed in Managing Consumer Group.trans2sql indicates whether to enable conversion to SQL statement. In Java code, if this parameter is carried, the conversion will be enabled.6. Execute DML statements in the source database.
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: Expenditure 2: Income',`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. Observe the consumption performed by the previously submitted job.
View the specific task logs on task managers.
View the specific Stdout information on task managers.
Yes
No
Was this page helpful?