PolarDB:Flink CDC for PolarDB-O

Prerequisites

• Determine the Flink-CDC version

  If you use Alibaba Cloud Realtime Compute for Apache Flink, you must determine the community Flink-CDC version that is compatible with your Ververica Runtime (VVR) version. For more information, see CDC and VVR Version Mapping.

  Note

  The Flink-CDC code repository is available at Flink-CDC.

• Determine the Debezium version

  In the pom.xml file of the corresponding Flink-CDC version, find the debezium.version property to determine the Debezium version.

  Note

  The Debezium code repository is available at Debezium.

• Determine the PgJDBC version

  In the pom.xml file of the corresponding Postgres-CDC version, find the org.postgresql dependency 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, the file path is flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/pom.xml.

  • The PgJDBC code repository is available at PgJDBC.

Procedure

The following JAR package for the PolarDB-O Flink CDC connector is built with JDK 8 by following the steps above: flink-sql-connector-postgres-cdc-2.3-SNAPSHOT.jar.

Source connector

This example shows how to use the PolarDB-O Flink CDC connector to synchronize the shipments table from the flink_source database to the shipments_sink table in the flink_sink database on a PolarDB for PostgreSQL (Compatible with Oracle) 2.0 cluster.

1. Prerequisites

   • Prepare PolarDB for PostgreSQL (Compatible with Oracle)

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

     2. Create a privileged account.

     3. View the cluster primary endpoint. If the PolarDB cluster and your Realtime Compute for Apache Flink workspace are in the same virtual private cloud (VPC), you can use the private endpoint. Otherwise, you must apply for and use a public endpoint.

     4. Configure an IP address whitelist for a cluster. Add the IP address of the Flink instance to the PolarDB cluster whitelist.

     5. In the console, create the source database flink_source and the destination database flink_sink. For details, see Create a database.

     6. Execute the following statements to create the shipments table in the flink_source database 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 flink_sink database.

        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)
         );

   • Prepare Realtime Compute for Apache Flink

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

        Note

        We recommend creating the Realtime Compute for Apache Flink workspace in the same Region and VPC as the PolarDB cluster. This allows you to use the PolarDB cluster's private primary endpoint for the connection.

     2. Create a custom connector and upload the PolarDB-O Flink CDC package you built. Select debezium-json for Formats. For details, see Create a custom connector.

2. Create a Flink job

   1. Log on to the Realtime Compute for Apache Flink console and create an SQL draft. For details, see Develop an SQL draft. Use the following Flink SQL code and replace the placeholders for your PolarDB cluster's primary endpoint, port, username, and password.

      Note

      The DATE type in PolarDB for PostgreSQL (Compatible with Oracle) is 64-bit, whereas the DATE type in Flink SQL and most databases is 32-bit. Therefore, you must map DATE columns in the source table to the TIMESTAMP type in both the Flink SQL source and sink tables. 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.

      In the top toolbar of the job editor, click Deploy.

      In the left-side navigation pane, choose Operations Center > Deployments. In the job list, find the target job and click Start in the Actions column.

   3. Test and verify the results.

      • After the job is deployed and in the running state, data from the shipments table is synchronized to the shipments_sink table in the flink_sink database.

        SELECT * FROM public.shipments_sink;

        The following result is returned:

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

      • Execute DML statements on the shipments table in the flink_source database. The changes are synchronized 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;

        The data in the shipments table is synchronized to the shipments_sink table in the flink_sink database.

        SELECT * FROM public.shipments_sink;

        The following result is returned:

         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

This example shows how to use the PolarDB-O Flink CDC pipeline connector to synchronize the shipments1 and shipments2 tables from a PolarDB for PostgreSQL (Compatible with Oracle) 2.0 cluster. For debugging, the sink uses the Print connector. In a production environment, choose a suitable sink connector based on your business requirements.

1. Prerequisites

   • Prepare PolarDB for PostgreSQL (Compatible with Oracle)

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

     2. Create a privileged account.

     3. View the cluster primary endpoint. If the PolarDB cluster and your Realtime Compute for Apache Flink workspace are in the same VPC, you can use the private endpoint. Otherwise, you must apply for and use a public endpoint.

     4. Configure an IP address whitelist for a cluster. Add the IP address of the Flink instance to the PolarDB cluster whitelist.

     5. In the console, create the source database flink_source. For details, see Create a database.

     6. Execute the following statements to create the shipments1 and shipments2 tables in the flink_source database 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();

   • Prepare Realtime Compute for Apache Flink

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

     Note

     We recommend creating the Realtime Compute for Apache Flink workspace in the same Region and VPC as the PolarDB cluster. This allows you to use the PolarDB cluster's private primary endpoint for the connection.

2. Create a Flink job

   1. Log on to the Realtime Compute for Apache Flink console and create a data ingestion draft. For details, see Flink CDC data ingestion. Use the following data ingestion configuration and replace the placeholders for your PolarDB cluster's primary endpoint, port, username, 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. In the More section on the left, add the pipeline connector you built. In the More Configurations panel on the right, confirm that the engine version is vvr-11.5-jdk11-flink-1.20 and add the required dependency files in the Additional Dependency Files area.

   3. Deploy and start the job.

      1. Click Deploy in the upper-right corner.

         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. Go to the Deployments page and click Enable.

   4. Test and verify the results.

      • After a deployment job runs successfully, its status is Running. You can find the CreateTableEvent and DataChangeEvent from the full data phase in the Job Log > Running Task Managers > Stdout log. On the deployment details page, click the Job Log tab, select Running Task Managers, and then click the Stdout tab to view the log output. Confirm that the output contains the following table creation events and data change events:

        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 statements on the shipments1 and shipments2 tables in the flink_source database. The changes are 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 find the DataChangeEvent for the incremental phase in the Job Logs > Running Task Managers > Stdout log:

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