PolarDB:PolarDB for PostgreSQL (Oracle 互換) と互換性のある 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 を使用して PolarDBO Flink CDC コネクタの JAR をパッケージングします：flink-sql-connector-postgres-cdc-2.3-SNAPSHOT.jar。

ソースコネクタ

以下の例では、PolarDB for PostgreSQL (Oracle 互換) 2.0 クラスターの flink_source データベースにある shipments テーブルを、PolarDBO Flink CDC を使用して flink_sink データベースの shipments_sink テーブルに同期する方法を示します。

1. 前提条件

   • PolarDB for PostgreSQL (Oracle 互換) の準備

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

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

     3. クラスターのプライマリエンドポイントを確認します。PolarDB クラスターと Realtime Compute for Apache Flink が同一の仮想プライベートクラウド（VPC）内にある場合は、プライベートエンドポイントを直接使用できます。それ以外の場合は、パブリックエンドポイントを申請してください。

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

     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 のリージョンおよび仮想プライベートクラウド（VPC）は、PolarDB クラスターと一致させることを推奨します。接続アドレスとして、PolarDB クラスターのプライマリエンドポイントのプライベートエンドポイントを直接使用できます。

     2. カスタムコネクタを作成し、パッケージング済みの PolarDBO Flink CDC をアップロードします。フォーマット は debezium-json に設定します。詳細な手順については、「カスタムコネクタの作成」をご参照ください。

        

2. Flink ジョブの作成

   1. Realtime Compute コンソールにログインし、新しい 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. ジョブをデプロイして開始します。

      

      

   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 for PostgreSQL (Oracle 互換) 2.0 クラスターの flink_source データベースにある shipments1 および shipments2 テーブルを、PolarDBO Flink CDC パイプラインコネクタを使用して同期する方法を示します。デバッグ中は、Print コネクタ を使用します。本番環境では、必要に応じて適切なコネクタを選択してください。

1. 前提条件

   • PolarDB for PostgreSQL (Oracle 互換) の準備

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

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

     3. クラスターのプライマリエンドポイントを確認します。PolarDB クラスターと Realtime Compute for Apache Flink が同一の仮想プライベートクラウド（VPC）内にある場合は、プライベートエンドポイントを直接使用できます。それ以外の場合は、パブリックエンドポイントを申請してください。

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

     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 のリージョンおよび仮想プライベートクラウド（VPC）は、PolarDB クラスターと一致させることを推奨します。接続アドレスとして、PolarDB クラスターのプライマリエンドポイントのプライベートエンドポイントを直接使用できます。

2. Flink ジョブの作成

   1. Realtime Compute コンソールにログインし、新しいデータインジェストの下書きを作成します。「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. 左側の詳細設定に、正常にパッケージングされたパイプラインコネクタを追加します。

   3. ジョブをデプロイして開始します。

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

      2. ジョブの O&M ページに移動し、開始をクリックします。

         

   4. テストおよび検証を行います。

      • ジョブが正常にデプロイされ実行状態になると、ジョブログ > 実行中のタスクマネージャー > Stdout ログに完全データフェーズの CreateTableEvent および DataChangeEvent が表示されます。

        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;

      • ジョブログ > 実行中のタスクマネージャー > Stdout ログに増分データフェーズの DataChangeEvent が表示されます。

        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=()}