Impor Data ke Instans ApsaraDB for SelectDB Menggunakan DataX - ApsaraDB for SelectDB

Topik ini menjelaskan cara menggunakan DataX Doris Writer untuk menyinkronkan data ke ApsaraDB for SelectDB.

Ikhtisar

DataX adalah alat open source yang disediakan oleh Alibaba Group untuk sinkronisasi offline antara sumber data berbeda, termasuk database relasional seperti MySQL dan Oracle, Sistem File Terdistribusi Hadoop (HDFS), Hive, MaxCompute, HBase, dan FTP. DataX menyediakan kemampuan sinkronisasi data yang stabil dan efisien. Anda dapat menggunakan DataX untuk membaca data dari sumber hulu dan menulisnya ke ApsaraDB for SelectDB menggunakan DataX Doris Writer.

Prasyarat

Lingkungan Maven telah diinstal.
Python 3.6 atau versi lebih baru telah diinstal.

Contoh

Contoh berikut menunjukkan cara mengimpor data dari database MySQL ke ApsaraDB for SelectDB menggunakan DataX di Linux.

Langkah 1: Konfigurasi lingkungan DataX

Unduh paket DataX. Untuk informasi lebih lanjut, kunjungi doris di GitHub.
Jalankan skrip init-env.sh dalam paket DataX untuk mengonfigurasi lingkungan pengembangan DataX.
```
sh init-env.sh
```
Kompilasi plug-in mysqlreader dan doriswriter.
1. Kompilasi proyek DataX.
```
cd DataX/
mvn package assembly:assembly -Dmaven.test.skip=true
```
  Hasil kompilasi disimpan di direktori target/datax/datax/.
  Catatan
  Jika Anda ingin menggunakan plug-in hdfsreader, hdfswriter, ossreader, dan osswriter, diperlukan paket JAR tambahan. Jika tidak memerlukan plug-in ini, Anda dapat menghapus modul-modul tersebut dari file DataX/pom.xml.
2. Kompilasi secara terpisah plug-in mysqlreader dan selectdbwriter.
```
mvn clean install -pl plugin-rdbms-util,mysqlreader,doriswriter -DskipTests
```

Langkah 2: Hasilkan data yang akan diimpor

Buat tabel uji 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

Gunakan Data Management (DMS) untuk menghasilkan data uji. Untuk informasi lebih lanjut, lihat Hasilkan Data Uji.

Langkah 3: Konfigurasikan sebuah ApsaraDB for SelectDB instance

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

Buat database uji dan tabel uji.

Jalankan pernyataan berikut untuk membuat database uji:
```
CREATE DATABASE test_db;
```

Jalankan pernyataan berikut untuk membuat 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;

Langkah 4: Sinkronkan data dari database MySQL ke instans ApsaraDB for SelectDB menggunakan DataX

Ajukan titik akhir publik untuk ApsaraDB for SelectDB. Untuk informasi lebih lanjut, lihat Ajukan atau Lepaskan Titik Akhir Publik.
Tambahkan alamat IP publik host DataX ke daftar putih alamat IP instans. Untuk informasi lebih lanjut, lihat Konfigurasikan Daftar Putih Alamat IP.

Buat file konfigurasi mysqlToSelectDB.json dan konfigurasikan informasi pekerjaan impor di dalam file. Contoh kode:

{
  "job":{
    "content":[
      {
        "reader":{
            "name": "mysqlreader",
            "parameter": {
                "column": [
                    "emp_no",
                    "birth_date",
                    "first_name",
                    "last_name",
                    "gender",
                    "hire_date"
                ],
                "where": "emp_no>0",
                "connection": [
                    {
                        "jdbcUrl": [
                            "jdbc:mysql://host:port/test_db?useUnicode=true&allowPublicKeyRetrieval=true&characterEncoding=utf-8"
                        ],
                        "table": [
                            "employees"
                        ]
                    }
                ],
                "password": "123456",
                "splitPk": "emp_no",
                "username": "admin"
            }
        },
        "writer":{
          "name":"doriswriter",
          "parameter":{
            "loadUrl":[
              "selectdb-cn-xxx-public.selectdbfe.rds.aliyuncs.com:8080"
            ],
            "loadProps":{
              "format":"json",
              "strip_outer_array":"true"
            },
            "column":[
                "emp_no",
                "birth_date",
                "first_name",
                "last_name",
                "gender",
                "hire_date"
            ],
            "username":"admin",
            "password":"123456",
            "postSql":[

            ],
            "preSql":[

            ],
            "connection":[
              {
                "jdbcUrl":"jdbc:mysql://selectdb-cn-xxx-public.selectdbfe.rds.aliyuncs.com:9030/test_db",
                "table":[
                  "employees"
                ],
                "selectedDatabase":"test_db"
              }
            ],
            "maxBatchRows":1000000,
            "batchSize":536870912000
          }
        }
      }
    ],
    "setting":{
      "errorLimit":{
        "percentage":0.02,
        "record":0
      },
      "speed":{
        "channel":5
      }
    }
  }
}

Parameter

