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

Puede usar DTS, DataWorks, Flink CDC o un catalog para migrar and Catalog to migrate datos desde orígenes PostgreSQL, como una instancia de PostgreSQL autogestionada, ApsaraDB RDS for PostgreSQL y PolarDB for PostgreSQL, hacia ApsaraDB for SelectDB. Elija el método de migración adecuado según el volumen de datos y su escenario de negocio.

Soluciones de migración

DTS, DataWorks, Flink CDC y un catalog permiten migrar datos de PostgreSQL a SelectDB, pero los datos compatibles varían en cada método. Elija el método de migración adecuado según su escenario de negocio.

Solución	Migración de datos históricos	Sincronización de datos incrementales	Migración de esquemas	Migración de bases de datos	Sincronización incremental de DDL	Verificación de datos
DTS	✔️	✔️	✔️	✔️	✔️	✔️
DataWorks	✔️	✔️	✔️	✔️	✔️	❌
Flink CDC	✔️	✔️	✔️	✔️	✔️	❌
catalog	✔️	❌	❌	❌	❌	❌

Requisitos previos

Asegúrese de que exista conectividad de red entre la instancia de PostgreSQL y la instancia de SelectDB.
- La instancia de PostgreSQL y la instancia de SelectDB deben estar en la misma VPC. Si no lo están, resuelva primero el problema de conectividad de red. Para obtener más información, consulte How to resolve network connectivity issues between a SelectDB instance and a data source?
- La dirección IP de la instancia de PostgreSQL debe estar agregada a la lista blanca de direcciones IP de SelectDB. Para obtener más información, consulte Set an IP address whitelist.
- Si la instancia de PostgreSQL tiene una lista blanca de direcciones IP, agregue el rango de direcciones IP del segmento de red de la instancia de SelectDB a dicha lista blanca.
  - Para obtener el rango de direcciones IP de la VPC donde reside su instancia de SelectDB, consulte How do I find the IP CIDR block of the VPC where my ApsaraDB for SelectDB instance is located?
  - Para obtener la dirección IP pública de su instancia de SelectDB, ejecute el comando ping en su endpoint público de SelectDB.

Procedimiento

Migrar con DTS

DTS permite migrar datos históricos y sincronizar datos incrementales de PostgreSQL a SelectDB. También es compatible con funciones como la migración de esquemas, la sincronización de DDL y la verificación de datos.

Para sincronizar datos, consulte los siguientes temas:
- Synchronize data from a PolarDB for PostgreSQL cluster to an ApsaraDB for SelectDB instance
- Synchronize data from an ApsaraDB RDS for PostgreSQL instance to an ApsaraDB for SelectDB instance
Para migrar datos, consulte los siguientes temas:
- Migrate data from a PolarDB for PostgreSQL cluster to an ApsaraDB for SelectDB instance
- Migrate data from an ApsaraDB RDS for PostgreSQL instance to an ApsaraDB for SelectDB instance

Migrar con DataWorks

Paso 1: Agregar orígenes de datos

Cuando desarrolle una tarea de sincronización de datos, cree orígenes de datos de PostgreSQL y de SelectDB en DataWorks.

Cree un PostgreSQL data source.

Cree un origen de datos de SelectDB. Para obtener más información, consulte Data Source Management. A continuación se muestran algunos parámetros de configuración del origen de datos de SelectDB:

Parámetro	Descripción
Data source name	El nombre del origen de datos.
JDBC URL	Cadena de conexión JDBC `jdbc:mysql://<ip>:<port>/<dbname>`. Busque el VPC Endpoint (o Public Endpoint) y el MySQL Port en la página Instance Details > Network Information de la consola de SelectDB. Ejemplo: `jdbc:mysql://selectdb-cn-4xl3jv1****.selectdbfe.rds.aliyuncs.com:9030/test_db`
FE HTTP endpoint	La URL de acceso HTTP es `<ip>:<port>`. Busque el VPC Endpoint (o Public Endpoint) y el HTTP Port en la página Instance Details > Network Information de la consola de SelectDB. Ejemplo: `selectdb-cn-4xl3jv1****.selectdbfe.rds.aliyuncs.com:8080`
Username	El nombre de usuario de la instancia de SelectDB.
Password	La contraseña del usuario de la instancia de SelectDB.

Paso 2: Configurar la sincronización de datos

Configure una tarea de sincronización de datos mediante uno de los siguientes métodos:

Migrar con Flink CDC

Flink utiliza Flink CDC para migrar datos de PostgreSQL a SelectDB. Flink CDC es compatible tanto con la migración de datos históricos como con la sincronización de datos incrementales. Además, ofrece capacidades integrales como la migración de esquemas y tablas, así como la sincronización de DDL.

