Realtime Compute for Apache Flink:SelectDB コネクタ

コネクタのアップロードと構成

1. JAR ファイルをクリックして、SelectDB コネクタ JAR (バージョン 1.15 ～ 1.17) をダウンロードします。

2. SelectDB コネクタ JAR を Realtime Compute for Apache Flink コンソールにアップロードします。詳細については、「カスタムコネクタの管理」をご参照ください。

3. SQL ドラフトを作成する、カスタム SelectDB コネクタを使用します。

   connector オプションを doris に設定します。その他のシンクオプションについては、「Doris シンクの構成項目」をご参照ください。

構文

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 Doris コネクタトピックの「型マッピング」セクションを参照してください。

コネクタの使用

このセクションでは、カスタム SelectDB コネクタを使用して、ApsaraDB RDS for MySQL から ApsaraDB for SelectDB にデータを同期する方法について説明します。

1. データ同期の準備をします。

   1. Flink ワークスペース、ApsaraDB RDS for MySQL インスタンス、および ApsaraDB for SelectDB インスタンスを作成します。

   2. ApsaraDB RDS for 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 を使用して ApsaraDB for 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 ワークスペースの vSwitch の [CIDR ブロック] を ApsaraDB for SelectDB インスタンスの IP アドレスホワイトリストに追加します。詳細については、「IP アドレスホワイトリストを構成する方法」をご参照ください。

2. Realtime Compute for Apache Flink コンソールで、SQL ジョブを開発して開始します。

   1. mysqlcatalog という名前の MySQL カタログを作成します。詳細については、「MySQL カタログの管理」をご参照ください。

   2. JAR ファイルをクリックして、SelectDB コネクタ (バージョン 1.15 ～ 1.17) JAR をダウンロードし、JAR ファイルをアップロードします。詳細については、「カスタムコネクタの管理」をご参照ください。

   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 を使用して ApsaraDB for SelectDB インスタンスに接続した後、selecttable テーブルのデータをクエリします。

   SELECT * FROM `selecttable` ;

構文

source:
   type: xxx

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

構成オプション

データ型マッピング