Postgres CDC（公测中）

Replication Slot说明

Flink PostgreSQL CDC 作业依赖 Replication Slot 来确保 WAL（Write-Ahead Log）不被过早清理，从而保障数据一致性。但若管理不当，可能引发磁盘空间浪费或数据读取延迟等问题。请遵循以下建议：

• 请及时清理不再使用的 Slot

  • Flink 不会自动删除 Replication Slot，即使作业已停止（尤其无状态重启场景），以防止因 WAL 被清除而导致数据丢失。

  • 若确认某作业不再启动，请手动删除其关联的 Replication Slot，释放磁盘空间。

    重要

    生命周期管理：将 Replication Slot 视为作业资源的一部分，随作业启停同步管理。

• 避免复用旧 Slot

  • 新作业应使用新的 Slot Name，而非复用旧 Slot。复用可能导致作业启动后需回溯大量历史 WAL，延迟读取最新数据。

  • PostgreSQL的逻辑复制要求一个 Slot 仅能被一个连接使用，不同作业必须使用不同的 Slot 名称。

    重要

    命名规范：自定义slot.name时，避免使用带数字后缀的名称（如 my_slot_1），以防与临时 Slot 冲突。

• 启用增量快照下的Slot行为

  • 前提条件：必须启用checkpoint，且Source 表必须声明主键。

  • Slot创建规则：

    • 未开启增量快照：仅支持单并发，使用 1 个全局 Slot。

    • 开启增量快照：

      • 全量阶段：每个 Source 并发子任务会创建一个临时 Slot，命名格式为${slot.name}_${task_id}。

      • 增量阶段：自动回收所有临时 Slot，仅保留 1 个全局 Slot。

  • 最大Slot数量：Source 并发数 + 1（全量阶段）

• 资源与性能

  • 若 PostgreSQL 的 Slot 数量或磁盘空间受限，应适当降低全量阶段的并发度（减少临时 Slot 数量），但会牺牲全量读取速度。

  • 若下游支持幂等写入，可设置：scan.incremental.snapshot.backfill.skip = true，跳过全量阶段的 Binlog 回溯，加快启动速度。

    此配置仅提供 At-Least-Once 语义。不适用于含聚合、维表 Join 等状态计算的作业（可能丢失中间状态所需的历史变更）。

• 不开启增量快照时，不支持在全表扫描阶段执行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	固定延迟重启策略下，尝试重启的最大次数。	无。

复用postgres订阅

Postgres CDC 使用 pgoutput 时，会通过 Publication 决定哪些表的变更可以被读取。

如果多个作业共用同一个 Publication，请确认该 Publication 已包含所有作业需要读取的表，否则部分表的增量变更可能无法被同步。

注意事项

debezium.publication.autocreate.mode 默认值为 all_tables。在该模式下：

• 如果 Publication 不存在，连接器会自动创建 FOR ALL TABLES 的 Publication。

• 如果 Publication 已存在，连接器只会直接复用，不会检查或修改其中的表范围。

如果配置为 filtered，连接器会按当前作业的表过滤规则创建或更新 Publication。多个作业共用同一个 Publication 时，后启动的作业可能会覆盖表范围，影响其他作业读取。

推荐做法

1. 生产环境建议手动创建和维护 Publication，并关闭自动创建/更新

   -- 创建一个名为 my_flink_pub 的发布，包含所有表（或指定表，每个作业创一个Publication）
   CREATE PUBLICATION my_flink_pub FOR TABLE table_a, table_b;
   -- 或者更简单，包含库里所有表
   CREATE PUBLICATION my_flink_pub FOR ALL TABLES;

   说明

   不建议订阅全库全表，如果数据库非常大，表非常多，这会造成网络带宽浪费和Flink 端 CPU 消耗。

2. 添加Flink配置

   • debezium.publication.name = 'my_flink_pub' （指定Publication名称）

   • debezium.publication.autocreate.mode = 'disabled' (禁止Flink启动时尝试创建或修改 Publication)

这样可以避免多个作业互相覆盖 Publication 配置。

语法结构

CREATE TABLE postgrescdc_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>',
  'decoding.plugin.name'= 'pgoutput',
  'scan.incremental.snapshot.enabled' = 'true',
  -- skip backfill 可以加速读取和减少资源，但是会有数据重复。如果下游幂等，建议开启
  'scan.incremental.snapshot.backfill.skip' = 'false',
  -- 生产上建议设置为filter或者disabled, 手动管理publication而非flink
  'debezium-publication.autocreate.mode' = 'disabled'
  -- 多Source时，为不同的Source配置不同的publication。
  --'debezium.publication.name' = 'my_flink_pub'
);

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;

数据摄入

自实时计算引擎11.4版本起，PostgreSQL连接器作为数据源可以在数据摄入YAML作业中使用。

语法结构

source:
  type: postgres
  name: PostgreSQL Source
  hostname: localhost
  port: 5432
  username: pg_username
  password: pg_password
  tables: db.scm.tbl
  slot.name: test_slot
  scan.startup.mode: initial
  server-time-zone: UTC
  connect.timeout: 120s
  decoding.plugin.name: decoderbufs

sink:
  type: ...

配置项