PolarDB:PolarDB-O 向け Flink CDC

前提条件

• Flink-CDC バージョンの決定

  Alibaba Cloud Realtime Compute for Apache Flink を使用する場合、お使いの Ververica Runtime (VVR) バージョンに対応するコミュニティ Flink-CDC バージョンを特定する必要があります。詳細については、「CDC と VVR のバージョンマッピング」をご参照ください。

  説明

  Flink-CDC のコードリポジトリは Flink-CDC にあります。

• Debezium バージョンの決定

  対応する Flink-CDC バージョンの pom.xml ファイルで、debezium.version プロパティを見つけて、Debezium バージョンを決定します。

  説明

  Debezium のコードリポジトリは Debezium にあります。

• PgJDBC バージョンの決定

  対応する Postgres-CDC バージョンの pom.xml ファイルで、org.postgresql 依存関係を見つけて、PgJDBC バージョンを決定します。

  説明

  • release-3.0 未満のバージョンの場合、ファイルパスは flink-connector-postgres-cdc/pom.xml です。

  • release-3.0 以降の場合、ファイルパスは flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/pom.xml です。

  • PgJDBC のコードリポジトリは PgJDBC にあります。

操作手順

上記の手順に従って JDK 8 でビルドされた PolarDB-O Flink CDC コネクタの JAR パッケージは次のとおりです：flink-sql-connector-postgres-cdc-2.3-SNAPSHOT.jar

ソースコネクタ

この例では、PolarDB-O Flink CDC コネクタを使用して、PolarDB for PostgreSQL (Compatible with Oracle) 2.0 クラスター上の flink_source データベースから shipments テーブルを flink_sink データベースの shipments_sink テーブルに同期する方法を示します。

1. 前提条件

   • PolarDB for PostgreSQL (Compatible with Oracle) の準備

     1. PolarDB クラスターの購入ページで、PolarDB for PostgreSQL (Compatible with Oracle) 2.0 クラスターを購入します。

     2. 特権アカウントを作成します。

     3. クラスターのプライマリエンドポイントを表示します。PolarDB クラスターと Realtime Compute for Apache Flink ワークスペースが同じ仮想プライベートクラウド (VPC) 内にある場合は、プライベートエンドポイントを使用できます。それ以外の場合は、パブリックエンドポイントを申請して使用する必要があります。

     4. クラスターの IP アドレスホワイトリストを設定します。Flink インスタンスの IP アドレスを PolarDB クラスターのホワイトリストに追加します。

     5. コンソールで、ソースデータベース flink_source と宛先データベース flink_sink を作成します。詳細については、「データベースの作成」をご参照ください。

     6. 次のステートメントを実行して、flink_source データベースに shipments テーブルを作成し、データを挿入します。

        CREATE TABLE public.shipments (
          shipment_id INT,
          order_id INT,
          origin TEXT,
          destination TEXT,
          is_arrived BOOLEAN,
          order_time DATE,
          PRIMARY KEY (shipment_id) 
        );
        ALTER TABLE public.shipments REPLICA IDENTITY FULL;
        INSERT INTO public.shipments SELECT 1, 1, 'test1', 'test1', false, now();

     7. 次のステートメントを実行して、flink_sink データベースに shipments_sink テーブルを作成します。

        CREATE TABLE public.shipments_sink (
           shipment_id INT,
           order_id INT,
           origin TEXT,
           destination TEXT,
           is_arrived BOOLEAN,
           order_time TIMESTAMP,
           PRIMARY KEY (shipment_id)
         );

   • Realtime Compute for Apache Flink の準備

     1. Realtime Compute コンソールにログオンし、Realtime Compute for Apache Flink インスタンスを購入します。詳細については、「Realtime Compute for Apache Flink のアクティベート」をご参照ください。

        説明

        Realtime Compute for Apache Flink ワークスペースは、PolarDB クラスターと同じ [リージョン] および [VPC] に作成することを推奨します。これにより、接続に PolarDB クラスターのプライベートプライマリエンドポイントを使用できます。

     2. カスタムコネクタを作成し、ビルドした PolarDB-O Flink CDC パッケージをアップロードします。[Formats] には debezium-json を選択します。詳細については、「カスタムコネクタの作成」をご参照ください。

