ApsaraDB for SelectDB:Use SeaTunnel to import data

Overview

SeaTunnel is an easy-to-use, high-performance distributed data integration platform that supports real-time synchronization of large amounts of data. You can use SeaTunnel to read large amounts of data from data sources such as MySQL, Hive, and Kafka, and then use SeaTunnel SelectDB Sink to write the data to ApsaraDB for SelectDB.

Prerequisites

SeaTunnel 2.3.1 or later is installed.

How it works

SeaTunnel allows you to write upstream data in the JSON or CSV format to ApsaraDB for SelectDB. The sink configurations vary with the data format.

JSON format

sink { 
  SelectDB { 
    load-url="ip:http_port" 
    jdbc-url="ip:mysql_port" 
    cluster-name="Cluster" 
    table.identifier="test_db.test_table" 
    username="admin" 
    password="****" 
    selectdb.config { 
      file.type="json" 
    } 
  }
}

CSV format

sink { 
  SelectDB { 
    load-url="ip:http_port" 
    jdbc-url="ip:mysql_port" 
    cluster-name="Cluster" 
    table.identifier="test_db.test_table" 
    username="admin" 
    password="****" 
    selectdb.config { 
      file.type="csv" 
      file.column_separator="," 
      file.line_delimiter="\n" 
    } 
  }
}

The following table describes the parameters in the sink configurations.

Example

In this example, SeaTunnel is used to import data from an upstream MySQL database to ApsaraDB for SelectDB. The following table describes the software versions in the example.

Prepare the environment

1. Configure SeaTunnel.

   1. Download and decompress the SeaTunnel installation package. In this example, the SeaTunnel installation package apache-seatunnel-2.3.3-bin.tar.gz is used.

      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. Modify the SEATUNNEL_HOME/config/plugin_config file. Retain only the required connectors.

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

   3. Install the SeaTunnel connector plug-in.

      sh bin/install-plugin.sh

   4. Download the MySQL driver and place it in the SEATUNNEL_HOME/jar directory.

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

2. Construct the data to be imported. In this example, a small amount of sample data is constructed in a MySQL database for import.

   1. Create a test table in the 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. Use Data Management (DMS) to generate test data. For more information, see Generate test data.

3. Configure an ApsaraDB for SelectDB instance.

   1. Create an ApsaraDB for SelectDB instance. For more information, see Create an instance.

   2. Connect to the ApsaraDB for SelectDB instance over the MySQL protocol. For more information, see Connect to an instance.

   3. Create a test database and a test table.

      1. Create a test database.

         CREATE DATABASE test_db;

      2. 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;

   4. Apply for a public endpoint for the ApsaraDB for SelectDB instance. For more information, see Apply for or release a public endpoint.

   5. Add the public IP address of SeaTunnel to the IP address whitelist of the ApsaraDB for SelectDB instance. For more information, see Configure an IP address whitelist.

Use the local SeaTunnel engine to synchronize data from the MySQL database to the ApsaraDB for SelectDB instance

1. Create the mysqlToSelectDB.conf configuration file and configure the job information.

   env {
     execution.parallelism = 2
     job.mode = "BATCH"
     checkpoint.interval = 10000
   }
   
   source{
     jdbc {
       url = "jdbc:mysql://host:ip/test_db"
       driver = "com.mysql.cj.jdbc.Driver"
       user = "admin"
       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="admin"
       password="****"
       selectdb.config {
           file.type="json"
       }
     }
   }

2. Submit the job.

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