兼容PolarDB PostgreSQL版（兼容Oracle）的Debezium connector

由于PolarDB PostgreSQL版（兼容Oracle）与社区PostgreSQL仅在少量数据类型和内置对象处理存在差异，本文为您介绍如何基于社区Debezium PostgreSQL connector，通过少量代码适配打包出支持PolarDB PostgreSQL版（兼容Oracle）的Debezium connector。

打包Debezium PolarDBO connector

使用说明

Debezium PolarDBO connector是通过PolarDB PostgreSQL版（兼容Oracle）数据库的逻辑复制读取增量变更，使用时需要满足以下条件：

• wal_level参数的值需设置为logical，即在预写式日志WAL（Write-ahead logging）中增加支持逻辑复制所需的信息。

  说明

  您可以通过控制台设置wal_level参数，详细操作请参考设置集群参数。修改该参数后集群将会重启，请在修改参数前做好业务安排，谨慎操作。

• 执行ALTER TABLE schema.table REPLICA IDENTITY FULL;命令设置订阅表的REPLICA IDENTITY为FULL（发出的插入和更新操作事件包含表中所有列的旧值），以保障该表数据同步的一致性。

  说明

  • REPLICA IDENTITY是PostgreSQL特有的表级设置，决定了逻辑解码插件在发生（INSERT）和更新（UPDATE）事件时，是否包含涉及的表列的旧值。REPLICA IDENTITY取值含义详情，请参见REPLICA IDENTITY。

  • 设置订阅表的REPLICA IDENTITY为FULL时可能需要锁表，进而影响业务，请在修改参数前做好业务安排。您可以通过以下命令查看当前配置是否为FULL：

    SELECT relreplident = 'f' FROM pg_class WHERE relname = 'tablename';

• 需要确保max_wal_senders和max_replication_slots的参数值均大于当前数据库复制槽已使用数和Kafka作业所需要的slot数量。

• 确保使用的是高权限账号或者同时拥有LOGIN和REPLICATION权限的普通账号，并且具有订阅表的SELECT权限用于全量数据查询。

• 只能连接PolarDB集群的主地址，集群地址不支持逻辑复制。

• connector.class参数指定为io.debezium.connector.postgresql.PolarDBOConnector。

• 建议将plugin.name参数设置为pgoutput，否则非UTF-8编码的数据库可能会发生增量解析乱码，详细介绍请参考社区文档。

示例

以下示例用于说明，如何通过Debezium PolarDBO connector，将PolarDB PostgreSQL版Oracle语法兼容 2.0集群中dbz_db库的t1和t2表，同步到Kafka消息队列中。

前提准备

1. Kafka准备

   1. 部署Kafka实例，并确保在Kafka Connect的主机上能够成功访问。您也可以直接使用云消息队列 Kafka 版，详情请参考快速入门。

   2. 在Kafka实例中创建一个名为pg_dbz_event的Topic，用于接收消息。

      说明

      测试场景为便于查看可以创建单分区Topic，对于生产环境请创建多分区Topic。

2. 在本地以distributed模式启动Kafka Connect，端口为8083。

   • 将上文打包的Debezium PolarDBO connectorJAR包拷贝到Kafka Connect的plugin.path目录中。

     # ${plugin.path} 请替换为具体的路径
     mkdir ${plugin.path}/debezium-connector-polardbo
     cp debezium-connector-postgres-polardbo-v1.0-2.6.2.Final.jar ${plugin.path}/debezium-connector-polardbo

3. PolarDB PostgreSQL版（兼容Oracle）准备

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

   2. 按照使用说明，完成PolarDB集群配置，满足Debezium PolarDBO connector使用前提。

   3. 创建高权限账户，详细操作请参考创建账号。

   4. 获取集群主地址，详细操作请参考查看连接地址。如果PolarDB集群和Kafka Connect在同一可用区，可直接使用私网地址，否则需要申请公网地址。将Kafka Connect实例地址添加到PolarDB集群白名单中，请参考设置集群白名单。

   5. 在控制台创建数据库dbz_db，详细步骤请参考创建数据库。

   6. 执行如下语句，在数据库dbz_db中创建表t1、t2，并写入数据。

      CREATE TABLE public.t1 (a int PRIMARY KEY, b text, c TIMESTAMP);
      ALTER TABLE public.t1 REPLICA IDENTITY FULL;
      INSERT INTO public.t1(a, b, c) VALUES(1, 'a', now());
      CREATE TABLE public.t2 (a int PRIMARY KEY, b text, c DATE);
      ALTER TABLE public.t2 REPLICA IDENTITY FULL;
      INSERT INTO public.t2(a, b, c) VALUES(1, 'a', now());

测试

1. 创建配置文件config/postgresql-connector.json，配置说明请参考社区文档。

   {
     "name": "dbz-polardb",
     "config": {
       "connector.class": "io.debezium.connector.postgresql.PolarDBOConnector",
       "database.hostname": "<yourHostname>", 
       "database.port": "<yourPort>", 
       "database.user": "<yourUserName>", 
       "database.password": "<yourPassWord>", 
       "database.dbname" : "dbz_db",
       "plugin.name": "pgoutput",
       "slot.name": "dbz_polardb",
       "table.include.list": "public.t1,public.t2",
       "topic.prefix": "polardb"
       "transforms": "Combine",
       "transforms.Combine.type": "io.debezium.transforms.ByLogicalTableRouter",
       "transforms.Combine.topic.regex": "(.*)",
       "transforms.Combine.topic.replacement": "pg_dbz_event"
     }
   }

   说明

   默认需要为每个表创建一个Topic，以上配置对Topic做了聚合。

2. 添加connector。

   curl -i -X POST -H "Accept:application/json" -H "Content-Type:application/json" 'http://localhost:8083/connectors' -d @config/postgresql-connector.json

   成功添加后，能够在Kafka的Topic中查询到全量数据。

   在 Kafka 实例的消息查询页签，选择查询方式为按位点查询，分区选择 0，起始位点设为 0，单击查询。结果显示两条 Debezium CDC 消息（位点 0 和 1），Key 中 __dbz__physicalTableIdentifier 分别为 polardb.public.t1 和 polardb.public.t2，确认全量数据已同步至 Kafka。

3. 在PolarDB集群的dbz_db库中执行以下DML语句：

   INSERT INTO public.t1(a, b, c) VALUES(2, 'b', now());
   UPDATE public.t1 SET b = 'c' WHERE a = 1;
   DELETE FROM public.t1 WHERE a = 2;
   INSERT INTO public.t1(a, b, c) VALUES(4, 'd', now());
   INSERT INTO public.t2(a, b, c) VALUES(2, 'b', now());
   UPDATE public.t2 SET b = 'c' WHERE a = 1;
   DELETE FROM public.t2 WHERE a = 2;
   INSERT INTO public.t2(a, b, c) VALUES(4, 'd', now());

   能够在Kafka的Topic中查询到增量数据。

   在消息队列的消息查询页面，将查询方式设置为按位点查询，选择目标分区和起始位点后单击查询。查询结果显示 t1 和 t2 表的 DML 操作已被 Debezium 采集为 CDC 消息，每条消息的 Key 包含表标识符（如 polardb.public.t1）及主键值，Value 包含 Debezium 格式的变更事件（含 before/after 字段），其中 DELETE 操作对应的 Value 为 0 Bytes。