2. Flink ジョブの作成

   1. Realtime Compute for Apache Flink コンソールにログオンし、SQL ドラフトを作成します。詳細については、「SQL ドラフトの開発」をご参照ください。次の Flink SQL コードを使用し、PolarDB クラスターのプライマリエンドポイント、ポート、ユーザー名、パスワードのプレースホルダーを置き換えてください。

      説明

      PolarDB for PostgreSQL (Oracle 互換) の DATE 型は 64 ビットですが、Flink SQL およびほとんどのデータベースの DATE 型は 32 ビットです。したがって、ソーステーブルの DATE 列を、Flink SQL のソーステーブルとシンクテーブルの両方で TIMESTAMP 型にマッピングする必要があります。そうしないと、ジョブは "java.time.DateTimeException: Invalid value for EpochDay (valid values -365243219162 - 365241780471): 1720891573000" のような型の不一致エラーで失敗します。

      CREATE TEMPORARY TABLE shipments (
         shipment_id INT,
         order_id INT,
         origin STRING,
         destination STRING,
         is_arrived BOOLEAN,
         order_time TIMESTAMP,
         PRIMARY KEY (shipment_id) NOT ENFORCED
       ) WITH (
         'connector' = 'polardbo-cdc',
         'hostname' = '<yourHostname>',
         'port' = '<yourPort>',
         'username' = '<yourUserName>',
         'password' = '<yourPassWord>',
         'database-name' = 'flink_source',
         'schema-name' = 'public',
         'table-name' = 'shipments',
         'decoding.plugin.name' = 'pgoutput',
         'slot.name' = 'flink'
       );
      CREATE TEMPORARY TABLE shipments_sink (
         shipment_id INT,
         order_id INT,
         origin STRING,
         destination STRING,
         is_arrived BOOLEAN,
         order_time TIMESTAMP,
         PRIMARY KEY (shipment_id) NOT ENFORCED
       ) WITH (
        'connector' = 'jdbc',
        'url' = 'jdbc:postgresql://<yourHostname>:<yourPort>/flink_sink',
        'table-name' = 'shipments_sink',
        'username' = '<yourUserName>',
        'password' = '<yourPassWord>'
      );
      INSERT INTO shipments_sink SELECT * FROM shipments;

   2. ジョブのデプロイと開始。

      ジョブエディターの上部ツールバーで、[Deploy] をクリックします。

      左側メニューで、[Operations Center] > [Deployments] を選択します。ジョブリストでターゲットジョブを見つけ、[操作] 列の [Start] をクリックします。

   3. 結果のテストと検証。

      • ジョブがデプロイされ、実行中の状態になると、shipments テーブルのデータが flink_sink データベースの shipments_sink テーブルに同期されます。

        SELECT * FROM public.shipments_sink;

        次の結果が返されます：

         shipment_id | order_id | origin | destination | is_arrived |     order_time      
        -------------+----------+--------+-------------+------------+---------------------
                   1 |        1 | test1  | test1       | f          | 2024-09-18 05:45:08
        (1 row)

      • flink_source データベースの shipments テーブルで DML ステートメントを実行します。変更はリアルタイムで同期されます。

        INSERT INTO public.shipments SELECT 2, 2, 'test2', 'test2', false, now();
        UPDATE public.shipments SET is_arrived = true WHERE shipment_id = 1;
        DELETE FROM public.shipments WHERE shipment_id = 2;
        INSERT INTO public.shipments SELECT 3, 3, 'test3', 'test3', false, now();
        UPDATE public.shipments SET is_arrived = true WHERE shipment_id = 3;

        shipments テーブルのデータが flink_sink データベースの shipments_sink テーブルに同期されます。

        SELECT * FROM public.shipments_sink;

        次の結果が返されます：

         shipment_id | order_id | origin | destination | is_arrived |     order_time      
        -------------+----------+--------+-------------+------------+---------------------
                   1 |        1 | test1  | test1       | t          | 2024-09-18 05:45:08
                   3 |        3 | test3  | test3       | t          | 2024-09-18 07:33:23
        (2 rows)

パイプラインコネクタ

この例では、PolarDB-O Flink CDC パイプラインコネクタを使用して、PolarDB for PostgreSQL (Compatible with Oracle) 2.0 クラスターから shipments1 および shipments2 テーブルを同期する方法を示します。デバッグのため、シンクには Print コネクタ を使用します。本番環境では、ビジネス要件に基づいて適切なシンクコネクタを選択してください。

