Migrate data from PostgreSQL to ApsaraDB for SelectDB - ApsaraDB for SelectDB

DTS, DataWorks, Flink CDC ve Catalog kullanarak kendi kendine yönetilen PostgreSQL, ApsaraDB RDS for PostgreSQL ve PolarDB for PostgreSQL gibi PostgreSQL kaynaklarından ApsaraDB for SelectDB örneğine veri geçirebilirsiniz. Veri hacmine ve iş senaryonuzun gereksinimlerine bağlı olarak uygun bir geçiş yöntemi seçebilirsiniz.

Geçiş çözümleri

DTS, DataWorks, Flink CDC ve catalog araçları PostgreSQL'den SelectDB örneğine veri aktarmak için kullanılabilir; ancak her yöntemin desteklediği veri türleri farklıdır. İş senaryonuza en uygun geçiş yöntemini belirleyebilirsiniz.

Çözüm	Geçmiş veri geçişi	Artımlı veri senkronizasyonu	Şema geçişi	Veritabanı geçişi	Artımlı DDL senkronizasyonu	Veri doğrulama
DTS	✔️	✔️	✔️	✔️	✔️	✔️
DataWorks	✔️	✔️	✔️	✔️	✔️	❌
Flink CDC	✔️	✔️	✔️	✔️	✔️	❌
catalog	✔️	❌	❌	❌	❌	❌

Ön koşullar

PostgreSQL örneği ile SelectDB örneği arasında ağ bağlantısının sağlandığından emin olun.
- PostgreSQL örneği ve SelectDB örneği aynı VPC içinde olmalıdır. Aksi takdirde öncelikle ağ bağlantısı sorununu çözmeniz gerekir. Daha fazla bilgi için bkz. SelectDB örneği ile veri kaynağı arasındaki ağ bağlantısı sorunlarını nasıl çözebilirim?
- PostgreSQL örneğinin IP adresini SelectDB IP adresi izin listesine ekleyin. Daha fazla bilgi için bkz. IP adresi izin listesi ayarlama.
- PostgreSQL örneğinde bir IP adresi izin listesi yapılandırılmışsa, SelectDB örneğinin bulunduğu ağ segmentinin IP adres aralığını PostgreSQL örneğinin izin listesine ekleyin.
  - SelectDB örneğinizin bulunduğu VPC'nin IP aralığını öğrenmek için bkz. ApsaraDB for SelectDB örneğimin bulunduğu VPC'nin IP CIDR bloğunu nasıl bulabilirim?
  - SelectDB örneğinizin genel IP adresini öğrenmek için uç noktasında ping komutunu çalıştırın SelectDB.

Yordam

DTS ile geçiş yapma

DTS, PostgreSQL'den SelectDB örneğine geçmiş veri geçişini ve artımlı veri senkronizasyonunu destekler. Ayrıca şema geçişi, DDL senkronizasyonu ve veri doğrulama gibi özellikler sunar.

Veri senkronizasyonu yapmak için aşağıdaki konulara bakın:
- PolarDB for PostgreSQL kümesinden ApsaraDB for SelectDB örneğine veri senkronize etme
- ApsaraDB RDS for PostgreSQL örneğinden ApsaraDB for SelectDB örneğine veri senkronize etme
Veri geçişi yapmak için aşağıdaki konulara bakın:
- PolarDB for PostgreSQL kümesinden ApsaraDB for SelectDB örneğine veri geçirme
- ApsaraDB RDS for PostgreSQL örneğinden ApsaraDB for SelectDB örneğine veri geçirme

DataWorks ile geçiş yapma

Adım 1: Veri kaynaklarını ekleme

Bir veri senkronizasyon görevi geliştirirken DataWorks'te PostgreSQL ve SelectDB veri kaynaklarını oluşturmanız gerekir.

Bir PostgreSQL veri kaynağı oluşturun.

Bir SelectDB veri kaynağı oluşturun. Daha fazla bilgi için bkz. Veri Kaynağı Yönetimi. SelectDB veri kaynağı için bazı yapılandırma parametreleri aşağıda verilmiştir:

Parametre	Açıklama
Data source name	Veri kaynağının adı.
JDBC URL	JDBC bağlantı dizesi `jdbc:mysql://<ip>:<port>/<dbname>`. SelectDB konsolunda Instance Details > Network Information sayfasından VPC Endpointnı (veya Public Endpointyı) ve MySQL Portnı bulun. Örnek: `jdbc:mysql://selectdb-cn-4xl3jv1****.selectdbfe.rds.aliyuncs.com:9030/test_db`
FE HTTP endpoint	HTTP erişim URL'si `<ip>:<port>` biçimindedir. SelectDB konsolunda Instance Details > Network Information sayfasından VPC Endpointnı (veya Public Endpointyı) ve HTTP Portnı bulun. Örnek: `selectdb-cn-4xl3jv1****.selectdbfe.rds.aliyuncs.com:8080`
Username	SelectDB örneğinin kullanıcı adı.
Password	SelectDB örneği kullanıcısının parolası.

Adım 2: Veri senkronizasyonunu yapılandırma

Aşağıdaki yöntemlerden birini kullanarak veri senkronizasyon görevini yapılandırın:

Flink CDC ile geçiş yapma

Flink, PostgreSQL'den SelectDB örneğine veri aktarmak için Flink CDC'yi kullanır. Flink CDC hem geçmiş veri geçişini hem de artımlı veri senkronizasyonunu destekler; şema ve tablo geçişi ile DDL senkronizasyonu gibi kapsamlı yetenekler sağlar.