Preparar el entorno

Configure un entorno de Flink. Este ejemplo usa un entorno independiente de Flink 1.16.

Descargue y extraiga el paquete flink-1.16.3-bin-scala_2.12.tgz. Para consultar otras versiones disponibles, visite 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

Vaya al directorio FLINK_HOME/lib y descargue flink-sql-connector-postgres-cdc-2.4.2 y flink-doris-connector-1.16-1.5.2. El siguiente código de ejemplo le servirá como referencia.

Nota
La migración de bases de datos es compatible con Flink 1.15 y versiones posteriores. Para saber cómo descargar Flink Doris Connector para distintas versiones de Flink, consulte 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
```
Inicie un clúster independiente de Flink. Ejemplo:
```
bin/start-cluster.sh
```
Cree una instancia de SelectDB. Para obtener más información, consulte Create an instance.
Conéctese a una instancia de SelectDB mediante el protocolo MySQL. Para obtener más información, consulte Connect to an instance.

Cree una base de datos y una tabla de prueba.

Cree una base de datos de prueba.
```
CREATE DATABASE test_db;
```

Cree una tabla de prueba.

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;

Enviar un trabajo de Flink CDC

Utilice la siguiente sintaxis para enviar el trabajo de 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=****

Parámetros

Parámetro	Obligatorio	Descripción
execution.checkpointing.interval	Sí	El intervalo de checkpoint de Flink. Este valor afecta la frecuencia de sincronización de datos. Se recomienda establecerlo en 10s.
parallelism.default	No	El paralelismo del trabajo de Flink. Puede aumentar este valor para mejorar la velocidad de sincronización de datos.
database	Sí	El nombre de la base de datos que se sincronizará con SelectDB.
including-tables	No	Las tablas de PostgreSQL que se sincronizarán. Puede usar una tubería (\|) para separar varios nombres de tablas. Se admiten expresiones regulares. Por ejemplo, `--including-tables table1\|tbl.*` sincroniza table1 y todas las tablas que comienzan con tbl.
excluding-tables	No	Las tablas que se excluirán. Este parámetro se configura de la misma manera que including-tables.
postgres-conf	Sí	Configuración de Postgres CDC Source. Para conocer los detalles de configuración, consulte Postgres CDC Connector, donde `hostname`, `username`, `password`, `database-name`, `table-name` y `schema-name` son obligatorios.
sink-conf	Sí	Configuraciones de Doris Sink. Para obtener más información, consulte Import data by using Flink.
table-conf	No	Configuraciones de la tabla de SelectDB. Corresponden a las propiedades especificadas al crear la tabla de SelectDB.

Migrar con un catalog

La función Catalog de SelectDB permite usar consultas federadas para acceder a PostgreSQL. De este modo, puede migrar datos históricos de PostgreSQL a SelectDB de forma fácil y rápida.

Conéctese a la instancia de SelectDB. Para obtener más información, consulte Connect to an instance.

Nota
Si inicia sesión mediante DMS, el comando SWITCH fallará. Se recomienda conectarse mediante un cliente MySQL.

Cree un catalog JDBC de 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"
);

Parámetros

Parámetro	Obligatorio	Descripción
user	Sí	El nombre de usuario de la cuenta de la base de datos.
password	Sí	La contraseña de la cuenta de la base de datos.
jdbc_url	Sí	La cadena de conexión JDBC.
driver_url	Sí	El nombre del archivo JAR del controlador JDBC.
driver_class	Sí	El nombre de clase del controlador JDBC.
lower_case_table_names	No	Indica si se sincronizan los nombres de bases de datos y tablas del origen de datos JDBC externo en minúsculas. Valor predeterminado: `"false"`
only_specified_database	No	Indica si se sincronizan únicamente las bases de datos especificadas. Valor predeterminado: `"false"`
include_database_list	No	Cuando `only_specified_database=true`, especifique las bases de datos que desea sincronizar, separadas por comas (,). Los nombres de las bases de datos distinguen entre mayúsculas y minúsculas. Valor predeterminado: `""`
exclude_database_list	No	Cuando `only_specified_database=true`, especifique las bases de datos que desea sincronizar. Separe varios nombres de bases de datos con comas (,). Los nombres de las bases de datos distinguen entre mayúsculas y minúsculas. Valor predeterminado: `""`

Para obtener más información, consulte JDBC data sources.

Después de crear una tabla en SelectDB, utilice la sintaxis ETL interna insert into select para sincronizar los datos. Para obtener más información sobre insert into, consulte Insert Into.
```
# Create a table.
CREATE TABLE selectdb_table ...

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