ApsaraDB for SelectDB:Migrate data from a MySQL data source

Create a data synchronization task

1. Log on to the ApsaraDB for SelectDB console.

2. In the top navigation bar, select the region where your instance resides.

3. In the left-side navigation pane, click Instances. On the Instances page, click the instance ID to go to the Instance Details page.

4. In the left-side navigation pane of the Instance Details page, click Data Pipeline. On the Data Synchronization tab, click Create Data Synchronization Task.

   Note

   In DTS, data synchronization covers both existing data migration and real-time incremental synchronization. Data migration in DTS is for existing historical data only.

5. Configure the source and destination databases.

   Section	Parameter	Description
   Task information	Task Name	The name of the DTS task. DTS generates a name automatically. Specify a descriptive name to identify the task easily. The name does not need to be unique.
   Source Database	Database Type	Select MySQL.
   	Access Method	Select Alibaba Cloud Instance.
   	Instance Region	The region where the source ApsaraDB RDS for MySQL instance resides.
   	RDS Instance ID	The ID of the source ApsaraDB RDS for MySQL instance. For example, rm-2z3m****.
   	Database Account	The database account for the source instance. For required permissions, see the "Permissions required for database accounts" section in Synchronize data from an ApsaraDB RDS for MySQL instance to an ApsaraDB for SelectDB instance.
   	Database Password	The password for the database account.
   	Encryption	Select Non-encrypted or SSL-encrypted. If you select SSL encryption, enable SSL on the RDS instance first. See Use a cloud certificate to enable SSL encryption.
   Destination Database	Database Type	Select SelectDB.
   	Access Method	Select Alibaba Cloud Instance.
   	Instance Region	The region where the destination ApsaraDB for SelectDB instance resides.
   	Instance ID	The ID of the destination ApsaraDB for SelectDB instance.
   	Database Account	The database account for the destination instance. For required permissions, see the "Permissions required for database accounts" section in Synchronize data from an ApsaraDB RDS for MySQL instance to an ApsaraDB for SelectDB instance.
   	Database Password	The password for the database account.

6. Click Test Connectivity and Proceed.

7. Configure the objects to synchronize and advanced settings.

   Parameter	Description
   Synchronization Types	By default, Incremental Data Synchronization is selected. Also select Schema Synchronization and Full Data Synchronization. After the precheck, DTS migrates the historical data of selected objects, which serves as the baseline for incremental synchronization. Important If you do not select Schema Synchronization, create tables using the Unique Key model in the destination SelectDB instance before starting. See Synchronize data from an ApsaraDB RDS for MySQL instance to an ApsaraDB for SelectDB instance and Data models.
   Source Objects	Select objects from the Source Objects section and click the arrow icon to move them to Selected Objects. You can select columns, tables, or databases. Important (1) If you select databases, DTS does not synchronize new tables created after the task starts. Select tables instead if you need to capture new tables. To change objects after the task starts, choose More > Reselect Objects in the task list. (2) If you select tables and need to rename them, the task supports up to 1,000 tables. For more than 1,000 tables, configure multiple tasks or synchronize at the database level.
   Selected Objects	To rename a single object in the destination, right-click it in Selected Objects. See Map the name of a single object. To rename multiple objects at once, click Batch Edit. See Map multiple object names at a time. Note (1) To filter which SQL operations to synchronize for a specific object, right-click the object and select the operations. See Synchronize data from an ApsaraDB RDS for MySQL instance to an ApsaraDB for SelectDB instance. (2) To filter rows by condition, right-click a table and specify a WHERE clause. See Specify filter conditions. (3) Renaming an object may cause dependent objects to fail synchronization.

8. (Optional) Click Next: Configure Database and Table Fields. In the dialog box, configure the Primary Key Column, Distribution Key, and Engine parameters for tables in the destination SelectDB instance.

   Note

   (1) This step is available only when Schema Synchronization is selected. To edit parameters, set Definition Status to All. (2) Primary Key Column accepts multiple columns. One or more primary key columns can also be used as the distribution key. Only Unique is available for Engine.

9. Save the task settings and run a precheck. Wait until the success rate reaches 100%, then click Next: Purchase Instance.

