ApsaraDB for SelectDB:PostgreSQL データソースからのデータ移行

ステップ 1: データソースを追加する

データ同期タスクを開発する場合は、PostgreSQL データソースと SelectDB データソースを DataWorks に追加する必要があります。

1. PostgreSQL データソースを追加します。 詳細については、「PostgreSQL データソース」をご参照ください。

2. SelectDB データソースを追加します。 詳細については、「データソースの追加と管理」をご参照ください。

次の表に、SelectDB データソースのパラメーターを示します。

ステップ 2: データ同期タスクを構成する

詳細については、以下のトピックをご参照ください。

• コードレス UI を使用してバッチ同期タスクを構成する

• コードエディターを使用してバッチ同期タスクを構成する

環境を準備する

Flink 環境を構築します。 この例では、バージョン 1.16 の Flink スタンドアロンクラスタがデプロイされています。

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-sql-connector-mysql-cdc-2.4.2 パッケージと flink-doris-connector-1.16-1.5.2 パッケージを FLINK_HOME/lib ディレクトリにダウンロードします。サンプルコード:

   説明

   データソースから完全データを同期する場合、Flink のバージョンは 1.15 以降である必要があります。異なるバージョンの Flink Doris Connector のダウンロード方法の詳細については、「org/apache/doris」をご参照ください。

   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 スタンドアローン クラスタを起動します。サンプル コード:

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

パラメータの説明