迁移PostgreSQL数据

步骤一：新增数据源

在进行数据同步任务开发时，您需要在DataWorks上分别创建PostgreSQL和SelectDB数据源。

1. 创建PostgreSQL数据源。

2. 创建SelectDB数据源，详情请参见创建并管理数据源。SelectDB数据源的部分配置参数如下所示：

   参数	说明
   数据源名称	数据源的名称。
   MySQL连接地址	JDBC连接串jdbc:mysql://<ip>:<port>/<dbname>。 您可在SelectDB控制台的实例详情 > 网络信息中获取VPC地址（或公网地址）和MySQL协议端口。 示例：jdbc:mysql://selectdb-cn-4xl3jv1****.selectdbfe.rds.aliyuncs.com:9030/test_db
   HTTP连接地址	HTTP协议访问地址<ip>:<port>。 您在SelectDB控制台的实例详情 > 网络信息中获取VPC地址（或公网地址）和HTTP协议端口。 示例：selectdb-cn-4xl3jv1****.selectdbfe.rds.aliyuncs.com:8080
   用户名	SelectDB实例的用户名。
   密码	SelectDB实例对应用户的密码。

步骤二：配置数据同步任务

请参见如下文档，配置数据迁移任务：

• 通过向导模式配置离线同步任务

• 通过脚本模式配置离线同步任务

准备环境

搭建Flink环境，本示例以Flink 1.16单机环境为例。

1. 下载flink-1.16.3-bin-scala_2.12.tgz，进行解压，示例如下。如果此版本已过期，请可以下载其他版本。更多版本，请参见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. 进入FLINK_HOME/lib目录中下载flink-sql-connector-mysql-cdc-2.4.2和flink-doris-connector-1.16-1.5.2，示例如下。

   说明

   整库同步支持Flink 1.15以上的版本，各个版本Flink Doris Connector的下载请参见Flink Doris Connector。

   cd flink-1.16.3
   cd lib/
   wget https://repo1.maven.org/maven2/com/ververica/flink-sql-connector-mysql-cdc/2.4.2/flink-sql-connector-mysql-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. 启动Flink Standalone集群，示例如下。

   bin/start-cluster.sh

4. 创建SelectDB实例。具体操作，请参见创建实例。

5. 通过MySQL协议连接SelectDB实例。具体操作，请参见连接实例。

6. 创建测试数据库和测试表。

   1. 创建测试数据库。

      CREATE DATABASE test_db;

   2. 创建测试表。

      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任务

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

参数说明