10. On the Purchase Instance page, configure the billing method and instance class. Read and agree to the Data Transmission Service (Pay-as-you-go) Service Terms, then click Buy and Start. Monitor task progress in the task list.

Set up the Flink environment

The following steps deploy a Flink standalone cluster version 1.16.

1. Download and extract the Flink package.

   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

   If this version is no longer available, download another version from Apache Flink downloads.

2. Download the MySQL CDC and SelectDB connector JARs to the FLINK_HOME/lib directory.

   cd flink-1.16.3/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://repo1.maven.org/maven2/com/selectdb/flink-selectdb-connector-1.16/2.1.0/flink-selectdb-connector-1.16-2.1.0.jar

3. Start the Flink standalone cluster.

   bin/start-cluster.sh

4. Create an ApsaraDB for SelectDB instance. See Create an instance.

5. Connect to the SelectDB instance over the MySQL protocol. See Connect to an instance.

6. Create a test database and table in SelectDB.

   -- Create the database
   CREATE DATABASE test_db;
   
   -- Create the 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;

Synchronize data using Flink CDC

Run the following command to start a Flink CDC synchronization job. All parameters use --mysql-conf and --sink-conf prefixes to separate MySQL source configuration from SelectDB sink configuration.

<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=****

Import historical data using a catalog

1. Connect to the SelectDB instance using a MySQL client.

   Note

   If you use Data Management (DMS) to connect, the SWITCH command is not supported. Use a MySQL client instead.

2. Create a Java Database Connectivity (JDBC) catalog for the MySQL data source.

   Parameter	Required	Default	Description
   user	Yes	—	The account used to connect to the MySQL database.
   password	Yes	—	The password for the account.
   jdbc_url	Yes	—	The JDBC URL for the MySQL database.
   driver_url	Yes	—	The name of the JDBC driver JAR file.
   driver_class	Yes	—	The JDBC driver class name.
   lower_case_table_names	No	false	Whether to synchronize database and table names in lowercase.
   only_specified_database	No	false	Whether to limit migration to specific databases.
   include_database_list	No	""	The databases to include. Takes effect only when only_specified_database is true. Separate multiple names with commas. Names are case-sensitive.
   exclude_database_list	No	""	The databases to exclude. Takes effect only when only_specified_database is true. Separate multiple names with commas. Names are case-sensitive.

   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"
   )

3. Create the destination table in SelectDB, then use INSERT INTO SELECT to copy the data.

   -- Create the destination table
   CREATE TABLE selectdb_table ...
   
   -- Import data from MySQL via the catalog
   INSERT INTO selectdb_table SELECT * FROM mysql_catalog.mysql_database.mysql_table;

   For details on the INSERT INTO SELECT syntax, see Import data by using INSERT INTO statements.

Add data sources

Before creating a synchronization task, add both a MySQL data source and a SelectDB data source to DataWorks.

1. Add a MySQL data source. See MySQL data source.

2. Add an ApsaraDB for SelectDB data source. See Add and manage data sources. Use the following parameters.

   Parameter	Description
   Data Source Name	The name of the data source.
   Host Address/IP Address	The JDBC URL in the format jdbc:mysql://<ip>:<port>/<dbname>. To get the VPC endpoint or public endpoint and MySQL port, go to the Instance Details page in the SelectDB console and view the Network Information section of the Basic Information page. Example: jdbc:mysql://selectdb-cn-4xl3jv1****.selectdbfe.rds.aliyuncs.com:9030/test_db
   HTTP Connection Address	The HTTP address and port in the format <ip>:<port>. To get the HTTP port, go to the Instance Details page and view the HTTP Port field in the Network Information section. Example: selectdb-cn-4xl3jv1****.selectdbfe.rds.aliyuncs.com:8080
   Username	The username of the owner account of the SelectDB instance.
   Password	The password of the owner account.

Configure a synchronization task

Configure a synchronization task using the codeless UI or the code editor:

• Synchronize all data in a MySQL database to SelectDB in real time

• Configure a batch synchronization task by using the codeless UI

• Configure a batch synchronization task by using the code editor