Realtime Compute for Apache Flink:SelectDB connector

Upload and configure the connector

1. Click JAR file to download a SelectDB connector JAR (version 1.15 to 1.17).

2. Upload the SelectDB connector JAR to the Realtime Compute for Apache Flink console. For more information, see Manage custom connectors.

3. Create an SQL draft and use the custom SelectDB connector.

   Set the connector option to doris. For information about other sink options, see Configuration items of Doris sink.

Syntax

CREATE TABLE selectdb_sink (
  emp_no       INT ,
  birth_date   DATE,
  first_name   STRING,
  last_name    STRING,
  gender       STRING,
  hire_date    DATE
) WITH (
  'connector' = 'doris',
  'fenodes' = 'selectdb-cn-*******.selectdbfe.rds.aliyuncs.com:8080',
  'table.identifier' = 'test.employees',
  'username' = 'admin',
  'password' = '****',
  'sink.enable-delete' = 'true'
);

Data type mappings

See the "Type Mapping" section of the Flink Doris connector topic in the Doris documentation.

Use the connector

This section illustrates how to synchronize data from ApsaraDB RDS for MySQL to ApsaraDB for SelectDB by using the custom SelectDB connector.

1. Prepare for data synchronization.

   1. Create a Flink workspace, an ApsaraDB RDS for MySQL instance, and an ApsaraDB for SelectDB instance.

   2. In the ApsaraDB RDS for MySQL console, create a database named order_dw_mysql and a table named orders, and import test data into the table.

      CREATE TABLE `orders` (
        order_id bigint not null primary key,
        user_id varchar(50) not null,
        shop_id bigint not null,
        product_id bigint not null,
        buy_fee decimal(20,2) not null,   
        create_time timestamp not null,
        update_time timestamp not null default now(),
        state int not null 
      );
      
      INSERT INTO orders VALUES
      (100001, 'user_001', 12345, 1, 5000.05, '2023-02-15 16:40:56', '2023-02-15 18:42:56', 1),
      (100002, 'user_002', 12346, 2, 4000.04, '2023-02-15 15:40:56', '2023-02-15 18:42:56', 1),
      (100003, 'user_003', 12347, 3, 3000.03, '2023-02-15 14:40:56', '2023-02-15 18:42:56', 1),
      (100004, 'user_001', 12347, 4, 2000.02, '2023-02-15 13:40:56', '2023-02-15 18:42:56', 1),
      (100005, 'user_002', 12348, 5, 1000.01, '2023-02-15 12:40:56', '2023-02-15 18:42:56', 1),
      (100006, 'user_001', 12348, 1, 1000.01, '2023-02-15 11:40:56', '2023-02-15 18:42:56', 1),
      (100007, 'user_003', 12347, 4, 2000.02, '2023-02-15 10:40:56', '2023-02-15 18:42:56', 1);

   3. After connecting to an ApsaraDB for SelectDB instance by using DMS, create a database named selectdb and a table named selecttable.

      CREATE DATABASE selectdb;
      
      CREATE TABLE `selecttable` (
        order_id bigint,
        user_id varchar(50),
        shop_id bigint,
        product_id bigint,
        buy_fee DECIMAL,   
        create_time DATETIME,
        update_time DATETIME,
        state int
       )DISTRIBUTED BY HASH(order_id) BUCKETS 10;

   4. Add the CIDR Block of the vSwitch in your Flink workspace to the IP address whitelist of your ApsaraDB for SelectDB instance. For more information, see How do I configure an IP address whitelist?.

2. In the Realtime Compute for Apache Flink console, develop an SQL job and start it.

   1. Create a MySQL catalog named mysqlcatalog. For more information, see Manage MySQL catalogs.

   2. Click JAR file to download the SelectDB connector (version 1.15 to 1.17) JAR, and upload the JAR file. For more information, see Manage custom connectors.

   3. Go to Development > ETL, click New to create a blank stream draft, and copy the following code to the draft:

      CREATE TEMPORARY TABLE  selectdb_sink (
        order_id BIGINT,
        user_id STRING,
        shop_id BIGINT,
        product_id BIGINT,
        buy_fee DECIMAL,   
        create_time TIMESTAMP(6),
        update_time TIMESTAMP(6),
        state int
      ) 
        WITH (
        'connector' = 'doris',
        'fenodes' = 'selectdb-cn-jfj3z******.selectdbfe.rds.aliyuncs.com:8080',
        'table.identifier' = 'selectdb.selecttable',
        'username' = 'admin',
        'password' = '${secret_values.selectdb}',
        'sink.enable-delete' = 'true'
      );
      
      INSERT INTO selectdb_sink SELECT * FROM `mysqlcatalog`.`order_dw_mysql`.`orders`;

   4. Click Deploy and start the deployment in the initial mode. For more information, see Create a deployment and Start a deployment.

3. After connecting to an ApsaraDB for SelectDB instance by using DMS, query data in the selecttable table.

   SELECT * FROM `selecttable` ;

Syntax

source:
   type: xxx

sink:
   type: doris
   name: Doris Sink
   fenodes: 127.0.0.1:8030
   username: root
   password: ""
   table.create.properties.replication_num: 1

Configuration options

Data type mappings