Migrar dados do PostgreSQL para o ApsaraDB for SelectDB - ApsaraDB for SelectDB: Central de Documentação do Alibaba Cloud

Use DTS, DataWorks, Flink CDC e Catalog para migrar dados de fontes PostgreSQL, como PostgreSQL autogerenciado, ApsaraDB RDS for PostgreSQL e PolarDB for PostgreSQL, para o ApsaraDB for SelectDB. Selecione o método de migração adequado com base no volume de dados e no cenário de negócios.

Soluções de migração

DTS, DataWorks, Flink CDC e catalog permitem migrar dados do PostgreSQL para o SelectDB, mas os dados suportados variam conforme o método. Escolha a abordagem apropriada de acordo com o cenário de negócios.

Solução	Migração de dados históricos	Sincronização incremental de dados	Migração de schema	Migração de banco de dados	Sincronização incremental de DDL	Verificação de dados
DTS	✔️	✔️	✔️	✔️	✔️	✔️
DataWorks	✔️	✔️	✔️	✔️	✔️	❌
Flink CDC	✔️	✔️	✔️	✔️	✔️	❌
catalog	✔️	❌	❌	❌	❌	❌

Pré-requisitos

Garanta a conectividade de rede entre a instância PostgreSQL e a instância do SelectDB.
- A instância PostgreSQL e a instância do SelectDB devem estar na mesma VPC. Caso contrário, resolva primeiro o problema de conectividade de rede. Para mais informações, consulte How to resolve network connectivity issues between a SelectDB instance and a data source?
- Adicione o endereço IP da instância PostgreSQL à lista de permissões de endereços IP do SelectDB. Para mais informações, consulte Set an IP address whitelist.
- Se a instância PostgreSQL tiver uma lista de permissões de endereços IP, adicione o intervalo de endereços IP do segmento de rede da instância do SelectDB à lista de permissões de endereços IP da instância PostgreSQL.
  - Para obter o intervalo de IPs da VPC onde reside a instância do SelectDB, consulte How do I find the IP CIDR block of the VPC where my ApsaraDB for SelectDB instance is located?
  - Para obter o endereço IP público da instância do SelectDB, execute o comando ping no endpoint público do SelectDB.

Procedimento

Migrar com DTS

O DTS suporta a migração de dados históricos e a sincronização de dados incrementais do PostgreSQL para o SelectDB. Também oferece recursos como migração de schema, sincronização de DDL e verificação de dados.

Para sincronizar dados, consulte os seguintes tópicos:
- 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 dados, consulte os seguintes tópicos:
- 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 com DataWorks

Etapa 1: Adicionar fontes de dados

Ao desenvolver uma tarefa de sincronização de dados, crie fontes de dados PostgreSQL e SelectDB no DataWorks.

Crie um PostgreSQL data source.

Crie uma fonte de dados do SelectDB. Para mais informações, consulte Data Source Management. Alguns parâmetros de configuração da fonte de dados do SelectDB são descritos abaixo:

Parâmetro	Descrição
Data source name	Nome da fonte de dados.
JDBC URL	String de conexão JDBC `jdbc:mysql://<ip>:<port>/<dbname>`. Encontre o VPC Endpoint (ou Public Endpoint) e a MySQL Port na página Instance Details > Network Information no console do SelectDB. Exemplo: `jdbc:mysql://selectdb-cn-4xl3jv1****.selectdbfe.rds.aliyuncs.com:9030/test_db`
FE HTTP endpoint	A URL de acesso HTTP é `<ip>:<port>`. Encontre o VPC Endpoint (ou Public Endpoint) e a HTTP Port na página Instance Details > Network Information no console do SelectDB. Exemplo: `selectdb-cn-4xl3jv1****.selectdbfe.rds.aliyuncs.com:8080`
Username	Nome de usuário da instância do SelectDB.
Password	Senha do usuário da instância do SelectDB.

Etapa 2: Configurar a sincronização de dados

Configure uma tarefa de sincronização de dados usando um dos métodos a seguir:

Migrar com Flink CDC

O Flink utiliza o Flink CDC para migrar dados do PostgreSQL para o SelectDB. O Flink CDC suporta tanto a migração de dados históricos quanto a sincronização de dados incrementais, oferecendo capacidades abrangentes como migração de schema e tabelas, além de sincronização de DDL.

