Migrate data from PostgreSQL

Step 1: Add data sources

When you develop a data synchronization task, you need to create PostgreSQL and SelectDB data sources in DataWorks.

1. Create a PostgreSQL data source.

2. Create a SelectDB data source. For more information, see Data Source Management. Some configuration parameters for the SelectDB data source are as follows:

   Parameter	Description
   Data source name	The name of the data source.
   JDBC URL	JDBC connection string jdbc:mysql://<ip>:<port>/<dbname>. Find the VPC Endpoint (or Public Endpoint) and MySQL Port on the Instance Details > Network Information page in the SelectDB console. Example: jdbc:mysql://selectdb-cn-4xl3jv1****.selectdbfe.rds.aliyuncs.com:9030/test_db
   FE HTTP endpoint	The HTTP access URL is <ip>:<port>. Find the VPC Endpoint (or Public Endpoint) and HTTP Port on the Instance Details > Network Information page in the SelectDB console. Example: selectdb-cn-4xl3jv1****.selectdbfe.rds.aliyuncs.com:8080
   Username	The username of the SelectDB instance.
   Password	The password for the user of the SelectDB instance.

Step 2: Configure data synchronization

Configure a data synchronization task by using one of the following methods:

• Use the Codeless UI

• Use the Code Editor

Prepare the environment

Set up a Flink environment. This example uses a Flink 1.16 standalone environment.

1. Download and extract the flink-1.16.3-bin-scala_2.12.tgz package. To find other available versions, see 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

2. Go to the FLINK_HOME/lib directory and download flink-sql-connector-postgres-cdc-2.4.2 and flink-doris-connector-1.16-1.5.2. The following sample code is for your reference.

   Note

   Database migration is supported in Flink 1.15 and later. For information about how to download the Flink Doris Connector for different Flink versions, see 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

3. Start a Flink standalone cluster. Example:

   bin/start-cluster.sh

4. Create a SelectDB instance. For more information, see Create an instance.

5. Connect to a SelectDB instance using the MySQL protocol. For more information, see Connect to an instance.

6. Create a test database and a test table.

   1. Create a test database.

      CREATE DATABASE test_db;

   2. Create a test table.

      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;

Submit a Flink CDC job

Use the following syntax to submit the Flink CDC job:

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

Parameters