How to synchronize data from RDS MySQL to DataHub - Data Transmission Service

Use Data Transmission Service (DTS) to stream incremental changes from an ApsaraDB RDS for MySQL instance to a DataHub project in real time. After synchronization is running, you can use Alibaba Cloud Realtime Compute for Apache Flink or StreamCompute to analyze the data.

Prerequisites

Before you begin, ensure that you have:

A DataHub project in the China (Hangzhou), China (Shanghai), China (Beijing), or China (Shenzhen) region. To create one, see Create a project.
Tables to synchronize with PRIMARY KEY or UNIQUE constraints.

Billing

Synchronization type	Fee
Schema synchronization and full data synchronization	Free of charge
Incremental data synchronization	Charged. For more information, see Billing overview.

Limitations

No historical data (CDC only): Initial full data synchronization is not supported. DTS synchronizes only changes that occur after the task starts.
Tables only: Only tables can be selected as objects to synchronize.
Avoid DDL during sync: DDL operations on synchronized tables while the task is running may cause synchronization to fail.

SQL operations that can be synchronized

Operation type	SQL statement
DML	INSERT, UPDATE, and DELETE
DDL	ADD COLUMN

Set up data synchronization

Step 1: Purchase a data synchronization instance

Purchase a data synchronization instance. For more information, see Purchase a DTS instance.

On the buy page, set Source Instance to MySQL, Destination Instance to DataHub, and Synchronization Topology to One-way Synchronization.

Step 2: Configure the task

Log on to the DTS console.
If you are redirected to the Data Management (DMS) console, click the icon in the to go to the previous version of the DTS console.
In the left-side navigation pane, click Data Synchronization.
At the top of the Synchronization Tasks page, select the region where the destination instance resides.
Find the data synchronization instance and click Configure Task in the Actions column.

Configure the source and destination instances.

Section	Parameter	Description
N/A	Synchronization task name	The task name that DTS automatically generates. Specify a descriptive name to make the task easy to identify. The name does not need to be unique.
Source instance details	Instance type	Select RDS Instance. Important For a self-managed MySQL database, prepare the required environment first. See Preparation overview.
	Instance region	The source region selected on the buy page. Read-only.
	Instance ID	The ID of the source ApsaraDB RDS instance.
	Database account	The account of the source database. Important For MySQL 5.5 or MySQL 5.6, this parameter is not required.
	Database password	The password of the account. Important For MySQL 5.5 or MySQL 5.6, this parameter is not required.
	Encryption	Select Non-encrypted or SSL-encrypted. If you select SSL-encrypted, enable SSL encryption on the RDS instance before configuring this task. See Configure the SSL encryption feature. Important This parameter is available only in the Chinese mainland and China (Hong Kong).
Destination instance details	Instance type	DataHub. Read-only.
	Instance region	The destination region selected on the buy page. Read-only.
	Project	The name of the destination DataHub project.

Click Set Whitelist and Next in the lower-right corner. DTS automatically adds its server CIDR blocks to the IP address whitelist of Alibaba Cloud database instances (such as ApsaraDB RDS for MySQL) or to the security group rules of Elastic Compute Service (ECS) instances hosting the database. If the self-managed database is hosted on multiple ECS instances, you must manually add the CIDR blocks of DTS servers to the security group rules of each ECS instance. For self-managed databases in a data center or hosted by a third-party cloud provider, manually add the DTS server CIDR blocks. See Add the CIDR blocks of DTS servers.
Warning
Adding DTS CIDR blocks to a whitelist or security group may introduce security risks. Take preventive measures such as strengthening password security, limiting exposed ports, auditing API calls, and reviewing whitelist rules regularly. For higher security, connect the database to DTS using Express Connect, VPN Gateway, or Smart Access Gateway.

Configure the synchronization policy and objects.