Parameter	Diperlukan	Nilai default	Deskripsi
jdbcUrl	Ya	Tidak tersedia	URL Java Database Connectivity (JDBC) yang digunakan untuk menyambungkan ke instans ApsaraDB for SelectDB, dalam format `jdbc:mysql://<ip>:<port>`. Untuk mendapatkan titik akhir instans ApsaraDB for SelectDB, lakukan operasi berikut: Masuk ke ApsaraDB for SelectDB dan buka halaman Instance Details. Di halaman Informasi Dasar, lihat titik akhir dan nomor port MySQL di bagian Network Information. ip: titik akhir virtual private cloud (VPC) atau titik akhir publik instans. port: nomor port MySQL instans. Contoh: `jdbc:mysql://selectdb-cn-4xl3jv1**.selectdbfe.rds.aliyuncs.com:9030`. Catatan** Jika host DataX dan instans ApsaraDB for SelectDB berada di VPC yang sama, gunakan titik akhir VPC instans. Jika host DataX dan instans ApsaraDB for SelectDB berada di VPC yang berbeda, gunakan titik akhir publik instans.
loadUrl	Ya	Tidak tersedia	URL yang digunakan untuk menyambungkan ke instans ApsaraDB for SelectDB melalui HTTP, dalam format `<ip>:<port>`. Untuk mendapatkan titik akhir instans ApsaraDB for SelectDB, lakukan operasi berikut: Masuk ke ApsaraDB for SelectDB dan buka halaman Instance Details. Di halaman Informasi Dasar, lihat titik akhir dan nomor port HTTP di bagian Network Information. ip: titik akhir VPC atau titik akhir publik instans. port: nomor port HTTP instans. Contoh: `selectdb-cn-4xl3jv1**.selectdbfe.rds.aliyuncs.com:8080`. Catatan** Jika host DataX dan instans ApsaraDB for SelectDB berada di VPC yang sama, gunakan titik akhir VPC instans. Jika host DataX dan instans ApsaraDB for SelectDB berada di VPC yang berbeda, gunakan titik akhir publik instans.
username	Ya	Tidak tersedia	Nama pengguna yang digunakan untuk menyambungkan ke ApsaraDB for SelectDB.
password	Ya	Tidak tersedia	Kata sandi nama pengguna yang digunakan untuk menyambungkan ke ApsaraDB for SelectDB.
connection.selectedDatabase	Ya	Tidak tersedia	Nama database tempat Anda ingin menulis data di ApsaraDB for SelectDB.
connection.table	Ya	Tidak tersedia	Nama tabel tempat Anda ingin menulis data di ApsaraDB for SelectDB.
column	Ya	Tidak tersedia	Bidang tempat Anda ingin menulis data di tabel tujuan. Bidang ini digunakan dalam data berformat JSON yang dihasilkan. Pisahkan beberapa bidang dengan koma (,). Contoh: `"column": ["id","name","age"]`.
preSql	Tidak	Tidak tersedia	Pernyataan standar yang akan dieksekusi sebelum data ditulis ke tabel tujuan.
postSql	Tidak	Tidak tersedia	Pernyataan standar yang akan dieksekusi setelah data ditulis ke tabel tujuan.
maxBatchRows	Tidak	500000	Jumlah maksimum baris yang dapat ditulis ke tabel tujuan dalam satu batch. Parameter ini digunakan bersama dengan parameter batchSize untuk menentukan jumlah data yang dapat ditulis dalam satu batch. Ketika satu batch data mencapai salah satu dari dua ambang batas, data batch ini mulai diimpor.
batchSize	Tidak	104857600	Jumlah maksimum data yang dapat ditulis ke tabel tujuan dalam satu batch. Parameter ini digunakan bersama dengan parameter maxBatchRows untuk menentukan jumlah data yang dapat ditulis dalam satu batch. Ketika satu batch data mencapai salah satu dari dua ambang batas, data batch ini mulai diimpor. Nilai default: 100 MB.
maxRetries	Tidak	3	Jumlah maksimum percobaan ulang yang diizinkan setelah gagal mengimpor satu batch data ke tabel tujuan.
labelPrefix	Tidak	datax_doris_writer_	Awalan label setiap batch data yang diimpor. Label unik terdiri dari awalan label dan UUID untuk memastikan bahwa data tidak diimpor berulang kali.
loadProps	Tidak	Tidak tersedia	Parameter permintaan yang sama dengan pekerjaan Stream Load. Untuk informasi lebih lanjut, lihat bagian Parameter topik "Impor data menggunakan Stream Load". Parameter format menentukan format data yang akan diimpor. Secara default, data yang akan diimpor berada dalam format CSV. Anda dapat mengimpor data dalam format JSON. Untuk informasi lebih lanjut, lihat bagian Konversi tipe data topik ini.
flushInterval	Tidak	30000	Interval penulisan data dalam batch. Nilai default: 30000. Unit: milidetik.

Jalankan perintah berikut untuk mengirimkan pekerjaan:

cd target/datax/datax/bin
python datax.py ../mysqlToSelectDB.json

Konversi tipe data

Secara default, data yang diimpor dikonversi menjadi string. \t digunakan sebagai pemisah kolom dan \n digunakan sebagai pemisah baris untuk mengonversi data menjadi file CSV yang akan diimpor ke instans ApsaraDB for SelectDB. Secara default, data diimpor dalam format CSV. Jika ingin mengubah pemisah kolom atau baris, modifikasi parameter dari parameter loadProps. Contoh kode:

"loadProps": {
    "format": "csv",
    "column_separator": "\\x01",
    "line_delimiter": "\\x02"
}

Jika ingin mengimpor data dalam format JSON, modifikasi parameter format dari parameter loadProps. Contoh kode:

"loadProps": {
    "format": "json",
    "strip_outer_array": true
}