Migrer des données de PostgreSQL vers ApsaraDB for SelectDB - ApsaraDB for SelectDB

Vous pouvez utiliser DTS, DataWorks, Flink CDC et Catalog pour migrer des données depuis des sources PostgreSQL, telles que PostgreSQL auto-géré, ApsaraDB RDS for PostgreSQL et PolarDB for PostgreSQL, vers ApsaraDB for SelectDB. Sélectionnez la méthode de migration adaptée en fonction du volume de données et de votre scénario métier.

Solutions de migration

DTS, DataWorks, Flink CDC et catalog permettent tous de migrer des données de PostgreSQL vers SelectDB, mais les types de données pris en charge varient selon la méthode. Choisissez la solution de migration appropriée à votre scénario métier.

Solution

Migration des données historiques

Synchronisation incrémentielle des données

Migration du schéma

Migration de base de données

Synchronisation incrémentielle des DDL

Vérification des données

DTS

✔️

DataWorks

✔️

❌

Flink CDC

✔️

❌

catalog

✔️

❌

Prérequis

Assurez la connectivité réseau entre l'instance PostgreSQL et l'instance SelectDB.
- L'instance PostgreSQL et l'instance SelectDB doivent se trouver dans le même VPC. Si ce n'est pas le cas, résolvez d'abord le problème de connectivité réseau. Pour plus d'informations, consultez la rubrique Comment résoudre les problèmes de connectivité réseau entre une instance SelectDB et une source de données ?
- L'adresse IP de l'instance PostgreSQL a été ajoutée à la liste d'autorisation d'adresses IP de SelectDB. Pour plus d'informations, consultez la rubrique Définir une liste d'autorisation d'adresses IP.
- Si l'instance PostgreSQL dispose d'une liste d'autorisation d'adresses IP, ajoutez la plage d'adresses IP du segment réseau de l'instance SelectDB à cette liste.
  - Pour obtenir la plage d'adresses IP du VPC où réside votre instance SelectDB, consultez la rubrique Comment trouver le bloc CIDR IP du VPC où se trouve mon instance ApsaraDB for SelectDB ?
  - Pour obtenir l'adresse IP publique de votre instance SelectDB, utilisez la commande ping sur son endpoint public SelectDB.

Procédure

Migrer avec DTS

DTS prend en charge la migration des données historiques et la synchronisation des données incrémentielles de PostgreSQL vers SelectDB. Il offre également des fonctionnalités telles que la migration de schéma, la synchronisation DDL et la vérification des données.

Pour synchroniser des données, consultez les rubriques suivantes :
- Synchroniser des données d'un cluster PolarDB for PostgreSQL vers une instance ApsaraDB for SelectDB
- Synchroniser des données d'une instance ApsaraDB RDS for PostgreSQL vers une instance ApsaraDB for SelectDB
Pour migrer des données, consultez les rubriques suivantes :
- Migrer des données d'un cluster PolarDB for PostgreSQL vers une instance ApsaraDB for SelectDB
- Migrer des données d'une instance ApsaraDB RDS for PostgreSQL vers une instance ApsaraDB for SelectDB

Migrer avec DataWorks

Étape 1 : Ajouter des sources de données

Lorsque vous développez une tâche de synchronisation de données, vous devez créer des sources de données PostgreSQL et SelectDB dans DataWorks.

Créez une source de données PostgreSQL.

Créez une source de données SelectDB. Pour plus d'informations, consultez la rubrique Gestion des sources de données. Voici certains paramètres de configuration pour la source de données SelectDB :

Paramètre	Description
Data source name	Le nom de la source de données.
JDBC URL	Chaîne de connexion JDBC `jdbc:mysql://<ip>:<port>/<dbname>`. Recherchez l'VPC Endpoint (ou l'Public Endpoint) et le MySQL Port sur la page Instance Details > Network Information dans la console SelectDB. Exemple : `jdbc:mysql://selectdb-cn-4xl3jv1****.selectdbfe.rds.aliyuncs.com:9030/test_db`
FE HTTP endpoint	L'URL d'accès HTTP est `<ip>:<port>`. Recherchez l'VPC Endpoint (ou l'Public Endpoint) et le HTTP Port sur la page Instance Details > Network Information dans la console SelectDB. Exemple : `selectdb-cn-4xl3jv1****.selectdbfe.rds.aliyuncs.com:8080`
Username	Le nom d'utilisateur de l'instance SelectDB.
Password	Le mot de passe de l'utilisateur de l'instance SelectDB.

Étape 2 : Configurer la synchronisation des données

Configurez une tâche de synchronisation de données en utilisant l'une des méthodes suivantes :

Migrer avec Flink CDC

Flink utilise Flink CDC pour migrer des données de PostgreSQL vers SelectDB. Flink CDC prend en charge la migration des données historiques et la synchronisation des données incrémentielles, et offre des fonctionnalités complètes telles que la migration de schémas et de tables ainsi que la synchronisation DDL.

