ApsaraDB for SelectDB:Use SeaTunnel to import data

JSON format

sink {
  SelectDB {
    load-url="<host>:<http_port>"
    jdbc-url="<host>:<mysql_port>"
    cluster-name="<cluster_name>"
    table.identifier="<db_name>.<table_name>"
    username="<username>"
    password="<password>"
    selectdb.config {
      file.type="json"
      file.strip_outer_array="false"
    }
  }
}

CSV format

sink {
  SelectDB {
    load-url="<host>:<http_port>"
    jdbc-url="<host>:<mysql_port>"
    cluster-name="<cluster_name>"
    table.identifier="<db_name>.<table_name>"
    username="<username>"
    password="<password>"
    selectdb.config {
      file.type="csv"
      file.column_separator=","
      file.line_delimiter="\n"
    }
  }
}

Parameters

Step 1: Set up SeaTunnel

1. Download and extract the SeaTunnel installation package.

   wget https://dlcdn.apache.org/seatunnel/2.3.3/apache-seatunnel-2.3.3-bin.tar.gz
   tar -xzvf apache-seatunnel-2.3.3-bin.tar.gz

2. Edit $SEATUNNEL_HOME/config/plugin_config to include only the connectors needed for this example.

   --connectors-v2--
   connector-cdc-mysql
   connector-selectdb-cloud
   connector-jdbc
   connector-fake
   connector-console
   connector-assert
   --end--

3. Install the connector plugins.

   sh bin/install-plugin.sh

4. Download the MySQL JDBC driver and place it in the lib/ directory.

   cd lib/
   wget https://repo1.maven.org/maven2/mysql/mysql-connector-java/8.0.28/mysql-connector-java-8.0.28.jar

Step 2: Prepare the source data in MySQL

1. Create the source table in your MySQL database.

   CREATE TABLE `employees` (
     `emp_no`     INT          NOT NULL,
     `birth_date` DATE         NOT NULL,
     `first_name` VARCHAR(14)  NOT NULL,
     `last_name`  VARCHAR(16)  NOT NULL,
     `gender`     ENUM('M','F') NOT NULL,
     `hire_date`  DATE         NOT NULL,
     PRIMARY KEY (`emp_no`)
   ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb3;

2. Populate the table with test data. You can use Data Management (DMS) to generate test data.

Step 3: Configure ApsaraDB for SelectDB

1. Create an ApsaraDB for SelectDB instance.

2. Connect to the instance over the MySQL protocol, then create the target database and table.

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

3. Apply for a public endpoint so SeaTunnel can reach the instance over the internet.

4. Add the public IP address of your SeaTunnel host to the IP address whitelist of the instance.

Step 4: Run the sync job

1. Create the job configuration file mysqlToSelectDB.conf.

   env {
     execution.parallelism = 2
     job.mode = "BATCH"
     checkpoint.interval = 10000
   }
   
   source {
     jdbc {
       url = "jdbc:mysql://<mysql_host>:<mysql_port>/test_db"
       driver = "com.mysql.cj.jdbc.Driver"
       user = "<mysql_username>"
       password = "<mysql_password>"
       query = "select * from employees"
     }
   }
   
   sink {
     SelectDBCloud {
       load-url="selectdb-cn-pe33hab****-public.selectdbfe.rds.aliyuncs.com:8080"
       jdbc-url="selectdb-cn-pe33hab****-public.selectdbfe.rds.aliyuncs.com:9030"
       cluster-name="new_cluster"
       table.identifier="test_db.employees"
       username="<selectdb_username>"
       password="<selectdb_password>"
       selectdb.config {
         file.type="json"
       }
     }
   }

   Replace the placeholder values with your actual connection details. The load-url and jdbc-url values are available on the Instance Details page under Network Information.

2. Submit the job.

   sh ./bin/seatunnel.sh --config ./mysqlToSelectDB.conf -e local

   After the job completes, query the test_db.employees table in ApsaraDB for SelectDB to verify that the data was imported successfully.