1. 前提条件

   • PolarDB for PostgreSQL (Compatible with Oracle) の準備

     1. PolarDB クラスターの購入ページで、PolarDB for PostgreSQL (Compatible with Oracle) 2.0 クラスターを購入します。

     2. 特権アカウントを作成します。

     3. クラスターのプライマリエンドポイントを表示します。PolarDB クラスターと Realtime Compute for Apache Flink ワークスペースが同じ VPC 内にある場合は、プライベートエンドポイントを使用できます。それ以外の場合は、パブリックエンドポイントを申請して使用する必要があります。

     4. クラスターの IP アドレスホワイトリストを設定します。Flink インスタンスの IP アドレスを PolarDB クラスターのホワイトリストに追加します。

     5. コンソールで、ソースデータベース flink_source を作成します。詳細については、「データベースの作成」をご参照ください。

     6. 次のステートメントを実行して、flink_source データベースに shipments1 および shipments2 テーブルを作成し、データを挿入します。

        CREATE TABLE public.shipments1 (
          shipment_id INT,
          order_id INT,
          origin TEXT,
          destination TEXT,
          is_arrived BOOLEAN,
          order_time DATE,
          PRIMARY KEY (shipment_id) 
        );
        ALTER TABLE public.shipments1 REPLICA IDENTITY FULL;
        INSERT INTO public.shipments1 SELECT 1, 1, 'test1', 'test1', false, now();
        CREATE TABLE public.shipments2 (
          shipment_id INT,
          order_id INT,
          origin TEXT,
          destination TEXT,
          is_arrived BOOLEAN,
          order_time DATE,
          PRIMARY KEY (shipment_id) 
        );
        ALTER TABLE public.shipments2 REPLICA IDENTITY FULL;
        INSERT INTO public.shipments2 SELECT 1, 1, 'test1', 'test1', false, now();

   • Realtime Compute for Apache Flink の準備

     Realtime Compute コンソールにログオンし、Realtime Compute for Apache Flink インスタンスを購入します。詳細については、「Realtime Compute for Apache Flink のアクティベート」をご参照ください。

     説明

     Realtime Compute for Apache Flink ワークスペースは、PolarDB クラスターと同じ [リージョン] および [VPC] に作成することを推奨します。これにより、接続に PolarDB クラスターのプライベートプライマリエンドポイントを使用できます。

