SelectDB

使用方法

1. 单击JAR包获取SelectDB自定义连接器（需要为1.15~1.17）。

2. 在实时计算开发控制台上，上传SelectDB自定义连接器，详情请参见管理自定义连接器。

3. 在SQL作业中使用SelectDB自定义连接器，作业开发详情请参见SQL作业开发。

   connector为表类型，固定值为doris。SelectDB自定义连接器结果表参数配置详情请参见Sink配置项。

语法结构

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'
);

类型映射

详情请参见Doris & Flink Column Type Mapping。

使用示例

本文以MySQL数据写入SelectDB为例为您详细介绍如何使用SelectDB自定义连接器。

1. 准备工作。

   1. 创建Flink工作空间、MySQL和SelectDB实例，详情请参见开通实时计算Flink版、第一步：快捷创建RDS MySQL实例与配置数据库和创建实例。

   2. 在MySQL中创建名称为order_dw_mysql的数据库和名称为orders的表并导入测试数据。

      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. 通过DMS连接云数据库SelectDB版实例后创建名称为selectdb的数据库和名称为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. 将实时计算Flink版的虚拟交换机的网段信息添加到SelectDB的白名单中，详情请参见如何设置白名单？。

2. 在 实时计算开发控制台上创建Flink SQL作业并启动。

   1. 创建名称为mysqlcatalog的MySQL Catalog，详情请参见管理MySQL Catalog。

   2. 单击JAR包获取SelectDB自定义连接器（需要为1.15~1.17），注册SelectDB自定义连接器，详情请参见管理自定义连接器。

   3. 在数据开发 > ETL新建作业草稿，代码示例如下。

      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. 单击部署后无状态启动作业，详情请参见部署作业和作业启动。

3. 通过DMS连接云数据库SelectDB版实例后，查询名称为selecttable的表数据。

   SELECT * FROM `selecttable` ;

语法结构

source:
   type: xxx

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

配置项

类型映射