Preparar o ambiente

Configure um ambiente Flink. Este exemplo utiliza um ambiente standalone do Flink 1.16.

Baixe e extraia o pacote flink-1.16.3-bin-scala_2.12.tgz. Para encontrar outras versões disponíveis, consulte 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

Acesse o diretório FLINK_HOME/lib e baixe o flink-sql-connector-postgres-cdc-2.4.2 e o flink-doris-connector-1.16-1.5.2. O código de exemplo a seguir serve como referência.

Nota
A migração de banco de dados é suportada no Flink 1.15 e versões posteriores. Para obter informações sobre como baixar o Flink Doris Connector para diferentes versões do 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 um cluster standalone do Flink. Exemplo:
```
bin/start-cluster.sh
```
Crie uma instância do SelectDB. Para mais informações, consulte Create an instance.
Conecte-se a uma instância do SelectDB usando o protocolo MySQL. Para mais informações, consulte Connect to an instance.

Crie um banco de dados de teste e uma tabela de teste.

Crie um banco de dados de teste.
```
CREATE DATABASE test_db;
```

Crie uma tabela de teste.

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 um job do Flink CDC

Use a sintaxe a seguir para enviar o job do 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	Obrigatório	Descrição
execution.checkpointing.interval	Sim	Intervalo de checkpoint do Flink. Afeta a frequência de sincronização de dados. Recomenda-se o valor de 10s.
parallelism.default	Não	Paralelismo do job do Flink. Aumente o paralelismo para melhorar a velocidade de sincronização de dados.
database	Sim	Nome do banco de dados para sincronização com o SelectDB.
including-tables	Não	Tabelas do PostgreSQL a serem sincronizadas. Use uma barra vertical (\|) para separar vários nomes de tabelas. Expressões regulares são suportadas. Por exemplo, `--including-tables table1\|tbl.*` sincroniza a table1 e todas as tabelas que começam com tbl.
excluding-tables	Não	Tabelas a serem excluídas. Configure este parâmetro da mesma forma que including-tables.
postgres-conf	Sim	Configuração do Postgres CDC Source. Para detalhes de configuração, consulte Postgres CDC Connector, onde `hostname`, `username`, `password`, `database-name`, `table-name` e `schema-name` são obrigatórios.
sink-conf	Sim	Configurações para o Doris Sink. Para mais informações, consulte Import data by using Flink.
table-conf	Não	Configurações para a tabela do SelectDB. Estas são as propriedades especificadas ao criar a tabela do SelectDB.

Migrar com um catalog

O recurso Catalog no SelectDB permite usar consultas federadas para acessar o PostgreSQL, facilitando e acelerando a migração de dados históricos do PostgreSQL para o SelectDB.

Conecte-se à instância do SelectDB. Para mais informações, consulte Connect to an instance.

Nota
Ao fazer login usando o DMS, o comando SWITCH falha. Recomendamos conectar-se usando um cliente MySQL.

Crie um catalog JDBC do 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	Obrigatório	Descrição
user	Sim	Nome de usuário da conta do banco de dados.
password	Sim	Senha da conta do banco de dados.
jdbc_url	Sim	String de conexão JDBC.
driver_url	Sim	Nome do arquivo JAR do driver JDBC.
driver_class	Sim	Nome da classe do driver JDBC.
lower_case_table_names	Não	Define se os nomes de bancos de dados e tabelas da fonte de dados JDBC externa devem ser sincronizados em letras minúsculas. Valor padrão: `"false"`
only_specified_database	Não	Define se apenas os bancos de dados especificados devem ser sincronizados. Valor padrão: `"false"`
include_database_list	Não	Quando `only_specified_database=true`, especifique os múltiplos bancos de dados a serem sincronizados, separados por vírgulas (,). Os nomes dos bancos de dados diferenciam maiúsculas de minúsculas. Valor padrão: `""`
exclude_database_list	Não	Quando `only_specified_database=true`, especifique os bancos de dados a excluir da sincronização. Use vírgulas (,) para separar vários nomes de bancos de dados. Os nomes diferenciam maiúsculas de minúsculas. Valor padrão: `""`

Para mais informações, consulte JDBC data sources.

Após criar uma tabela no SelectDB, utilize a sintaxe ETL in-database insert into select para sincronizar os dados. Para mais informações 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;
```