ApsaraDB for SelectDB: Import data using DataX - ApsaraDB for SelectDB

This topic describes how to use the DataX Doris Writer plug-in to synchronize data to an ApsaraDB for SelectDB instance.

Overview

DataX is an open source tool from Alibaba for offline data synchronization between various heterogeneous data sources. These sources include relational databases such as MySQL and Oracle, as well as HDFS, Hive, MaxCompute, HBase, and FTP. You can use DataX to read data from an upstream data source and then write the data to an ApsaraDB for SelectDB instance by using the DataX Doris Writer plug-in.

Prerequisites

A Maven environment is installed.
Python 3.6 or later is installed.

Example

This example shows how to use DataX to import data from a MySQL data source to an ApsaraDB for SelectDB instance in a Linux environment.

Step 1: Configure the DataX environment

Download the DataX package. The plug-in code is available from the Doris community.
Run the init-env.sh script in the DataX package to build the development environment.
```
sh init-env.sh
```
Compile the mysqlreader and doriswriter plug-ins.
1. Compile the entire DataX project.
```
cd DataX/
mvn package assembly:assembly -Dmaven.test.skip=true
```
  The build output is located in the target/datax/datax/ directory.
  Note
  The hdfsreader, hdfswriter, ossreader, and osswriter plug-ins require additional JAR packages. If you do not need these plug-ins, you can remove their modules from the DataX/pom.xml file.
2. Compile the mysqlreader and doriswriter plug-ins individually.
```
mvn clean install -pl plugin-rdbms-util,mysqlreader,doriswriter -DskipTests
```

Step 2: Prepare source data

Create a test table in MySQL.

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

Use Data Management (DMS) to generate test data. For more information, see Generate test data.

Step 3: Configure the ApsaraDB for SelectDB instance

Connect to the ApsaraDB for SelectDB instance using 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;

Step 4: Synchronize data using DataX

Apply for a public endpoint for the ApsaraDB for SelectDB instance. For more information, see Apply for and release a public endpoint.
Add the public IP address of the DataX host to the IP address whitelist of the instance. For more information, see Configure an IP address whitelist.

Create a configuration file named mysqlToSelectDB.json to define the import job.

{
  "job":{
    "content":[
      {
        "reader":{
            "name": "mysqlreader",
            "parameter": {
                "column": [
                    "emp_no",
                    "birth_date",
                    "first_name",
                    "last_name",
                    "gender",
                    "hire_date"
                ],
                "where": "emp_no>0",
                "connection": [
                    {
                        "jdbcUrl": [
                            "jdbc:mysql://host:port/test_db?useUnicode=true&allowPublicKeyRetrieval=true&characterEncoding=utf-8"
                        ],
                        "table": [
                            "employees"
                        ]
                    }
                ],
                "password": "123456",
                "splitPk": "emp_no",
                "username": "admin"
            }
        },
        "writer":{
          "name":"doriswriter",
          "parameter":{
            "loadUrl":[
              "selectdb-cn-xxx-public.selectdbfe.rds.aliyuncs.com:8080"
            ],
            "loadProps":{
              "format":"json",
              "strip_outer_array":"true"
            },
            "column":[
                "emp_no",
                "birth_date",
                "first_name",
                "last_name",
                "gender",
                "hire_date"
            ],
            "username":"admin",
            "password":"123456",
            "postSql":[

            ],
            "preSql":[

            ],
            "connection":[
              {
                "jdbcUrl":"jdbc:mysql://selectdb-cn-xxx-public.selectdbfe.rds.aliyuncs.com:9030/test_db",
                "table":[
                  "employees"
                ],
                "selectedDatabase":"test_db"
              }
            ],
            "maxBatchRows":1000000,
            "batchSize":536870912000
          }
        }
      }
    ],
    "setting":{
      "errorLimit":{
        "percentage":0.02,
        "record":0
      },
      "speed":{
        "channel":5
      }
    }
  }
}

Parameters

