Migrasi data dari PostgreSQL ke ApsaraDB for SelectDB - ApsaraDB for SelectDB

Anda dapat menggunakan DTS, DataWorks, Flink CDC, dan Catalog untuk meng migrasi data dari sumber PostgreSQL—seperti PostgreSQL yang dikelola sendiri, ApsaraDB RDS for PostgreSQL, dan PolarDB for PostgreSQL—ke ApsaraDB for SelectDB. Pilih metode migrasi yang sesuai berdasarkan volume data dan skenario bisnis Anda.

Solusi migrasi

DTS, DataWorks, Flink CDC, dan Catalog semuanya dapat digunakan untuk meng migrasi data dari PostgreSQL ke SelectDB, tetapi jenis data yang didukung berbeda-beda untuk setiap metode. Pilih metode migrasi yang tepat sesuai dengan skenario bisnis Anda.

Solusi	Migrasi data historis	Sinkronisasi data inkremental	Migrasi skema	Migrasi database	Sinkronisasi DDL inkremental	Verifikasi data
DTS	✔️	✔️	✔️	✔️	✔️	✔️
DataWorks	✔️	✔️	✔️	✔️	✔️	❌
Flink CDC	✔️	✔️	✔️	✔️	✔️	❌
catalog	✔️	❌	❌	❌	❌	❌

Prasyarat

Pastikan konektivitas jaringan antara instans PostgreSQL dan instans SelectDB.
- Instans PostgreSQL dan instans SelectDB harus berada dalam VPC yang sama. Jika tidak, Anda harus terlebih dahulu menyelesaikan masalah konektivitas jaringan. Untuk informasi selengkapnya, lihat Cara menyelesaikan masalah konektivitas jaringan antara instans SelectDB dan sumber data?
- Alamat IP instans PostgreSQL telah ditambahkan ke daftar putih alamat IP SelectDB. Untuk informasi selengkapnya, lihat Setel daftar putih alamat IP.
- Jika instans PostgreSQL memiliki daftar putih alamat IP, rentang alamat IP segmen jaringan untuk instans SelectDB telah ditambahkan ke daftar putih alamat IP instans PostgreSQL.
  - Untuk mendapatkan rentang IP VPC tempat instans SelectDB Anda berada, lihat Bagaimana cara menemukan blok CIDR IP VPC tempat instans ApsaraDB for SelectDB saya berada?
  - Untuk mendapatkan alamat IP publik instans SelectDB Anda, gunakan perintah ping pada titik akhir publikSelectDB-nya.

Prosedur

Migrasi dengan DTS

DTS mendukung migrasi data historis dan sinkronisasi data inkremental dari PostgreSQL ke SelectDB. Fitur ini juga mendukung migrasi skema, sinkronisasi DDL, dan verifikasi data.

Untuk menyinkronkan data, lihat topik-topik berikut:
- Sinkronisasi data dari kluster PolarDB for PostgreSQL ke instans ApsaraDB for SelectDB
- Sinkronisasi data dari instans ApsaraDB RDS for PostgreSQL ke instans ApsaraDB for SelectDB
Untuk meng migrasi data, lihat topik-topik berikut:
- Migrasi data dari kluster PolarDB for PostgreSQL ke instans ApsaraDB for SelectDB
- Migrasi data dari instans ApsaraDB RDS for PostgreSQL ke instans ApsaraDB for SelectDB

Migrasi dengan DataWorks

Langkah 1: Tambahkan sumber data

Saat mengembangkan tugas sinkronisasi data, Anda perlu membuat sumber data PostgreSQL dan SelectDB di DataWorks.

Buat sumber data PostgreSQL.

Buat sumber data SelectDB. Untuk informasi selengkapnya, lihat Manajemen Sumber Data. Beberapa parameter konfigurasi untuk sumber data SelectDB adalah sebagai berikut:

