Import data to ApsaraDB for SelectDB using BitSail-ApsaraDB for SelectDB(SelectDB)-阿里云帮助中心

ApsaraDB for SelectDB integrates with BitSail, enabling you to use the SelectDB sink to import table data into an ApsaraDB for SelectDB instance. This topic describes how to use the SelectDB sink to import data into ApsaraDB for SelectDB.

Overview

BitSail is a high-performance, distributed data integration engine. It supports data synchronization between heterogeneous data sources and supports offline, real-time, full, and incremental data integration. The BitSail engine reads large amounts of data from sources like MySQL, Hive, and Kafka, and the SelectDB sink then writes that data to ApsaraDB for SelectDB.

Prerequisites

BitSail version 0.1.0 or later.

How it works

Configure the SelectDB connector parameters in the job.writer section of your BitSail configuration file. The following code provides an example:

{ 
  "job": { 
    "writer": { 
      "class": "com.bytedance.bitsail.connector.selectdb.sink.SelectdbSink", 
      "load_url": "<selectdb_http_address>", 
      "jdbc_url": "<selectdb_mysql_address>", 
      "cluster_name": "<selectdb_cluster_name>", 
      "user": "<username>", 
      "password": "<password>", 
      "table_identifier": "<selectdb_table_identifier>", 
      "columns": [ 
        { 
          "index": 0, 
          "name": "id", 
          "type": "int" 
        }, 
        { 
          "index": 1, 
          "name": "bigint_type", 
          "type": "bigint" 
        }, 
        { 
          "index": 2, 
          "name": "string_type", 
          "type": "varchar" 
        }, 
        { 
          "index": 3, 
          "name": "double_type", 
          "type": "double" 
        }, 
        { 
          "index": 4, 
          "name": "date_type", 
          "type": "date" 
        } 
      ] 
    } 
  }
}

The following table describes the parameters.

Parameter	Required	Description
class	Yes	The class of the ApsaraDB for SelectDB writer connector. Default: `com.bytedance.bitsail.connector.selectdb.sink.SelectdbSink`.
load_url	Yes	The access address and HTTP port of the ApsaraDB for SelectDB instance. You can find the VPC Endpoint (or Public Endpoint) and HTTP Port on the Instance Details > Network Information page in the ApsaraDB for SelectDB console. Example: `selectdb-cn-4xl3jv1****.selectdbfe.rds.aliyuncs.com:8080`.
jdbc_url	Yes	ApsaraDB for SelectDB instance's access address and MySQL protocol port. You can find the VPC Endpoint (or Public Endpoint) and MySQL Port on the Instance Details > Network Information page of the ApsaraDB for SelectDB console. Example: `selectdb-cn-4xl3jv1****.selectdbfe.rds.aliyuncs.com:9030`.
cluster_name	Yes	The name of the target cluster in ApsaraDB for SelectDB, as an instance can contain multiple clusters.
user	Yes	The username for your ApsaraDB for SelectDB instance.
password	Yes	The password for the specified user of the ApsaraDB for SelectDB instance.
table_identifier	Yes	The name of the target table in ApsaraDB for SelectDB, in the format `database_name.table_name`. Example: `test_db.test_table`.
writer_parallelism_num	No	The number of concurrent writes to ApsaraDB for SelectDB.
sink_flush_interval_ms	No	The flush interval in milliseconds for upsert mode. Default: 5000.
sink_max_retries	No	The maximum number of write retries. Default: 3.
sink_buffer_size	No	The maximum buffer size in bytes. Default: 1048576 (1 MB).
sink_buffer_count	No	The initial number of buffers. Default: 3.
sink_enable_delete	No	Specifies whether to synchronize `DELETE` events.
sink_write_mode	No	The write mode. The only supported value is `BATCH_UPSERT`.
stream_load_properties	No	Additional parameters to include in the Stream Load URL, specified as a map of key-value pairs (Map<String, String>).
load_contend_type	No	The data format for the Stream Load job. Valid values: `CSV` or `JSON`. Default: `JSON`.
csv_field_delimiter	No	The field delimiter for data in CSV format. Default: `,` (comma).
csv_line_delimiter	No	The line delimiter for data in CSV format. Default: `\n`.

