Postgres CDC源表

功能描述

Postgres的CDC源表，即Postgres的流式源表，用于依次读取PostgreSQL数据库全量快照数据和变更数据，保证不多读一条也不少读一条数据。即使发生故障，也能采用Exactly Once方式处理。

前提条件

• PostgreSQL CDC要求Postgre版本为9.6或者10，11，12。
• 要与实例建立增强型跨源连接，且用户可以根据实际所需设置相应安全组规则。
  • 如何建立增强型跨源连接，请参考《数据湖探索用户指南》中增强型跨源连接章节。
  • 如何设置安全组规则，请参见《虚拟私有云用户指南》中“安全组”章节。
• Flink跨源开发场景中直接配置跨源认证信息存在密码泄露的风险，优先推荐您使用DLI提供的跨源认证。

  跨源认证简介及操作方法请参考跨源认证简介。

注意事项

• 创建Flink OpenSource SQL作业时，在作业编辑界面的“运行参数”处，“Flink版本”需要选择“1.12”，勾选“保存作业日志”并设置保存作业日志的OBS桶，方便后续查看作业日志。
• PostgreSQL的版本不能低于PostgreSQL 11。
• 若Postgres表有update等操作，需要在PostgreSQL中执行下列语句。注意：test.cdc_order需要修改为实际的数据库和表。

  ALTER TABLE test.cdc_order REPLICA IDENTITY FULL

• 使用前请确认当前PostgreSQL是否包含默认的插件，可在PostgreSQL中使用下述语句查询当前插件。

  SELECT name FROM pg_available_extensions;

  若不包含默认插件名“decoderbufs”，则需要在创建PostgreSQL CDC源表中配置参数“decoding.plugin.name”，该参数指定PostgreSQL中已有的插件。

语法格式

create table postgresCdcSource (
  attr_name attr_type 
  (',' attr_name attr_type)* 
  (','PRIMARY KEY (attr_name, ...) NOT ENFORCED)
)
with (
  'connector' = 'postgres-cdc',   
  'hostname' = 'PostgresHostname',
  'username' = 'PostgresUsername',
  'password' = 'PostgresPassword',
  'database-name' = 'PostgresDatabaseName',
  'schema-name' = 'PostgresSchemaName',
  'table-name' = 'PostgresTableName'
);

参数说明

示例

该示例是利用Postgres-CDC实时读取RDS PostgreSQL中的数据，并写入到Print结果表中，其具体步骤如下（当前示例使用的数据库引擎版本是RDS PostgreSQL 11.11）：

1. 参考增强型跨源连接，根据PostgreSQL所在的虚拟私有云和子网创建相应的增强型跨源，并绑定所要使用的Flink弹性资源池。
2. 设置PostgreSQL的安全组，添加入向规则使其对Flink的队列网段放通。参考测试地址连通性根据PostgreSQL的地址测试队列连通性。若能连通，则表示跨源已经绑定成功，否则表示未成功。
3. 在PostgreSQL中创建数据库flink，并创建名为test的schema。
4. 在PostgreSQL中flink数据库的test schema下创建表名为cdc_order的表，SQL语句参考如下：

   create table test.cdc_order(
     order_id VARCHAR,
     order_channel VARCHAR,
     order_time VARCHAR,
     pay_amount FLOAT8,
     real_pay FLOAT8,
     pay_time VARCHAR,
     user_id VARCHAR,
     user_name VARCHAR,
     area_id VARCHAR,
     primary key(order_id)
   );

5. 在PostgreSQL中执行下列SQL语句。如果不执行如下命令，后续Flink作业将会运行报错，具体报错信息详情参见错误信息。

   ALTER TABLE test.cdc_order REPLICA IDENTITY FULL

6. 参考创建Flink OpenSource作业，创建flink opensource sql作业，输入以下作业脚本，提交运行作业。
   注意：创建作业时，在作业编辑界面的“运行参数”处，“Flink版本”选择“1.12”，勾选“保存作业日志”并设置保存作业日志的OBS桶，方便后续查看作业日志。如下脚本中的加粗参数请根据实际环境修改。

   create table postgresCdcSource(
     order_id string,
     order_channel string,
     order_time string,
     pay_amount double,
     real_pay double,
     pay_time string,
     user_id string,
     user_name string,
     area_id STRING,
     primary key (order_id) not enforced
   ) with (
     'connector' = 'postgres-cdc',
     'hostname' = 'PostgresHostname',
     'username' = 'PostgresUsername',
     'password' = 'PostgresPassword',
     'database-name' = 'flink',
     'schema-name' = 'test',
     'table-name' = 'cdc_order'
   );
   
   create table printSink(
     order_id string,
     order_channel string,
     order_time string,
     pay_amount double,
     real_pay double,
     pay_time string,
     user_id string,
     user_name string,
     area_id STRING,
     primary key(order_id) not enforced
   ) with (
     'connector' = 'print'
   );
   
   insert into printSink select * from postgresCdcSource;

7. 在PostgreSQL中执行以下命令：

   insert into test.cdc_order
     (order_id,
     order_channel,
     order_time,
     pay_amount,
     real_pay,
     pay_time,
     user_id,
     user_name,
     area_id) values
     ('202103241000000001', 'webShop', '2021-03-24 10:00:00', '100.00', '100.00', '2021-03-24 10:02:03', '0001', 'Alice', '330106'),
     ('202103251202020001', 'miniAppShop', '2021-03-25 12:02:02', '60.00', '60.00', '2021-03-25 12:03:00', '0002', 'Bob', '330110');
   
   update test.cdc_order set order_channel = 'webShop' where order_id = '202103251202020001';
   
   delete from test.cdc_order where order_id = '202103241000000001';

8. 按照如下方式查看taskmanager.out文件中的数据结果：
   1. 登录DLI管理控制台，选择“作业管理 > Flink作业”。
   2. 单击对应的Flink作业名称，选择“运行日志”，单击“OBS桶”，根据作业运行的日期，找到对应日志的文件夹。
   3. 进入对应日期的文件夹后，找到名字中包含“taskmanager”的文件夹进入，下载获取taskmanager.out文件查看结果日志。

   数据结果参考如下：

   +I(202103241000000001,webShop,2021-03-24 10:00:00,100.0,100.0,2021-03-24 10:02:03,0001,Alice,330106)
   +I(202103251202020001,miniAppShop,2021-03-25 12:02:02,60.0,60.0,2021-03-25 12:03:00,0002,Bob,330110)
   -U(202103251202020001,miniAppShop,2021-03-25 12:02:02,60.0,60.0,2021-03-25 12:03:00,0002,Bob,330110)
   +U(202103251202020001,webShop,2021-03-25 12:02:02,60.0,60.0,2021-03-25 12:03:00,0002,Bob,330110)
   -D(202103241000000001,webShop,2021-03-24 10:00:00,100.0,100.0,2021-03-24 10:02:03,0001,Alice,330106)

常见问题

• Q：Flink作业运行失败，作业运行日志中如下报错信息，应该怎么解决？

  org.postgresql.util.PSQLException: ERROR: logical decoding requires wal_level >= logical

• A：需要调节PostgreSQL的配置参数wal_level为logical，并重新启动。具体操作可以参考设置wal_level参数为logical。

  PostgreSQL参数修改完成后，需要重启下RDS PostgreSQL实例，使得参数生效。