Ortamı hazırlama

Bir Flink ortamı kurun. Bu örnekte Flink 1.16 bağımsız (standalone) ortamı kullanılmaktadır.

flink-1.16.3-bin-scala_2.12.tgz paketini indirin ve açın. Diğer mevcut sürümleri bulmak için bkz. 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

FLINK_HOME/lib dizinine gidin ve flink-sql-connector-postgres-cdc-2.4.2 ile flink-doris-connector-1.16-1.5.2 dosyalarını indirin. Aşağıdaki örnek kod referans amaçlıdır.

Not
Veritabanı geçişi Flink 1.15 ve sonraki sürümlerde desteklenir. Farklı Flink sürümleri için Flink Doris Connector'u nasıl indireceğinizi öğrenmek için bkz. 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
```
Bağımsız bir Flink kümesini başlatın. Örnek:
```
bin/start-cluster.sh
```
Bir SelectDB örneği oluşturun. Daha fazla bilgi için bkz. Örnek oluşturma.
MySQL protokolünü kullanarak bir SelectDB örneğine bağlanın. Daha fazla bilgi için bkz. Örneğe bağlanma.

Test veritabanı ve test tablosu oluşturun.

Test veritabanı oluşturun.
```
CREATE DATABASE test_db;
```

Test tablosu oluşturun.

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;

Flink CDC işini gönderme

Flink CDC işini göndermek için aşağıdaki sözdizimini kullanın:

<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=****

Parametreler

Parametre	Gerekli	Açıklama
execution.checkpointing.interval	Evet	Flink kontrol noktası aralığı. Bu değer veri senkronizasyon sıklığını etkiler. 10s değeri önerilir.
parallelism.default	Hayır	Flink işinin paralellik derecesi. Veri senkronizasyon hızını artırmak için paralellik değerini yükseltebilirsiniz.
database	Evet	SelectDB üzerinde senkronize edilecek veritabanı adı.
including-tables	Hayır	Senkronize edilecek PostgreSQL tabloları. Birden çok tablo adını ayırmak için dikey çizgi (\|) kullanabilirsiniz. Normal ifadeler desteklenir. Örneğin, `--including-tables table1\|tbl.*` ifadesi table1 tablosunu ve tbl ile başlayan tüm tabloları senkronize eder.
excluding-tables	Hayır	Hariç tutulacak tablolar. Bu parametre including-tables parametresiyle aynı şekilde yapılandırılır.
postgres-conf	Evet	Postgres CDC Source yapılandırması. Yapılandırma ayrıntıları için bkz. Postgres CDC Connector; burada `hostname`, `username`, `password`, `database-name`, `table-name` ve `schema-name` alanları zorunludur.
sink-conf	Evet	Doris Sink yapılandırmaları. Daha fazla bilgi için bkz. Flink kullanarak veri içe aktarma.
table-conf	Hayır	SelectDB tablosu yapılandırmaları. Bunlar, SelectDB tablosu oluştururken belirtilen özelliklerdir.

Catalog kullanarak geçiş yapma

SelectDB içindeki Catalog özelliği, PostgreSQL'e federatif sorgularla erişmenizi sağlar. Bu sayede PostgreSQL'den SelectDB'ye geçmiş veri geçişini kolay ve hızlı bir şekilde gerçekleştirebilirsiniz.

SelectDB örneğine bağlanın. Daha fazla bilgi için bkz. Örneğe bağlanma.

Not
DMS kullanarak oturum açtığınızda SWITCH komutu başarısız olur. MySQL istemcisi kullanarak bağlanmanızı öneririz.

PostgreSQL JDBC kataloğu oluşturun.

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"
);

Parametreler

Parametre	Gerekli	Açıklama
user	Evet	Veritabanı hesabının kullanıcı adı.
password	Evet	Veritabanı hesabının parolası.
jdbc_url	Evet	JDBC bağlantı dizesi.
driver_url	Evet	JDBC sürücüsünün JAR dosya adı.
driver_class	Evet	JDBC sürücüsünün sınıf adı.
lower_case_table_names	Hayır	Harici JDBC veri kaynağından gelen veritabanı ve tablo adlarının küçük harfle senkronize edilip edilmeyeceğini belirtir. Varsayılan değer: `"false"`
only_specified_database	Hayır	Yalnızca belirtilen veritabanlarının senkronize edilip edilmeyeceğini belirtir. Varsayılan değer: `"false"`
include_database_list	Hayır	`only_specified_database=true` olduğunda, virgülle (,) ayrılmış şekilde senkronize edilecek birden çok veritabanını belirtin. Veritabanı adları büyük/küçük harfe duyarlıdır. Varsayılan değer: `""`
exclude_database_list	Hayır	`only_specified_database=true` olduğunda, senkronize edilecek veritabanlarını belirtin. Birden çok veritabanı adını ayırmak için virgül (,) kullanın. Veritabanı adları büyük/küçük harfe duyarlıdır. Varsayılan değer: `""`

Daha fazla bilgi için bkz. JDBC veri kaynakları.

SelectDB'de tablo oluşturduktan sonra veritabanı içi ETL sözdizimi olan insert into select kullanarak verileri senkronize edebilirsiniz. insert into hakkında daha fazla bilgi için bkz. Insert Into.
```
# Create a table.
CREATE TABLE selectdb_table ...

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