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

ステップ1：データソースの追加

データ同期タスクを開発する場合、DataWorks で PostgreSQL と SelectDB のデータソースを作成する必要があります。

1. PostgreSQL データソースを作成してください。

2. SelectDB データソースを作成します。詳細については、「データソース管理」をご参照ください。SelectDB データソースの一部の設定パラメーターは次のとおりです：

   パラメーター	説明
   [データソース名]	データソースの名前。
   [JDBC URL]	JDBC 接続文字列 jdbc:mysql://:/。 SelectDB コンソールのインスタンスの詳細 > ネットワーク情報 ページで、VPC アドレス (または インターネットアドレス) と MySQL プロトコルポート を見つけます。 例： jdbc:mysql://selectdb-cn-4xl3jv1****.selectdbfe.rds.aliyuncs.com:9030/test_db
   [FE HTTP エンドポイント]	HTTP アクセス URL は <ip>:<port> です。 SelectDB コンソールの インスタンスの詳細 > ネットワーク情報 ページで、VPC アドレス (または インターネットアドレス) と HTTP プロトコルポート を確認します。 例： selectdb-cn-4xl3jv1****.selectdbfe.rds.aliyuncs.com:8080
   [ユーザー名]	SelectDB インスタンスのユーザー名。
   [パスワード]	SelectDB インスタンスのユーザーのパスワード

ステップ2：データ同期の設定

次のいずれかの方法でデータ同期タスクを設定してください。

• コードレス UI の使用

• コードエディターの使用

環境の準備

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-postgres-cdc-2.4.2 と flink-doris-connector-1.16-1.5.2 をダウンロードしてください。 次のサンプルコードは参考用です。

   説明

   データベース移行は Flink 1.15 以降でサポートされています。 Flink のバージョンごとに Flink Doris Connector をダウンロードする方法については、「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. 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 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=****

パラメーター