Parameter	Description
Initial synchronization	Select Initial Schema Synchronization. DTS synchronizes the schemas of the selected objects to the destination DataHub project.
Objects to synchronize	In the Available section, select tables and click to move them to the Selected section. To rename synchronized objects in the destination, use the object name mapping feature. See Rename an object to be synchronized.
Enable new naming rules for additional columns	Select Yes or No. Before configuring this parameter, check whether the additional column names conflict with existing columns in the destination topic. Otherwise, the data synchronization task may fail or data may be lost. See Naming rules for additional columns.
Rename databases and tables	Use the object name mapping feature to rename synchronized objects in the destination. See Object name mapping.
Replicate temporary tables when DMS performs DDL operations	Controls whether DTS synchronizes temporary tables generated by online DDL operations in DMS. Yes: DTS synchronizes temporary tables. Large online DDL operations may increase synchronization latency. No: DTS skips temporary tables and synchronizes only the original DDL from the source. Destination tables may be locked.
Retry time for failed connections	Default: 720 minutes (12 hours). If DTS reconnects within this period, synchronization resumes. Otherwise, the task fails. DTS charges apply during retry periods. Set this value based on your business needs, and release the instance if the source and destination instances are released.

(Optional) In the Selected section, hover over a table name and click Edit to configure a shard key for table partitioning.
Click Precheck in the lower-right corner.
DTS runs a precheck before starting the task. The task starts only after the precheck passes. If a precheck item fails, click the icon next to the failed item to view details. Fix the issue and run the precheck again, or ignore the failed item and rerun if no fix is needed.
After the Precheck Passed message appears, close the Precheck dialog box. The task starts automatically.
Wait for initial synchronization to complete. The task status changes to Synchronizing. Check the task status on the Synchronization Tasks page.

DataHub topic schema

When DTS synchronizes incremental data to a DataHub topic, it adds metadata columns to the topic.

In this example, id, name, and address are data fields. When the previous naming rules are used, DTS adds the dts_ prefix to all fields, including the original fields from the source. When the new naming rules are used, DTS does not add a prefix to the original source fields.

Topic定义

Previous column name	New column name	Type	Description
`dts_record_id`	`new_dts_sync_dts_record_id`	String	Unique ID of the incremental log entry. Auto-increments for each new entry. In disaster recovery scenarios, rollback may cause ID duplication. For UPDATE operations, both log entries (pre-update and post-update) share the same `dts_record_id`.
`dts_operation_flag`	`new_dts_sync_dts_operation_flag`	String	The operation type. Valid values: `I` (INSERT), `D` (DELETE), `U` (UPDATE), `F` (full data synchronization).
`dts_instance_id`	`new_dts_sync_dts_instance_id`	String	The server ID of the database.
`dts_db_name`	`new_dts_sync_dts_db_name`	String	The database name.
`dts_table_name`	`new_dts_sync_dts_table_name`	String	The table name.
`dts_utc_timestamp`	`new_dts_sync_dts_utc_timestamp`	String	The operation timestamp in UTC. This is also the timestamp of the log file.
`dts_before_flag`	`new_dts_sync_dts_before_flag`	String	Indicates whether the row values are pre-update values. Valid values: `Y`, `N`. See the before and after flag values table.
`dts_after_flag`	`new_dts_sync_dts_after_flag`	String	Indicates whether the row values are post-update values. Valid values: `Y`, `N`. See the before and after flag values table.

Before and after flag values

The dts_before_flag and dts_after_flag columns indicate whether a row in the topic represents the state before or after a data change.

Operation	Log entries generated	dts_before_flag	dts_after_flag	Row content
INSERT	1	`N`	`Y`	The newly inserted values.
DELETE	1	`Y`	`N`	The deleted values.
UPDATE (first entry)	2 (same `dts_record_id`)	`Y`	`N`	Pre-update values.
UPDATE (second entry)	2 (same `dts_record_id`)	`N`	`Y`	Post-update values.

INSERT operation

The log entry contains the newly inserted values. dts_before_flag is N and dts_after_flag is Y.

UPDATE operation

DTS generates two log entries with the same dts_record_id, dts_operation_flag, and dts_utc_timestamp values. The first entry captures the pre-update values (dts_before_flag=Y, dts_after_flag=N). The second entry captures the post-update values (dts_before_flag=N, dts_after_flag=Y).

UPDATE操作

DELETE operation

The log entry contains the deleted values. dts_before_flag is Y and dts_after_flag is N.

DELETE操作

What's next

After the synchronization task is running, use Alibaba Cloud Realtime Compute for Apache Flink to analyze the data in the DataHub project. For more information, see What is Alibaba Cloud Realtime Compute for Apache Flink?