Migrate MySQL data to SelectDB - ApsaraDB for SelectDB - Alibaba Cloud Documentation Center

You can use DTS, Flink CDC, a catalog, or DataWorks to migrate data from MySQL sources (such as self-managed MySQL, ApsaraDB RDS for MySQL, and PolarDB for MySQL) to ApsaraDB for SelectDB. Choose a method based on the data volume and your use case.

Comparison of migration methods

DTS, DataWorks, Flink CDC, and a catalog can all migrate MySQL data to ApsaraDB for SelectDB, but their capabilities differ. Choose the method that best fits your scenario.

Method	Historical data migration	Incremental data synchronization	Schema migration	Database migration	DDL synchronization	Data verification
DTS	✔️	✔️	✔️	✔️	✔️	✔️
DataWorks	✔️	✔️	✔️	✔️	✔️	❌
Flink CDC	✔️	✔️	✔️	✔️	✔️	❌
catalog	✔️	❌	❌	❌	❌	❌

The following sections outline the basic steps for each method. For details, refer to each method's documentation.

Prerequisites

Ensure that the MySQL instance and the SelectDB instance can communicate over the network.
- The MySQL instance and the SelectDB instance are in the same VPC. If not, resolve the network connectivity issue first. For instructions, see How do I resolve network connectivity issues between an ApsaraDB for SelectDB instance and a data source?
- Add the IP address of the MySQL instance to the SelectDB IP address whitelist. For more information, see Configure an IP address whitelist.
- If your MySQL instance has an IP address whitelist, add the IP CIDR block of the SelectDB instance's VPC to the MySQL IP address whitelist.
  - To get the IP range of the VPC where your SelectDB instance resides, see How do I find the IP CIDR block of the VPC where my ApsaraDB for SelectDB instance is located?
  - To get the public IP address of your SelectDB instance, use the ping command on its public endpointSelectDB.

Migrate data with DTS

DTS supports historical data migration and incremental data synchronization from MySQL to ApsaraDB for SelectDB, and provides capabilities such as database and table migration, DDL synchronization, and data verification. The following example uses an ApsaraDB RDS for MySQL instance. For more information, see Use DTS to import data.

Procedure

Log on to the ApsaraDB for SelectDB console.
In the upper-left corner, select the region where your instance resides.
On the Instances page, click the target Instance ID to go to the Instance Details page.
In the left-side navigation pane, click Data Pipeline, and then select the Data Synchronization tab.
Note
DTS data synchronization combines historical data migration for one-time transfers with incremental data synchronization for real-time updates.
Click Create Synchronization Task and configure the source and destination databases.

Category	Configuration	Description
Task Information	Task Name	DTS automatically generates a task name. We recommend that you specify a descriptive name for easy identification. The name does not need to be unique.
Source Database	Database Type	Select MySQL.
	Access Method	Select Alibaba Cloud Instance.
	Instance Region	Select the region where the source ApsaraDB RDS for MySQL instance is located.
	RDS Instance ID	Select the ID of the source ApsaraDB RDS for MySQL instance. Example: rm-2z3m****.
	Database Account	Enter the database account of the source ApsaraDB RDS for MySQL instance. For the required permissions, see Permissions required for database accounts.
	Database Password	Enter the password for the specified database account.
	Encryption	Select Non-encrypted or SSL-encrypted as needed. If you set this to SSL-encrypted, you must enable SSL encryption for the RDS for MySQL instance beforehand. For more information, see Use a cloud certificate to quickly enable SSL link encryption.
Destination Database	Database Type	Select SelectDB.
	Access Method	Select Alibaba Cloud Instance.
	Instance Region	Select the region where the destination SelectDB instance resides.
	Instance ID	Select the ID of the destination SelectDB instance.
	Database Account	Enter the database account of the destination SelectDB instance. For the required permissions, see Permissions required for database accounts.
	Database Password	Enter the password for the specified database account.

After you complete the configuration, click Test Connection and Proceed at the bottom of the page.
Configure the task objects and advanced settings.

