Flink PostgreSQL CDC配置和常见问题

部分内容来源：流计算 Oceanus 数据库 PostgreSQL CDC-SQL 开发指南-文档中心-腾讯云

介绍

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

版本说明

使用范围

PostgreSQL CDC 只支持作为源表。支持的 PostgreSQL 版本为9.6及以上版本。

PostgreSQL 数据库配置准备

postgresql.conf 配置

# 更改wal发送最大进程数（默认值为10），这个值和上面的solts设置一样
max_wal_senders = 10    # max number of walsender processes
wal_keep_segments = 2   # in logfile segments; 0 disables
# 中断那些停止活动超过指定毫秒数的复制连接，可以适当设置大一点（默认60s）
wal_sender_timeout = 60s    # in milliseconds; 0 disables
#更改solts最大数量（默认值为10），flink-cdc默认一张表占用一个slots
max_replication_slots = 10   # max number of replication slots
#指定为logical
wal_level = logical         # minimal, replica, or logical
# - Archiving -
archive_mode = on
archive_command = 'test ! -f /var/lib/pgsql/11/arch/%f && cp %p /var/lib/pgsql/11/arch/%f'

pg_hba.conf

 添加

# IPv4 local connections:
 
host    all                 all              0.0.0.0/0               md5
 
host    replication         all              0.0.0.0/0               md5

FlinkCDC DDL 定义

CREATE TABLE postgres_cdc_source_table (
  id INT,
  name STRING,
  PRIMARY KEY (`id`) NOT ENFORCED -- 如果要同步的数据库表定义了主键, 则这里也需要定义
) WITH (
  'connector' = 'postgres-cdc',             -- 固定值 'postgres-cdc'
  'hostname' = 'yourHostname',              -- 数据库的 IP
  'port' = '5432',                          -- 数据库的访问端口
  'username' = 'yourUserName',              -- 数据库访问的用户名（需要提供 REPLICATION、LOGIN、SCHEMA、DATABASE、SELECT权限）
  'password' = 'yourPassWord',              -- 数据库访问的密码
  'database-name' = 'yourDatabaseName',     -- 需要同步的数据库
  'schema-name' = 'yourSchemaName',         -- 需要同步的数据表所属schema (支持正则表达式)
  'table-name' = 'yourTableName',           -- 需要同步的数据表名 (支持正则表达式)
  'debezium.slot.name' = 'customslotname'   -- 定义一个唯一slot名称，可以包含小写字母、数字和下划线字符
);

WITH 参数

类型映射

Postgres CDC 和 Flink 字段类型对应关系如下：

代码示例

CREATE TABLE postgres_cdc_source_table (
  id INT,
  name STRING,
  PRIMARY KEY (`id`) NOT ENFORCED -- 如果要同步的数据库表定义了主键, 则这里也需要定义
) WITH (
  'connector' = 'postgres-cdc',             -- 固定值 'postgres-cdc'
  'hostname' = 'yourHostname',              -- 数据库的 IP
  'port' = '5432',                          -- 数据库的访问端口
  'username' = 'yourUserName',              -- 数据库访问的用户名（需要提供 REPLICATION、LOGIN、SCHEMA、DATABASE、SELECT权限）
  'password' = 'yourPassWord',              -- 数据库访问的密码
  'database-name' = 'yourDatabaseName',     -- 需要同步的数据库
  'schema-name' = 'yourSchemaName',         -- 需要同步的数据表所属schema (支持正则表达式)
  'table-name' = 'yourTableName',           -- 需要同步的数据表名 (支持正则表达式)
  'debezium.slot.name' = 'customslotname'   -- 定义一个唯一slot名称，可以包含小写字母、数字和下划线字符
);
 
CREATE TABLE `print_table` (
  `id` INT,
  `name` STRING
) WITH (
 'connector' = 'print'
);
insert into print_table select * from postgres_cdc_source_table;

来源： 流计算 Oceanus 数据库 PostgreSQL CDC-SQL 开发指南-文档中心-腾讯云

注意事项

用户权限

用来同步的用户至少具有 REPLICATION、LOGIN、SCHEMA、DATABASE、SELECT 权限。

CREATE ROLE debezium_user REPLICATION LOGIN; 
GRANT USAGE ON SCHEMA schema_name TO debezium_user;
GRANT USAGE ON DATABASE schema_name TO debezium_user;
GRANT SELECT ON scheam_name.table_name, scheam_name.table_name TO debezium_user;

postgres CDC 遇到的错误：

1. number of requested standby connections exceeds max_wal_senders (currently 1)。

max_wal_senders 须大于 wal_keep_segments

2.org.postgresql.util.PSQLException: ERROR: could not access file "decoderbufs": No such file or directory。

如果是 table sql 方式，pgoutput是 PostgreSQL 10+ 中的标准逻辑解码输出插件。需要设置一下。添加如下配置

WITH(

       'decoding.plugin.name' = 'pgoutput'

)

3 .io.debezium.DebeziumException: Failed to start replication stream at LSN{0/1100AA50}; when setting up multiple connectors for the same database host, please make sure to use a distinct replication slot name for each.

org.postgresql.util.PSQLException: ERROR: replication slot "flink" is active for PID 26632

同一个数据源存在两个复制槽，因此需要为每个指定不同的名称。

WITH(

       'debezium.slot.name' = '自定义名称'

)