ApsaraDB for SelectDB：通過SeaTunnel匯入資料

概述

SeaTunnel是一款簡單易用、高效能的分布式Data Integration平台，支援海量資料即時同步。您可以通過SeaTunnel平台讀取MySQL、Hive、Kafka等資料來源中的海量資料，然後由SeaTunnel SelectDB Sink將資料寫入到ApsaraDB for SelectDB中。

前提條件

SeaTunnel 2.3.1版本及以上。

使用方式

SeaTunnel支援以JSON格式或CSV格式將上遊資料寫入到ApsaraDB for SelectDB，不同寫入方式的配置文法如下。

JSON格式

sink { 
  SelectDB { 
    load-url="ip:http_port" 
    jdbc-url="ip:mysql_port" 
    cluster-name="Cluster" 
    table.identifier="test_db.test_table" 
    username="admin" 
    password="****" 
    selectdb.config { 
      file.type="json" 
    } 
  }
}

CSV格式

sink { 
  SelectDB { 
    load-url="ip:http_port" 
    jdbc-url="ip:mysql_port" 
    cluster-name="Cluster" 
    table.identifier="test_db.test_table" 
    username="admin" 
    password="****" 
    selectdb.config { 
      file.type="csv" 
      file.column_separator="," 
      file.line_delimiter="\n" 
    } 
  }
}

參數說明如下。

使用樣本

以MySQL資料來源為例，為您介紹如何通過SeaTunnel將上遊的MySQL資料匯入至ApsaraDB for SelectDB。樣本中各軟體版本如下：

環境準備

1. 配置SeaTunnel環境。

   1. 下載並解壓SeaTunnel安裝包。本樣本中使用SeaTunnel安裝包：apache-seatunnel-2.3.3-bin.tar.gz。

      wget https://dlcdn.apache.org/seatunnel/2.3.3/apache-seatunnel-2.3.3-bin.tar.gz
      tar -xzvf apache-seatunnel-2.3.3-bin.tar.gz

   2. 修改SEATUNNEL_HOME/config/plugin_config設定檔，保留需要的Connector外掛程式。

      --connectors-v2--
      connector-cdc-mysql
      connector-selectdb-cloud
      connector-jdbc
      connector-fake
      connector-console
      connector-assert
      --end--

   3. 安裝SeaTunnel Connector外掛程式。

      sh bin/install-plugin.sh

   4. 下載MySQL驅動並放至SEATUNNEL_HOME/jar目錄。

      cd lib/
      wget https://repo1.maven.org/maven2/mysql/mysql-connector-java/8.0.28/mysql-connector-java-8.0.28.jar

2. 構造需要匯入的資料。本文以MySQL為例，構造少量範例資料來完成匯入。

   1. 建立MySQL測試表。

      CREATE TABLE `employees` (
        `emp_no` INT NOT NULL,
        `birth_date` DATE NOT NULL,
        `first_name` VARCHAR(14) NOT NULL,
        `last_name` VARCHAR(16) NOT NULL,
        `gender` ENUM('M','F') NOT NULL,
        `hire_date` DATE NOT NULL,
        PRIMARY KEY (`emp_no`)
      ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb3

   2. 使用DMS構建測試資料，詳情請參見測試資料構建。

3. 配置ApsaraDB for SelectDB執行個體。

   1. 建立ApsaraDB for SelectDB執行個體，詳情請參見建立執行個體。

   2. 通過MySQL協議串連ApsaraDB for SelectDB執行個體，詳情請參見串連執行個體。

   3. 建立測試資料庫和測試表。

      1. 建立測試資料庫。

         CREATE DATABASE test_db;

      2. 建立測試表。

         USE test_db;
         CREATE TABLE employees (
             emp_no       INT NOT NULL,
             birth_date   DATE,
             first_name   VARCHAR(20),
             last_name    VARCHAR(20),
             gender       CHAR(2),
             hire_date    DATE
         )
         UNIQUE KEY(`emp_no`)
         DISTRIBUTED BY HASH(`emp_no`) BUCKETS 1;

   4. 開通ApsaraDB for SelectDB公網地址，詳情請參見申請和釋放公網地址。

   5. 將SeaTunnel環境的公網IP添加到IP白名單中，詳情請參見設定白名單。

通過SuaTunnel本地引擎同步MySQL資料到SelectDB

1. 建立設定檔mysqlToSelectDB.conf，配置任務資訊。

   env {
     execution.parallelism = 2
     job.mode = "BATCH"
     checkpoint.interval = 10000
   }
   
   source{
     jdbc {
       url = "jdbc:mysql://host:ip/test_db"
       driver = "com.mysql.cj.jdbc.Driver"
       user = "admin"
       password = "****"
       query = "select * from employees"
     }
   }
    
   sink {
     SelectDBCloud {
       load-url="selectdb-cn-pe33hab****-public.selectdbfe.rds.aliyuncs.com:8080"
       jdbc-url="selectdb-cn-pe33hab****-public.selectdbfe.rds.aliyuncs.com:9030"
       cluster-name="new_cluster"
       table.identifier="test_db.employees"
       username="admin"
       password="****"
       selectdb.config {
           file.type="json"
       }
     }
   }

2. 命令列提交任務。

   sh ./bin/seatunnel.sh --config ./mysqlToSelectDB.conf -e local