Example

This example demonstrates how to use BitSail to generate mock data and import it into ApsaraDB for SelectDB.

Environment setup

Configure your BitSail environment. Download and decompress the BitSail installation package.
```
wget feilun-justtmp.oss-cn-hongkong.aliyuncs.com/bitsail.tar.gz
tar -zxvf bitsail.tar.gz
```
Configure your 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.
```
CREATE TABLE `test_table` (
  `id` BIGINT(20) NULL,
  `bigint_type` BIGINT(20) NULL,
  `string_type` VARCHAR(100) NULL,
  `double_type` DOUBLE NULL,
  `decimal_type` DECIMALV3(27, 9) NULL,
  `date_type` DATEV2 NULL,
  `partition_date` DATEV2 NULL
) ENGINE=OLAP
DUPLICATE KEY(`id`)
COMMENT 'OLAP'
DISTRIBUTED BY HASH(`id`) BUCKETS 10
PROPERTIES (
  "light_schema_change" = "true"
);
```
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 your BitSail environment to the instance's IP address whitelist. For more information, see Configure an IP address whitelist.

Run a local BitSail job

Create a job configuration file named test.json. This example uses the FakeSource class from the BitSail package to generate mock data for the import.

{
  "job": {
    "common": {
      "job_id": -2413,
      "job_name": "bitsail_fake_to_selectdb_test",
      "instance_id": -20413,
      "user_name": "user"
    },
    "reader": {
      "class": "com.bytedance.bitsail.connector.legacy.fake.source.FakeSource",
      "total_count": 300,
      "rate": 10000,
      "random_null_rate": 0,
      "unique_fields": "id",
      "columns_with_fixed_value": [
        {
          "name": "partition_date",
          "fixed_value": "2022-10-10"
        }
      ],
      "columns": [
        {
          "index": 0,
          "name": "id",
          "type": "long"
        },
        {
          "index": 1,
          "name": "bigint_type",
          "type": "long"
        },
        {
          "index": 2,
          "name": "string_type",
          "type": "string"
        },
        {
          "index": 3,
          "name": "double_type",
          "type": "double"
        },
        {
          "index": 4,
          "name": "decimal_type",
          "type": "double"
        },
        {
          "index": 5,
          "name": "date_type",
          "type": "date.date"
        },
        {
          "index": 6,
          "name": "partition_date",
          "type": "string"
        }
      ]
    },
    "writer": {
      "class": "com.bytedance.bitsail.connector.selectdb.sink.SelectdbSink",
      "load_url": "selectdb-cn-4xl3jv1****.selectdbfe.rds.aliyuncs.com:8080",
      "jdbc_url": "selectdb-cn-4xl3jv1****.selectdbfe.rds.aliyuncs.com:9030",
      "cluster_name": "new_cluster",
      "user": "admin",
      "password": "****",
      "table_identifier": "test_db.test_table",
      "columns": [
        {
          "index": 0,
          "name": "id",
          "type": "bigint"
        },
        {
          "index": 1,
          "name": "bigint_type",
          "type": "bigint"
        },
        {
          "index": 2,
          "name": "string_type",
          "type": "varchar"
        },
        {
          "index": 3,
          "name": "double_type",
          "type": "double"
        },
        {
          "index": 4,
          "name": "decimal_type",
          "type": "double"
        },
        {
          "index": 5,
          "name": "date_type",
          "type": "date"
        },
        {
          "index": 6,
          "name": "partition_date",
          "type": "date"
        }
      ]
    }
  }
}

Submit the job from the command line.

bash bin/bitsail run --engine flink --execution-mode run --deployment-mode local --conf test.json