云原生数据库 PolarDB：兼容PolarDB PostgreSQL版（兼容Oracle）的Flink CDC

操作前提

• 确定Flink-CDC版本

  如果您使用的是阿里云实时计算 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。

操作步骤

按照以上流程基于JDK8打包出PolarDBO Flink CDC连接器的JAR包：flink-sql-connector-postgres-cdc-2.3-SNAPSHOT.jar。

Source连接器

以下示例用于说明，如何通过PolarDBO Flink CDC，将PolarDB PostgreSQL版（兼容Oracle） 2.0集群中flink_source库的shipments表，同步到flink_sink库的shipments_sink表中。

1. 前提准备

   • PolarDB PostgreSQL版（兼容Oracle）准备

     1. 在PolarDB集群购买页面，购买PolarDB PostgreSQL版（兼容Oracle） 2.0集群。

     2. 创建高权限账号。

     3. 查看集群主地址，如果PolarDB集群和实时计算 Flink 版在同一专有网络内，可直接使用私网地址，否则需要申请公网地址。

     4. 设置集群白名单：将Flink实例地址添加到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)
         );

   • 实时计算 Flink 版准备

     1. 登录实时计算控制台，购买实时计算 Flink 版实例，详细操作请参考开通实时计算Flink版。

        说明

        建议实时计算 Flink 版的地域和专有网络和PolarDB集群保持一致，连接地址可以直接使用PolarDB集群主地址的私网地址。

     2. 创建自定义连接器，上传打包好的PolarDBO Flink CDC，Formats选择debezium-json，详细步骤请参考创建自定义连接器。

        

2. 创建Flink作业

   1. 登录实时计算控制台，新建一个SQL作业草稿，请参考作业开发地图。使用以下Flink SQL语句，修改PolarDB集群主地址，端口，账号和密码。

      说明

      PolarDB PostgreSQL版（兼容Oracle）的DATE类型是64位，而Flink SQL以及大部分数据库的DATE类型为32位。因此，源表中DATE类型的列，在Flink SQL的source和sink表中都必须要指定为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)

Pipeline连接器

以下示例用于说明，如何通过PolarDBO Flink CDC Pipeline连接器，同步PolarDB PostgreSQL版（兼容Oracle） 2.0集群中flink_source库的shipments1表和shipments2表。在调试过程中，Sink使用Print连接器，在生产环境中，请根据业务需求选择合适的连接器。

1. 前提准备

   • PolarDB PostgreSQL版（兼容Oracle）准备

     1. 在PolarDB集群购买页面，购买PolarDB PostgreSQL版（兼容Oracle） 2.0集群。

     2. 创建高权限账号。

     3. 查看集群主地址，如果PolarDB集群和实时计算 Flink 版在同一专有网络内，可直接使用私网地址，否则需要申请公网地址。

     4. 设置集群白名单：将Flink实例地址添加到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();

   • 实时计算 Flink 版准备

     登录实时计算控制台，购买实时计算 Flink 版实例，详细操作请参考开通实时计算Flink版。

     说明

     建议实时计算 Flink 版的地域和专有网络和PolarDB集群保持一致，连接地址可以直接使用PolarDB集群主地址的私网地址。

2. 创建Flink作业

   1. 登录实时计算控制台，新建一个数据摄入草稿，请参考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. 在左侧更多配置中添加上述打包成功的Pipeline连接器。

   3. 部署并启动作业。

      1. 单击右上角的部署。

      2. 进入作业运维页面，单击启动。

         

   4. 测试与验证。

      • 部署作业运行成功后，即状态为运行中。您可以从作业日志 > 运行Task Managers > 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;

      • 您可以从作业日志 > 运行Task Managers > 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=()}