Parameter	Deskripsi
Data source name	Nama sumber data.
JDBC URL	String koneksi JDBC `jdbc:mysql://<ip>:<port>/<dbname>`. Temukan VPC Endpoint (atau Public Endpoint) dan MySQL Port pada halaman Instance Details > Network Information di konsol SelectDB. Contoh: `jdbc:mysql://selectdb-cn-4xl3jv1****.selectdbfe.rds.aliyuncs.com:9030/test_db`
FE HTTP endpoint	URL akses HTTP adalah `<ip>:<port>`. Temukan VPC Endpoint (atau Public Endpoint) dan HTTP Port pada halaman Instance Details > Network Information di konsol SelectDB. Contoh: `selectdb-cn-4xl3jv1****.selectdbfe.rds.aliyuncs.com:8080`
Username	Username instans SelectDB.
Password	Password pengguna instans SelectDB.

Langkah 2: Konfigurasi sinkronisasi data

Konfigurasikan tugas sinkronisasi data dengan menggunakan salah satu metode berikut:

Migrasi dengan Flink CDC

Flink menggunakan Flink CDC untuk meng migrasi data dari PostgreSQL ke SelectDB. Flink CDC mendukung migrasi data historis dan sinkronisasi data inkremental, serta menyediakan kemampuan komprehensif seperti migrasi skema dan tabel serta sinkronisasi DDL.

Persiapkan lingkungan

Siapkan lingkungan Flink. Contoh ini menggunakan lingkungan standalone Flink 1.16.

Unduh dan ekstrak paket flink-1.16.3-bin-scala_2.12.tgz. Untuk mencari versi lain yang tersedia, lihat Apache Flink.

wget https://archive.apache.org/dist/flink/flink-1.16.3/flink-1.16.3-bin-scala_2.12.tgz
tar -zxvf flink-1.16.3-bin-scala_2.12.tgz

Buka direktori FLINK_HOME/lib dan unduh flink-sql-connector-postgres-cdc-2.4.2 dan flink-doris-connector-1.16-1.5.2. Kode contoh berikut disediakan sebagai referensi.
Catatan
Migrasi database didukung di Flink 1.15 dan versi lebih baru. Untuk informasi tentang cara mengunduh Flink Doris Connector untuk versi Flink yang berbeda, lihat Flink Doris Connector.
```
cd flink-1.16.3
cd lib/
wget https://repo.maven.apache.org/maven2/org/apache/flink/flink-sql-connector-postgres-cdc/2.4.2/flink-sql-connector-postgres-cdc-2.4.2.jar
wget https://repo.maven.apache.org/maven2/org/apache/doris/flink-doris-connector-1.16/1.5.2/flink-doris-connector-1.16-1.5.2.jar
```
Jalankan kluster standalone Flink. Contoh:
```
bin/start-cluster.sh
```
Buat instans SelectDB. Untuk informasi selengkapnya, lihat Buat instans.
Hubungkan ke instans SelectDB menggunakan protokol MySQL. Untuk informasi selengkapnya, lihat Hubungkan ke instans.

Buat database uji dan tabel uji.

Buat database uji.
```
CREATE DATABASE test_db;
```

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;

Kirim pekerjaan Flink CDC

Gunakan sintaks berikut untuk mengirim pekerjaan Flink CDC:

<FLINK_HOME>/bin/flink run \
    -Dexecution.checkpointing.interval=10s \
    -Dparallelism.default=1 \
    -c org.apache.doris.flink.tools.cdc.CdcTools \
    lib/flink-doris-connector-1.16-1.5.2.jar \
    postgres-sync-database \
    --database db1\
    --postgres-conf hostname=127.0.0.1 \
    --postgres-conf port=5432 \
    --postgres-conf username=postgres \
    --postgres-conf password="123456" \
    --postgres-conf database-name=postgres \
    --postgres-conf schema-name=public \
    --postgres-conf slot.name=test \
    --postgres-conf decoding.plugin.name=pgoutput \
    --including-tables "tbl1|test.*" \
    --sink-conf fenodes=selectdb-cn-****.selectdbfe.rds.aliyuncs.com:8080 \
    --sink-conf username=admin \
    --sink-conf password=****

