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

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

DataWorks に PostgreSQL データソースおよび SelectDB データソースを追加します。

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

2. SelectDB データソースを追加します。詳細については、「データソースの追加と管理」をご参照ください。SelectDB データソースを追加する際には、以下のパラメーターを使用します。

   パラメーター	説明	例
   データソース名	このデータソース接続の名前。	selectdb_prod
   ホストアドレス／IP アドレス	jdbc:mysql://<ip>:<port>/<dbname> 形式の JDBC URL。SelectDB コンソールの「ネットワーク情報」セクション（「インスタンスの詳細」ページ内）から、VPC エンドポイントまたはパブリックエンドポイントと MySQL ポートを取得します。	jdbc:mysql://selectdb-cn-4xl3jv1****.selectdbfe.rds.aliyuncs.com:9030/test_db
   HTTP 接続アドレス	<ip>:<port> 形式の HTTP エンドポイント。SelectDB コンソールの「ネットワーク情報」セクション（「インスタンスの詳細」ページ内）から、VPC エンドポイントまたはパブリックエンドポイントと HTTP ポートを取得します。	selectdb-cn-4xl3jv1****.selectdbfe.rds.aliyuncs.com:8080
   ユーザー名	SelectDB インスタンスの所有者アカウントのユーザー名。	admin
   パスワード	SelectDB インスタンスの所有者アカウントのパスワード。	—

ステップ 2：データ同期タスクの構成

ビジュアルエディタまたはコードエディタのいずれかを使用して、バッチ同期タスクを作成します。

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

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

環境のセットアップ

以下の手順では、Flink バージョン 1.16 のスタンドアロンクラスターをデプロイします。

1. 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

   バージョン 1.16.3 が利用不可の場合、Apache Flink のダウンロードページから後続のバージョンをダウンロードしてください。

2. CDC コネクタおよび SelectDB コネクタを lib/ ディレクトリにダウンロードします。

   完全データ同期には Flink 1.15 以降が必要です。その他のコネクタバージョンについては、「Maven 上の org/apache/doris」をご参照ください。

   cd flink-1.16.3/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. SelectDB にテスト用データベースおよびテーブルを作成します。

   CREATE DATABASE test_db;
   
   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 ジョブの送信

以下のコマンドは、PostgreSQL ソースから指定されたフィルターに一致するすべてのテーブルを同期し、SelectDB データベース db1 に書き込む 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=****

パラメーターのリファレンス：