Configuration	Description
Synchronization Types	DTS always selects Incremental Data Synchronization. By default, you must also select Schema Synchronization and Full Data Synchronization. After the precheck, DTS initializes the destination cluster with the full data of the selected source objects, which serves as the baseline for subsequent incremental synchronization. Important Data types are converted when data is synchronized from MySQL to SelectDB. If you do not select Schema Synchronization, you must first create a table with a corresponding schema that uses the Unique Key model in the destination SelectDB instance. For more information, see Data type mappings and Data model.
Source Objects	In the Source Objects box, click the objects that you want to synchronize, and then click to move the objects to the Selected Objects box. You can select objects at the database, table, or column level. Important If you select objects at the database level, DTS does not synchronize data for tables that are created after the task starts. If you plan to add tables during the synchronization, select objects at the table level. You can later change the selection by using Modify Synchronized Objects. If you select objects at the table level and need to edit them, such as by mapping table or column names, a single synchronization task supports a maximum of 1,000 tables. If this limit is exceeded, an error is reported when you submit the task. In this case, split the tables to be synchronized into batches and configure multiple tasks, or configure a database-level synchronization task.
Selected Objects	To rename a single object in the destination instance, right-click the object in the Selected Objects box. For more information, see Map a single object name. To rename multiple objects in bulk, click Batch Edit in the upper-right corner of the Selected Objects box. For more information, see Map multiple object names in bulk. Note To select the SQL operations to be synchronized at the database or table level, right-click the object to be synchronized in the Selected Objects box, and then select the desired SQL operations in the dialog box that appears. For a list of supported operations, see SQL operations that support incremental synchronization. To filter data by using a WHERE clause, right-click the table to be synchronized in the Selected Objects box, and then set the filter condition in the dialog box that appears. For more information, see Set filter conditions. If you use the object name mapping feature, other objects that depend on the remapped object may fail to be synchronized.

Optional: After you complete the preceding configurations, click Next: Configure Database and Table Fields to set the Primary Key Column, Distribution Key, and Engine for the destination tables.

Note

This step is available only when you select Synchronization Types for Schema Synchronization when you configure task objects. You can set Definition Status to All and then modify the settings.
You can select multiple columns to form a composite Primary Key Column, and you must select one or more columns from the Primary Key Column as the Distribution Key. For the Engine, only Unique is supported.

Save the task and run a precheck. After the Precheck Success Rate reaches 100%, click Next: Purchase.
On the Purchase page, select the billing method and link specification for the data synchronization instance. Read and select Data Transmission Service (Pay-As-You-Go) Service Terms. Click Purchase and Start to start the synchronization task. You can view the task progress on the data synchronization page.

Migrate data with DataWorks

DataWorks Data Integration can synchronize data from MySQL to ApsaraDB for SelectDB. For more information, see Import data by using DataWorks.

Important

Writing to fields of the BITMAP, HLL (HyperLogLog), or QUANTILE_STATE data type is not supported.

Add data sources

When developing a data synchronization task, create a MySQL data source and a SelectDB data source in DataWorks.

Create a MySQL data source. For more information, see MySQL data source.

Create a SelectDB data source. For more information, see Data source management. The following table describes some of the configuration parameters for the SelectDB data source.

Parameter	Description
Data Source Name	The name of the data source.
MySQL Connection URL	Enter the JDBC connection string `jdbc:mysql://<ip>:<port>/<dbname>`. You can obtain the VPC Endpoint (or Public Endpoint) and MySQL Port from the Instance Details > Network Information page in the ApsaraDB for SelectDB console. Example: `jdbc:mysql://selectdb-cn-4xl3jv1****.selectdbfe.rds.aliyuncs.com:9030/test_db`
HTTP Connection URL	Enter the HTTP protocol endpoint in the format `<ip>:<port>`. You can obtain the VPC Endpoint (or Public Endpoint) and HTTP Port from the Instance Details > Network Information page in the ApsaraDB for SelectDB console. Example: `selectdb-cn-4xl3jv1****.selectdbfe.rds.aliyuncs.com:8080`
Username	Enter the username for the ApsaraDB for SelectDB instance.
Password	Enter the password for the specified user of the ApsaraDB for SelectDB instance.

Configure data synchronization

To configure a data migration task, see the following documents:

Migrate data with Flink CDC

Flink provides three methods to migrate data from MySQL to SelectDB: Flink SQL, Flink CDC, and DataStream. Flink CDC supports historical data migration, incremental data synchronization, database and table migration, and DDL synchronization. The following example uses Flink CDC to synchronize data from MySQL to ApsaraDB for SelectDB. For more information, see Use Flink to import data.

Procedure

Prepare the environment

Set up a Flink environment. This example uses a Flink 1.16 standalone environment.

Download and decompress flink-1.16.3-bin-scala_2.12.tgz. If this version is unavailable, download another version. For more versions, see Apache Flink.
```
wget https://archive.apache.org/dist/flink/flink-1.16.3/flink-1.16.3-bin-scala_2.12.tgz
tar -zxvf flink-1.16.3-bin-scala_2.12.tgz
```

Go to the FLINK_HOME/lib directory and download flink-sql-connector-mysql-cdc-2.4.2 and flink-doris-connector-1.16-1.5.2.

cd flink-1.16.3
cd lib/
wget https://repo1.maven.org/maven2/com/ververica/flink-sql-connector-mysql-cdc/2.4.2/flink-sql-connector-mysql-cdc-2.4.2.jar
wget https://repo.maven.apache.org/maven2/org/apache/doris/flink-doris-connector-1.16/1.5.2/flink-doris-connector-1.16-1.5.2.jar