2. Flink ジョブの作成

   1. Realtime Compute for Apache Flink コンソールにログオンし、データ取り込みドラフトを作成します。詳細については、「Flink CDC データ取り込み」をご参照ください。次のデータ取り込み設定を使用し、PolarDB クラスターのプライマリエンドポイント、ポート、ユーザー名、パスワードのプレースホルダーを置き換えてください。

      source:
         type: polardbo
         name: PolarDB Oracle Source
         hostname: '<yourHostname>'
         port: '<yourPort>'
         username: '<yourUserName>'
         password: '<yourPassWord>'
         tables: flink_source.public.shipments[12]
         decoding.plugin.name:  pgoutput
         slot.name: pgtest
      sink:
        type: values
        name: values Sink
        print.enabled: true

   2. 左側の More セクションに、作成したパイプラインコネクタを追加します。右側の [詳細設定] パネルで、エンジンバージョンが vvr-11.5-jdk11-flink-1.20 であることを確認し、[追加の依存ファイル] 領域に必要な依存ファイルを追加します。

   3. ジョブのデプロイと開始。

      1. 右上隅のデプロイメントをクリックします。

         source:
           type: polardbo
           name: PolarDB Oracle Source
           hostname: xxx
           port: xxx
           username: xxx
           password: xxx
           tables: flink_source.public.shipments[12]
           decoding.plugin.name: pgoutput
           slot.name: pgtest
         sink:
           type: values
           name: values Sink
           print.enabled: true

      2. 「デプロイメント」ページに移動し、Enable をクリックします。

   4. 結果のテストと検証。

      • デプロイメントジョブが正常に実行されると、ステータスは「実行中」になります。全データフェーズの CreateTableEvent と DataChangeEvent は、[ジョブログ] > [実行中のタスクマネージャー] > [Stdout] ログで確認できます。デプロイメント詳細ページで、[ジョブログ] タブをクリックし、[実行中のタスクマネージャー] を選択し、次に [Stdout] タブをクリックしてログ出力を表示します。出力に次のテーブル作成イベントとデータ変更イベントが含まれていることを確認します。

        CreateTableEvent{tableId=public.shipments2, schema=columns={`shipment_id` INT NOT NULL,`order_id` INT,`origin` STRING,`destination` STRING,`is_arrived` BOOLEAN,`order_time` TIMESTAMP(6)}, primaryKeys=shipment_id, options=()}
        CreateTableEvent{tableId=public.shipments1, schema=columns={`shipment_id` INT NOT NULL,`order_id` INT,`origin` STRING,`destination` STRING,`is_arrived` BOOLEAN,`order_time` TIMESTAMP(6)}, primaryKeys=shipment_id, options=()}
        DataChangeEvent{tableId=public.shipments2, before=[], after=[1, 1, test1, test1, false, 2026-01-07T16:30:44], op=INSERT, meta=()}
        DataChangeEvent{tableId=public.shipments1, before=[], after=[1, 1, test1, test1, false, 2026-01-07T16:30:44], op=INSERT, meta=()}

      • flink_source データベースの shipments1 および shipments2 テーブルで DML ステートメントを実行します。変更はリアルタイムで同期されます。

        INSERT INTO public.shipments1 SELECT 2, 2, 'test2', 'test2', false, now();
        UPDATE public.shipments1 SET is_arrived = true WHERE shipment_id = 1;
        DELETE FROM public.shipments1 WHERE shipment_id = 2;
        INSERT INTO public.shipments1 SELECT 3, 3, 'test3', 'test3', false, now();
        UPDATE public.shipments1 SET is_arrived = true WHERE shipment_id = 3;
        INSERT INTO public.shipments2 SELECT 2, 2, 'test2', 'test2', false, now();
        UPDATE public.shipments2 SET is_arrived = true WHERE shipment_id = 1;
        DELETE FROM public.shipments2 WHERE shipment_id = 2;
        INSERT INTO public.shipments2 SELECT 3, 3, 'test3', 'test3', false, now();
        UPDATE public.shipments2 SET is_arrived = true WHERE shipment_id = 3;

      • 増分フェーズの DataChangeEvent は、[ジョブログ] > [実行中のタスクマネージャー] > [Stdout] ログで確認できます：

        DataChangeEvent{tableId=public.shipments1, before=[], after=[2, 2, test2, test2, false, 2026-01-07T16:44:50], op=INSERT, meta=()}
        DataChangeEvent{tableId=public.shipments1, before=[1, 1, test1, test1, false, 2026-01-07T16:30:44], after=[1, 1, test1, test1, true, 2026-01-07T16:30:44], op=UPDATE, meta=()}
        DataChangeEvent{tableId=public.shipments1, before=[2, 2, test2, test2, false, 2026-01-07T16:44:50], after=[], op=DELETE, meta=()}
        DataChangeEvent{tableId=public.shipments1, before=[], after=[3, 3, test3, test3, false, 2026-01-07T16:44:50], op=INSERT, meta=()}
        DataChangeEvent{tableId=public.shipments1, before=[3, 3, test3, test3, false, 2026-01-07T16:44:50], after=[3, 3, test3, test3, true, 2026-01-07T16:44:50], op=UPDATE, meta=()}
        DataChangeEvent{tableId=public.shipments2, before=[], after=[2, 2, test2, test2, false, 2026-01-07T16:44:50], op=INSERT, meta=()}
        DataChangeEvent{tableId=public.shipments2, before=[1, 1, test1, test1, false, 2026-01-07T16:30:44], after=[1, 1, test1, test1, true, 2026-01-07T16:30:44], op=UPDATE, meta=()}
        DataChangeEvent{tableId=public.shipments2, before=[2, 2, test2, test2, false, 2026-01-07T16:44:50], after=[], op=DELETE, meta=()}
        DataChangeEvent{tableId=public.shipments2, before=[], after=[3, 3, test3, test3, false, 2026-01-07T16:44:50], op=INSERT, meta=()}
        DataChangeEvent{tableId=public.shipments2, before=[3, 3, test3, test3, false, 2026-01-07T16:44:50], after=[3, 3, test3, test3, true, 2026-01-07T16:44:50], op=UPDATE, meta=()}