Parameter

Parameter	Wajib	Deskripsi
execution.checkpointing.interval	Ya	Interval checkpoint Flink. Ini memengaruhi frekuensi sinkronisasi data. Nilai 10s direkomendasikan.
parallelism.default	Tidak	Paralelisme pekerjaan Flink. Anda dapat meningkatkan paralelisme untuk mempercepat sinkronisasi data.
database	Ya	Nama database untuk sinkronisasi ke SelectDB.
including-tables	Tidak	Tabel PostgreSQL yang akan disinkronkan. Anda dapat menggunakan tanda pipa (\|) untuk memisahkan beberapa nama tabel. Ekspresi reguler didukung. Contohnya, `--including-tables table1\|tbl.*` menyinkronkan table1 dan semua tabel yang diawali dengan tbl.
excluding-tables	Tidak	Tabel yang dikecualikan. Parameter ini dikonfigurasi dengan cara yang sama seperti including-tables.
postgres-conf	Ya	Konfigurasi Postgres CDC Source. Untuk detail konfigurasi, lihat Postgres CDC Connector, di mana `hostname`, `username`, `password`, `database-name`, `table-name`, dan `schema-name` wajib diisi.
sink-conf	Ya	Konfigurasi untuk Doris Sink. Untuk informasi selengkapnya, lihat Impor data menggunakan Flink.
table-conf	Tidak	Konfigurasi untuk tabel SelectDB. Ini adalah properti yang ditentukan saat Anda membuat tabel SelectDB.

Migrasi dengan katalog

Fitur Catalog di SelectDB memungkinkan Anda menggunakan kueri federasi untuk mengakses PostgreSQL, sehingga memudahkan dan mempercepat migrasi data historis dari PostgreSQL ke SelectDB.

Hubungkan ke instans SelectDB. Untuk informasi selengkapnya, lihat Hubungkan ke instans.
Catatan
Saat login menggunakan DMS, perintah SWITCH gagal. Disarankan untuk menghubungkan menggunakan client MySQL.

Buat catalog JDBC PostgreSQL.

CREATE CATALOG jdbc_postgresql PROPERTIES (
    "type"="jdbc",
    "user"="root",
    "password"="123456",
    "jdbc_url" = "jdbc:postgresql://127.0.0.1:5432/demo",
    "driver_url" = "postgresql-42.5.1.jar",
    "driver_class" = "org.postgresql.Driver"
);

Parameter

Parameter	Wajib	Deskripsi
user	Ya	Username untuk akun database.
password	Ya	Password untuk akun database.
jdbc_url	Ya	String koneksi JDBC.
driver_url	Ya	Nama file JAR driver JDBC.
driver_class	Ya	Nama kelas driver JDBC.
lower_case_table_names	Tidak	Menentukan apakah nama database dan tabel dari sumber data JDBC eksternal disinkronkan dalam huruf kecil. Nilai default: `"false"`
only_specified_database	Tidak	Menentukan apakah hanya database yang ditentukan yang disinkronkan. Nilai default: `"false"`
include_database_list	Tidak	Saat `only_specified_database=true`, tentukan beberapa database yang akan disinkronkan, dipisahkan dengan koma (,). Nama database bersifat case-sensitive. Nilai default: `""`
exclude_database_list	Tidak	Saat `only_specified_database=true`, tentukan database yang akan disinkronkan. Gunakan koma (,) untuk memisahkan beberapa nama database. Nama database bersifat case-sensitive. Nilai default: `""`

Untuk informasi selengkapnya, lihat Sumber data JDBC.

Setelah membuat tabel di SelectDB, Anda dapat menggunakan sintaks ETL dalam bentuk pernyataan INSERT INTO SELECT untuk menyinkronkan data. Untuk informasi selengkapnya tentang INSERT INTO, lihat Insert Into.
```
# Buat tabel.
CREATE TABLE selectdb_table ...

# Migrasi data.
INSERT INTO selectdb_table SELECT * FROM jdbc_postgresql.pg_database.pg_table;
```