Start the Flink Standalone cluster.
```
bin/start-cluster.sh
```
Create an ApsaraDB for SelectDB instance. For more information, see Create an instance.
Connect to the ApsaraDB for SelectDB instance over the MySQL protocol. For more information, see Connect to an instance.

Create a test database and a test table.

Create a test database.
```
CREATE DATABASE test_db;
```

Create a test table.

USE test_db;
CREATE TABLE employees (
    emp_no       int NOT NULL,
    birth_date   date,
    first_name   varchar(20),
    last_name    varchar(20),
    gender       char(2),
    hire_date    date
)
UNIQUE KEY(`emp_no`)
DISTRIBUTED BY HASH(`emp_no`) BUCKETS 1;

Submit a Flink CDC job

Use the following syntax to submit a Flink CDC job:

<FLINK_HOME>/bin/flink run \
    -Dexecution.checkpointing.interval=10s \
    -Dparallelism.default=1 \
    -c org.apache.doris.flink.tools.cdc.CdcTools \
    lib/flink-doris-connector-1.16-1.5.2.jar \
    mysql-sync-database \
    --database test_db \
    --including-tables "tbl1|test.*" \
    --mysql-conf hostname=127.0.0.1 \
    --mysql-conf username=root \
    --mysql-conf password=123456 \
    --mysql-conf database-name=mysql_db \
    --sink-conf fenodes=selectdb-cn-****.selectdbfe.rds.aliyuncs.com:8080 \
    --sink-conf username=admin \
    --sink-conf password=****

Parameter	Description
execution.checkpointing.interval	The Flink checkpoint interval, which affects the data synchronization frequency. A value of 10s is recommended.
parallelism.default	The parallelism of the Flink job. You can increase the parallelism to improve data synchronization throughput.
database	The name of the destination database in SelectDB.
including-tables	The MySQL tables to be synchronized. You can use a vertical bar (\|) to separate multiple tables. Regular expressions are supported. For example, `--including-tables table1\|tbl.*` specifies that table1 and all tables whose names start with tbl are synchronized.
excluding-tables	The tables that you do not want to synchronize. The configuration method is the same as that for the including-tables parameter.
mysql-conf	The MySQL CDC source configuration. For more information, see MySQL CDC Connector. The `hostname`, `username`, `password`, and `database-name` parameters are required.
sink-conf	All configurations for the Doris sink. For more information, see Use Flink to import data.
table-conf	The configuration items for a SelectDB table. These correspond to the content within the PROPERTIES clause of a CREATE TABLE statement.

Note

To synchronize data, add the required Flink CDC dependencies, such as flink-sql-connector-mysql-cdc-${version}.jar and flink-sql-connector-oracle-cdc-${version}.jar, to the $FLINK_HOME/lib directory.
Database-level synchronization requires Flink 1.15 or later. For Flink Doris Connector downloads, see Flink Doris Connector.

Migrate data with a catalog

The catalog feature in SelectDB uses federated queries to access MySQL, simplifying historical data migration. The following example synchronizes data from a MySQL database to ApsaraDB for SelectDB. For more information, see JDBC data source.

Procedure

Connect to the SelectDB instance. For more information, see Connect to an instance.
Note
The SWITCH statement does not work when you connect by using DMS. We recommend connecting with a MySQL client.

Create a MySQL JDBC catalog.

CREATE CATALOG jdbc_mysql PROPERTIES (
    "type"="jdbc",
    "user"="root",
    "password"="123456",
    "jdbc_url" = "jdbc:mysql://127.0.0.1:3306/demo",
    "driver_url" = "mysql-connector-java-8.0.25.jar",
    "driver_class" = "com.mysql.cj.jdbc.Driver",
    "checksum" = "fdf55dcef04b09f2eaf42b75e61ccc9a"
)

Parameter description

Parameter	Required	Default	Description
user	Yes	None	The database account.
password	Yes	None	The password for the database account.
jdbc_url	Yes	None	The JDBC connection string.
driver_url	Yes	None	The file name of the JDBC driver JAR package.
driver_class	Yes	None	The class name of the JDBC driver.
lower_case_table_names	No	"false"	Specifies whether to synchronize the database and table names from the external JDBC data source in lowercase.
only_specified_database	No	"false"	Specifies whether to synchronize only the specified databases.
include_database_list	No	""	When `only_specified_database=true`, specifies a comma-separated list of databases to synchronize. Database names are case-sensitive.
exclude_database_list	No	""	When `only_specified_database=true`, specifies a comma-separated list of databases to exclude from synchronization. Database names are case-sensitive.

Create the destination table in SelectDB, and then use the INSERT INTO ... SELECT statement to synchronize data. For more information about INSERT INTO, see INSERT INTO.

# Create the destination table.
CREATE TABLE selectdb_table ...

# Migrate data.
INSERT INTO selectdb_table SELECT * FROM mysql_catalog.mysql_database.mysql_table;