Préparer l'environnement

Configurez un environnement Flink. Cet exemple utilise un environnement autonome Flink 1.16.

Téléchargez et extrayez le package flink-1.16.3-bin-scala_2.12.tgz. Pour trouver d'autres versions disponibles, consultez 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
```
Accédez au répertoire FLINK_HOME/lib et téléchargez flink-sql-connector-postgres-cdc-2.4.2 et flink-doris-connector-1.16-1.5.2. Le code d'exemple suivant est fourni à titre de référence.

Remarque
La migration de base de données est prise en charge à partir de Flink 1.15. Pour savoir comment télécharger le connecteur Flink Doris pour différentes versions de Flink, consultez 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
```
Démarrez un cluster Flink autonome. Exemple :
```
bin/start-cluster.sh
```
Créez une instance SelectDB. Pour plus d'informations, consultez la rubrique Créer une instance.
Connectez-vous à une instance SelectDB en utilisant le protocole MySQL. Pour plus d'informations, consultez la rubrique Se connecter à une instance.

Créez une base de données de test et une table de test.

Créez une base de données de test.
```
CREATE DATABASE test_db;
```

Créez une table de test.

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;

Soumettre une tâche Flink CDC

Utilisez la syntaxe suivante pour soumettre la tâche 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=****

Paramètres

Paramètre	Obligatoire	Description
execution.checkpointing.interval	Oui	L'intervalle de checkpoint Flink. Cela affecte la fréquence de synchronisation des données. Une valeur de 10s est recommandée.
parallelism.default	Non	Le parallélisme de la tâche Flink. Vous pouvez augmenter le parallélisme pour améliorer la vitesse de synchronisation des données.
database	Oui	Le nom de la base de données pour la synchronisation vers SelectDB.
including-tables	Non	Les tables PostgreSQL à synchroniser. Vous pouvez utiliser une barre verticale (\|) pour séparer plusieurs noms de tables. Les expressions régulières sont prises en charge. Par exemple, `--including-tables table1\|tbl.*` synchronise table1 et toutes les tables commençant par tbl.
excluding-tables	Non	Les tables à exclure. Ce paramètre se configure de la même manière que including-tables.
postgres-conf	Oui	Configuration de la source Postgres CDC. Pour plus de détails sur la configuration, consultez Connecteur Postgres CDC, où `hostname`, `username`, `password`, `database-name`, `table-name` et `schema-name` sont obligatoires.
sink-conf	Oui	Configurations pour le Sink Doris. Pour plus d'informations, consultez la rubrique Importer des données à l'aide de Flink.
table-conf	Non	Configurations pour la table SelectDB. Il s'agit des propriétés spécifiées lors de la création de la table SelectDB.

Migrer avec un catalog

La fonctionnalité Catalog dans SelectDB vous permet d'utiliser des requêtes fédérées pour accéder à PostgreSQL, facilitant ainsi la migration rapide des données historiques de PostgreSQL vers SelectDB.

Connectez-vous à l'instance SelectDB. Pour plus d'informations, consultez la rubrique Se connecter à une instance.

Remarque
Lorsque vous vous connectez via DMS, la commande SWITCH échoue. Nous vous recommandons de vous connecter à l'aide d'un client MySQL.

Créez un 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"
);

Paramètres

Paramètre

Obligatoire

Description

user

Oui

Le nom d'utilisateur du compte de base de données.

password

Oui

Le mot de passe du compte de base de données.

jdbc_url

Oui

La chaîne de connexion JDBC.

driver_url

Oui

Le nom du fichier JAR du pilote JDBC.

driver_class

Oui

Le nom de classe du pilote JDBC.

lower_case_table_names

Non

Indique s'il faut synchroniser les noms de bases de données et de tables de la source de données JDBC externe en minuscules.

Valeur par défaut : "false"

only_specified_database

Non

Indique s'il faut synchroniser uniquement les bases de données spécifiées.

Valeur par défaut : "false"

include_database_list

Non

Lorsque only_specified_database=true, spécifiez les multiples bases de données à synchroniser, séparées par des virgules (,). Les noms de bases de données sont sensibles à la casse.

Valeur par défaut : ""

exclude_database_list

Non

Lorsque only_specified_database=true, spécifiez les bases de données à synchroniser. Utilisez des virgules (,) pour séparer plusieurs noms de bases de données. Les noms de bases de données sont sensibles à la casse.

Valeur par défaut : ""

Pour plus d'informations, consultez la rubrique Sources de données JDBC.

Après avoir créé une table dans SelectDB, vous pouvez utiliser la syntaxe ETL intégrée insert into select pour synchroniser les données. Pour plus d'informations sur insert into, consultez la rubrique Insert Into.
```
# Create a table.
CREATE TABLE selectdb_table ...

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