实时计算Flink版：Postgres CDC（公测中）

背景信息

特色功能

Postgres CDC连接器接入CDC增量快照框架（实时计算引擎VVR 8.0.6及以上版本）。Postgres CDC读取历史全量数据后，自动切换到WAL变更日志读取，保证不多读也不少读数据。即使发生故障，也能保证Exactly Once语义处理数据。Postgres CDC源表提供了并发读取全量数据，无锁读取和断点续传的能力。

作为源表，功能与优势详情如下：

• 流批一体，支持读取全量和增量数据，无需维护两套流程。

• 支持并发读取全量数据，性能水平扩展。

• 全量读取无缝切换增量读取，自动缩容，节省计算资源。

• 全量阶段读取支持断点续传，更稳定。

• 无锁读取全量数据，不影响线上业务。

前提条件

Postgres CDC连接器通过PostgreSQL数据库的逻辑复制读取CDC变更流数据，支持阿里云RDS PostgreSQL、Amazon RDS PostgreSQL和自建PostgreSQL。

完成配置后确保有下列的条件：

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

• 订阅表的REPLICA IDENTITY为FULL（发出的插入和更新操作事件包含表中所有列的旧值），以保障该表数据同步的一致性。

  说明

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

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

• 确保账户系统权限为SUPERUSER或者同时拥有LOGIN和REPLICATION权限，并且具有订阅表的SELECT权限用于全量数据查询。

注意事项

• 仅实时计算引擎8.0.6及以上版本支持Postgres CDC增量快照功能。

• 请及时管理Replication Slot，以免出现磁盘空间浪费的问题。

  为了防止在Flink作业重启过程中由于Checkpoint对应的WAL（Write-Ahead Log）段被清除而引发数据丢失，Flink作业不会自动移除Replication Slot。因此，如果确认特定的Flink作业不会再次启动，应当手动删除相关的Replication Slot，以释放其占用的资源。另外，如果PostgreSQL的Replication Slot的确认位点长时间不向前推进，PostgreSQL不会清理该槽位点之后的WAL条目，这可能会导致未使用的WAL积累而占用过多的磁盘空间。

• 开启增量快照时，Postgres CDC连接器必须开启Checkpoint，并且Source表必须声明主键。Source多并发读取全量数据时会创建多个临时的Replication Slot。

  不开启增量快照读取的PostgreSQL CDC Source仅支持单一并发，因此只需要一个全局Slot。当开启增量快照时，PostgreSQL CDC Source在全量阶段所需的最大Slot数量为Source数量 * 并发数 + 1。进入增量阶段后，系统自动回收在全量阶段创建的Slot，仅保留一个全局Slot。如果Slot数量有限，需要控制全量阶段的并发数量，这样做的缺点是会降低读取速度。如果下游算子或存储支持幂等性，可以启用scan.incremental.snapshot.backfill.skip = true以跳过全量阶段的日志读取，这样做的缺点是仅能提供至少一次（At-Least Once）的语义保证。如果SQL要做聚合、关联等操作，不建议跳过全量阶段日志的读取。

• 不开启增量快照时，Postgres CDC连接器不支持在全表扫描阶段执行Checkpoint。

  不开启增量快照时，如果您的作业在全表扫描阶段触发Checkpoint，则可能由于Checkpoint超时导致作业Failover。因此，建议您在其他配置中配置如下参数，具体操作请参见如何配置作业运行参数？。避免在全量同步阶段由于Checkpoint超时导致Failover。

  execution.checkpointing.interval: 10min
  execution.checkpointing.tolerable-failed-checkpoints: 100
  restart-strategy: fixed-delay
  restart-strategy.fixed-delay.attempts: 2147483647

  相关的参数说明详情如下表所示。

  参数	说明	备注
  execution.checkpointing.interval	Checkpoint的时间间隔。	单位是Duration类型，例如10min或30s。
  execution.checkpointing.tolerable-failed-checkpoints	容忍Checkpoint失败的次数。	该参数的取值与Checkpoint调度间隔时间的乘积就是允许的快照读取时间。 说明 如果表特别大，建议将该参数值配置得大一些。
  restart-strategy	重启策略。	参数取值如下： fixed-delay：固定延迟重启策略。 failure-rate：故障率重启策略。 exponential-delay：指数延迟重启策略。 详情请参见Restart Strategies。
  restart-strategy.fixed-delay.attempts	固定延迟重启策略下，尝试重启的最大次数。	无。

语法结构

CREATE TABLE postgrescdc_source (
  shipment_id INT,
  order_id INT,
  origin STRING,
  destination STRING,
  is_arrived BOOLEAN
) WITH (
  'connector' = 'postgres-cdc',
  'hostname' = '<yourHostname>',
  'port' = '5432',
  'username' = '<yourUserName>',
  'password' = '<yourPassWord>',
  'database-name' = '<yourDatabaseName>',
  'schema-name' = '<yourSchemaName>',
  'table-name' = '<yourTableName>'
);

WITH参数

类型映射

使用示例

CREATE TABLE source (
  id INT NOT NULL,
  name STRING,
  description STRING,
  weight DECIMAL(10,3)
) WITH (
  'connector' = 'postgres-cdc',
  'hostname' = '<host name>',
  'port' = '<port>',
  'username' = '<user name>',
  'password' = '<password>',
  'database-name' = '<database name>',
  'schema-name' = '<schema name>',
  'table-name' = '<table name>'
);

SELECT * FROM source;

相关文档

• 实时计算Flink版支持的连接器列表，请参见支持的连接器。

• 将数据写入PolarDB PostgreSQL版（Oracle语法兼容1.0）结果表，请参见PolarDB PostgreSQL版（Oracle语法兼容1.0）。

• 如果您需要读写RDS MySQL、PolarDB for MySQL或者自建MySQL数据库，请使用MySQL连接器。