ApsaraDB for SelectDB：Gunakan SeaTunnel untuk mengimpor data

Ikhtisar

SeaTunnel adalah platform integrasi data terdistribusi berperforma tinggi yang mudah digunakan dan mendukung sinkronisasi real-time sejumlah besar data. Anda dapat menggunakan SeaTunnel untuk membaca data dari sumber seperti MySQL, Hive, dan Kafka, lalu menulisnya ke ApsaraDB for SelectDB.

Prasyarat

Pastikan SeaTunnel versi 2.3.1 atau lebih baru telah diinstal.

Cara kerjanya

SeaTunnel memungkinkan Anda menulis data upstream dalam format JSON atau CSV ke ApsaraDB for SelectDB. Konfigurasi sink bervariasi tergantung pada format data.

Format 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" 
    } 
  }
}

Format 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" 
    } 
  }
}

Tabel berikut menjelaskan parameter dalam konfigurasi sink.

Contoh

Dalam contoh ini, SeaTunnel digunakan untuk mengimpor data dari database MySQL hulu ke ApsaraDB for SelectDB. Tabel berikut menjelaskan versi perangkat lunak dalam contoh.

Siapkan lingkungan

1. Konfigurasikan SeaTunnel.

   1. Unduh dan ekstrak paket instalasi SeaTunnel. Dalam contoh ini, paket instalasi SeaTunnel apache-seatunnel-2.3.3-bin.tar.gz digunakan.

      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. Ubah file SEATUNNEL_HOME/config/plugin_config. Pertahankan hanya konektor yang diperlukan.

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

   3. Instal plugin konektor SeaTunnel.

      sh bin/install-plugin.sh

   4. Unduh driver MySQL dan letakkan di direktori 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. Buat data yang akan diimpor. Dalam contoh ini, sejumlah kecil data sampel dibuat di database MySQL untuk diimpor.

   1. Buat tabel uji di database 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. Gunakan Data Management (DMS) untuk menghasilkan data uji. Untuk informasi lebih lanjut, lihat Hasilkan data uji.

3. Konfigurasikan instans ApsaraDB for SelectDB.

   1. Buat instans ApsaraDB for SelectDB. Untuk informasi lebih lanjut, lihat Buat instans.

   2. Sambungkan ke instans ApsaraDB for SelectDB melalui protokol MySQL. Untuk informasi lebih lanjut, lihat Sambungkan ke instans.

   3. Buat database uji dan tabel uji.

      1. Buat database uji.

         CREATE DATABASE test_db;

      2. Buat tabel uji.

         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. Ajukan titik akhir publik untuk instans ApsaraDB for SelectDB. Untuk informasi lebih lanjut, lihat Ajukan atau lepaskan titik akhir publik.

   5. Tambahkan alamat IP publik SeaTunnel ke daftar putih alamat IP instans ApsaraDB for SelectDB. Untuk informasi lebih lanjut, lihat Konfigurasikan daftar putih alamat IP.

Gunakan mesin SeaTunnel lokal untuk menyinkronkan data dari database MySQL ke instans ApsaraDB for SelectDB

1. Buat file konfigurasi mysqlToSelectDB.conf dan konfigurasikan informasi pekerjaan.

   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. Kirim pekerjaan.

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