Parameter	Required	Default	Description
jdbcUrl	Yes	N/A	The JDBC connection URL in the format `jdbc:mysql://<ip>:<port>`. You can find the IP address and MySQL port number on the Instance Details > Network Information page for your ApsaraDB for SelectDB instance in the console. IP address: The VPC Endpoint or Public Endpoint. Port: The MySQL Port. Example: `jdbc:mysql://selectdb-cn-4xl3jv1**.selectdbfe.rds.aliyuncs.com:9030` Note** If the DataX host and the ApsaraDB for SelectDB instance are in the same VPC, use the VPC endpoint. Otherwise, use the public endpoint.
loadUrl	Yes	N/A	The HTTP endpoint for Stream Load, in the format `<ip>:<port>`. You can find the IP address and HTTP port number on the Instance Details > Network Information page for your ApsaraDB for SelectDB instance in the console. IP address: The VPC Endpoint or Public Endpoint. Port: The HTTP Port. Example: `selectdb-cn-4xl3jv1**.selectdbfe.rds.aliyuncs.com:8080` Note** If the DataX host and the ApsaraDB for SelectDB instance are in the same VPC, use the VPC endpoint. Otherwise, use the public endpoint.
username	Yes	N/A	The username for your ApsaraDB for SelectDB instance.
password	Yes	N/A	The password for the specified user of the ApsaraDB for SelectDB instance.
connection.selectedDatabase	Yes	N/A	The name of the destination database in your ApsaraDB for SelectDB instance.
connection.table	Yes	N/A	The name of the destination table in your ApsaraDB for SelectDB instance.
column	Yes	N/A	The columns in the destination table where you want to write data. These column names are used as the field names in the generated JSON data. Separate column names with commas (,). Example: `"column": ["id","name","age"]`.
preSql	No	N/A	SQL statements to execute before the data writing process begins.
postSql	No	N/A	SQL statements to execute after the data writing process completes.
maxBatchRows	No	500000	The maximum number of rows to import in a single batch. This parameter, along with `batchSize`, controls the size of each import batch. DataX loads a batch when either the `maxBatchRows` or `batchSize` threshold is reached.
batchSize	No	104857600	The maximum data size, in bytes, to import in a single batch. This parameter, along with `maxBatchRows`, controls the size of each import batch. DataX loads a batch when either the `maxBatchRows` or `batchSize` threshold is reached. The default value is 100 MB.
maxRetries	No	3	The number of times to retry a failed batch import.
labelPrefix	No	datax_doris_writer_	The prefix for the load job label. The final label for each batch is a combination of `labelPrefix` and a UUID, which ensures that each load job is globally unique and prevents duplicate data imports.
loadProps	No	N/A	Properties for the Stream Load job. These properties correspond to the parameters of a Stream Load request. For more information, see Stream Load parameters. Use the `format` property to specify the data format. The default format is CSV. JSON is also supported. For more details, see Data type conversion.
flushInterval	No	30000	The time interval, in milliseconds, for flushing data batches. The default value is 30000.

Submit the job from the command line.

cd target/datax/datax/bin
python datax.py ../mysqlToSelectDB.json

Troubleshoot Stream Load connection timeouts in DataX/doriswriter

Symptom

When you use DataX (doriswriter) in DataWorks Data Integration to synchronize data to an ApsaraDB for SelectDB instance, the job fails with a Connection timed out error.

Cause

When doriswriter writes data, it uses Stream Load. The FE node redirects the write request to a BE node. As a result, the client must be able to connect directly to the IP addresses of the BE nodes. Configuring a custom route for the FE endpoint alone is not sufficient.

Solution

Connect to the SelectDB instance using the MySQL protocol and run the following statement to obtain the IP addresses of all BE nodes.
```
SHOW BACKENDS;
```
In DataWorks, add custom routes for all of the BE node IP addresses that you obtained in the previous step.
Run the synchronization job again and verify that the connection succeeds.

Data type conversion

By default, DataX converts all incoming data into strings and creates a CSV file for the import, using a tab character (\t) as the column delimiter and a newline character (\n) as the row delimiter. To change the delimiters, configure the loadProps parameter as shown in the following example.

"loadProps": {
    "format": "csv",
    "column_separator": "\\x01",
    "line_delimiter": "\\x02"
}

To change the import format to JSON, set the format property within loadProps as shown in the following example.

"loadProps": {
    "format": "json",
    "strip_outer_array": true
}