ApsaraDB for SelectDB：遷移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=****

參數說明