PolarDB:Flink CDC compatible with PolarDB for PostgreSQL (Compatible with Oracle)

Prerequisites

• Determine the Flink-CDC version.

  If you use Alibaba Cloud Realtime Compute for Apache Flink, determine the community Flink-CDC version compatible with the corresponding Ververica Runtime (VVR) version. For details, see CDC and VVR version mapping.

  Note

  For the Flink-CDC code repository, see Flink-CDC.

• Determine the Debezium version.

  In the pom.xml file of the corresponding Flink-CDC version, search for the keyword debezium.version to determine the Debezium version.

  Note

  For the Debezium code repository, see Debezium.

• Determine the PgJDBC version.

  In the pom.xml file of the corresponding Postgres-CDC version, search for the keyword org.postgresql to determine the PgJDBC version.

  Note

  • For versions earlier than release-3.0, the file path is: flink-connector-postgres-cdc/pom.xml.

  • For release-3.0 and later versions, the file path is: flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/pom.xml.

  • For the PgJDBC code repository, see PgJDBC.

Procedure

Following the above steps, package the PolarDBO Flink CDC connector JAR using JDK 8: flink-sql-connector-postgres-cdc-2.3-SNAPSHOT.jar.

Source connector

The following example illustrates how to synchronize the `shipments` table from the `flink_source` database in a PolarDB for PostgreSQL (Compatible with Oracle) 2.0 cluster to the `shipments_sink` table in the `flink_sink` database using PolarDBO Flink CDC.

1. Prerequisites

   • PolarDB for PostgreSQL (Compatible with Oracle) preparation

     1. Purchase a PolarDB for PostgreSQL (Compatible with Oracle) 2.0 cluster on the PolarDB cluster purchase page.

     2. Create a privileged account.

     3. View the cluster's primary endpoint. If the PolarDB cluster and Realtime Compute for Apache Flink are in the same virtual private cloud (VPC), you can directly use the private endpoint. Otherwise, request a public endpoint.

     4. Set the cluster whitelist: Add the Flink instance address to the PolarDB cluster whitelist.

     5. Create source database `flink_source` and target database `flink_sink` in the console. For detailed steps, see Create a database.

     6. Execute the following statement to create the `shipments` table in the source database `flink_source` and insert data.

        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. Execute the following statement to create the `shipments_sink` table in the target database `flink_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 Preparation

     1. Log on to the Realtime Compute console and purchase a Realtime Compute for Apache Flink instance. For more information, see Purchase Realtime Compute for Apache Flink.

        Note

        We recommend that the Realtime Compute for Apache Flink Region and Virtual Private Cloud (VPC) be consistent with those of the PolarDB cluster. You can directly use the private endpoint of the PolarDB cluster's primary endpoint as the connection address.

     2. Create a custom connector and upload the packaged PolarDBO Flink CDC. Set Formats to debezium-json. For detailed steps, see Create a custom connector.

        

2. Create a Flink job

   1. Log on to the Realtime Compute console. Create a new SQL job draft. For guidance, see the job development map. Use the following Flink SQL statements. Replace the PolarDB cluster primary endpoint, port, username, and password with your values.

      Note

      The DATE type in PolarDB for PostgreSQL (Compatible with Oracle) is 64-bit. In contrast, the DATE type in Flink SQL and most databases is 32-bit. Therefore, columns of type DATE in the source table must be defined as TIMESTAMP in both the source and sink tables in Flink SQL. Otherwise, the job fails with a type mismatch error, such as: “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 and start the job.

      

      

   3. Test and validate.

      • After the job deploys successfully and its status changes to running, data from the shipments table is synced to the shipments_sink table in the destination database flink_sink.

        SELECT * FROM public.shipments_sink;

        The result is as follows:

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

      • Run DML statements on the shipments table in the source database flink_source. New inserts and updates sync in real time.

        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;

        Data in the shipments table is now synced to the shipments_sink table in the destination database flink_sink.

        SELECT * FROM public.shipments_sink;

        The result is as follows:

         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)

Pipeline connector

The following example illustrates how to synchronize the `shipments1` and `shipments2` tables from the `flink_source` database in a PolarDB for PostgreSQL (Compatible with Oracle) 2.0 cluster using the PolarDBO Flink CDC Pipeline connector. During debugging, use the Print connector. In a production environment, choose the appropriate connector as needed.

1. Prerequisites

   • PolarDB for PostgreSQL (Compatible with Oracle) preparation

     1. Purchase a PolarDB for PostgreSQL (Compatible with Oracle) 2.0 cluster on the PolarDB cluster purchase page.

     2. Create a privileged account.

     3. View the cluster's primary endpoint. If the PolarDB cluster and Realtime Compute for Apache Flink are in the same virtual private cloud (VPC), you can directly use the private endpoint. Otherwise, request a public endpoint.

     4. Set the cluster whitelist: Add the Flink instance address to the PolarDB cluster whitelist.

     5. Create source database `flink_source` in the console. For detailed steps, see Create a database.

     6. Execute the following statement to create the `shipments1` and `shipments2` tables in the source database `flink_source` and insert data.

        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 preparation

     Log on to the Realtime Compute console and purchase a Realtime Compute for Apache Flink instance. For detailed operations, see Enable Realtime Compute for Apache Flink.

     Note

     We recommend that the region and virtual private cloud (VPC) of Realtime Compute for Apache Flink match those of the PolarDB cluster. The connection address can directly use the private endpoint of the PolarDB cluster's primary endpoint.

2. Create a Flink job.

   1. Log on to the Realtime Compute console and create a new data ingestion draft. See Flink CDC Data Ingestion Job Quick Start. Use the following data ingestion configuration, modifying the PolarDB cluster's primary endpoint, port, account, and password.

      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. Add the successfully packaged Pipeline connector in the More Configurations on the left.

   3. Deploy and start the job.

      1. Click Deploy in the upper right corner.

      2. Go to the job O&M page and click Start.

         

   4. Test and verify.

      • After the job is successfully deployed and running, you can see `CreateTableEvent` and `DataChangeEvent` for the full data phase in the Job logs > Running Task Managers > Stdout logs.

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

      • Execute DML on the `shipments1` and `shipments2` tables in the source database `flink_source`. New additions and modifications will also be synchronized in real time.

        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;

      • You can see `DataChangeEvent` for the incremental data phase in the Job logs > Running Task Managers > Stdout logs:

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