Realtime Compute for Apache Flink：Konektor SelectDB

Unggah dan konfigurasikan konektor

1. Klik file JAR untuk mengunduh file JAR konektor SelectDB (versi 1.15 hingga 1.17).

2. Unggah file JAR konektor SelectDB ke Konsol Realtime Compute for Apache Flink. Untuk informasi lebih lanjut, lihat Kelola Konektor Kustom.

3. Buat draft SQL dan gunakan konektor SelectDB kustom.

   Atur opsi connector menjadi doris. Untuk informasi tentang opsi sink lainnya, lihat Item Konfigurasi Sink Doris.

Sintaksis

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

Pemetaan tipe data

Lihat bagian "Pemetaan Tipe" topik konektor Flink Doris di dokumentasi Doris.

Gunakan konektor

Bagian ini mengilustrasikan cara menyinkronkan data dari ApsaraDB RDS for MySQL ke ApsaraDB for SelectDB menggunakan konektor SelectDB kustom.

1. Persiapkan sinkronisasi data.

   1. Buat ruang kerja Flink, instans ApsaraDB RDS for MySQL, dan instans ApsaraDB for SelectDB.

   2. Di konsol ApsaraDB RDS for MySQL, buat database bernama order_dw_mysql dan tabel bernama orders, lalu impor data uji ke dalam tabel tersebut.

      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. Setelah terhubung ke instans ApsaraDB for SelectDB menggunakan DMS, buat database bernama selectdb dan tabel bernama 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. Tambahkan CIDR Block dari vSwitch di ruang kerja Flink Anda ke daftar putih alamat IP instans ApsaraDB for SelectDB Anda. Untuk informasi lebih lanjut, lihat Bagaimana cara mengonfigurasi daftar putih alamat IP?.

2. Di Konsol Realtime Compute for Apache Flink, kembangkan pekerjaan SQL dan mulailah.

   1. Buat katalog MySQL bernama mysqlcatalog. Untuk informasi lebih lanjut, lihat Kelola Katalog MySQL.

   2. Klik file JAR untuk mengunduh file JAR konektor SelectDB (versi 1.15 hingga 1.17), lalu unggah file JAR tersebut. Untuk informasi lebih lanjut, lihat Kelola Konektor Kustom.

   3. Pergi ke Development > ETL, klik New untuk membuat draft stream kosong, dan salin kode berikut ke dalam 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. Klik Deploy dan mulai penerapan dalam mode awal. Untuk informasi lebih lanjut, lihat Buat Penerapan dan Mulai Penerapan.

3. Setelah terhubung ke instans ApsaraDB for SelectDB menggunakan DMS, kueri data di tabel selecttable.

   SELECT * FROM `selecttable` ;

Sintaksis

source:
   type: xxx

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

Opsi konfigurasi

Pemetaan tipe data