PostgreSQL CDC Pipeline Source Connector 用于读取 PostgreSQL 表的初始快照和后续 WAL 增量变更，并将变更事件传递给 Flink CDC Pipeline 下游。

功能概述 #

版本兼容性 #

特色功能 #

• 全量加增量同步：initial 模式先读取一致性快照，再持续读取 WAL。
• 并行增量快照：将表快照拆分为多个 Chunk，提高全量读取并行度和恢复能力。
• 多表选择：使用 database.schema.table 和正则表达式选择或排除表。
• 多种启动位置：社区基线支持全量加增量、仅最新增量、复制槽已提交位置和仅快照。
• 源端元数据：可输出操作时间、database、schema 和 table 名称供 Transform 或下游使用。
• Schema Change 推断：使用 pgoutput 时，可以根据 Relation 消息与缓存 Schema 的差异推断加列、删列、重命名列和修改列类型事件。

前提条件 #

1. 已准备实现 PostgreSQL 兼容协议并提供逻辑复制、WAL、Replication Slot 等能力的实例，并确保 Flink 资源池与数据库网络可达。
2. PostgreSQL 已启用逻辑复制，wal_level 设置为 logical，并为 max_replication_slots、max_wal_senders 保留足够容量。
3. 已准备具有目标表读取权限和逻辑复制权限的数据库用户，并允许 Flink 资源池通过 pg_hba.conf 或云数据库白名单连接。
4. 已为目标表配置主键或适当的 Replica Identity，以确保 UPDATE、DELETE 可被正确识别。
5. 已为每个运行中的作业规划唯一的 Replication Slot；使用 pgoutput 时已确认 Connector 用户能够使用或创建 Publication。
6. 已使用 Flink 加密变量保存 PostgreSQL 用户名和密码。
7. 已确认同一 Pipeline 中匹配的所有表属于同一个 PostgreSQL database。

使用限制 #

• Connector 仅支持作为 Flink CDC Pipeline Source。
• 一次 Pipeline 只能连接一个 PostgreSQL database；tables 中的 database 必须相同。
• Schema Evolution 的处理结果受 pipeline.schema.change.behavior 和下游 Sink 能力影响。
• schema-change.enabled 默认关闭，且仅支持与 pgoutput 配合使用。
• TRUNCATE TABLE 和 DROP TABLE 事件会被忽略，不会传递给下游。
• Replication Slot 不能同时被多个活跃消费者使用；每个并行运行的作业应使用独立 Slot。
• Connector 在 Slot 不存在且用户权限充足时自动创建 Slot；默认不会在作业停止时删除 Slot。
• 使用 pgoutput 时，Connector 可以在权限充足时自动创建 Publication；已有 Publication 会按其现有定义复用，作业停止时不会自动删除。
• Slot 长期不消费会阻止旧 WAL 清理，可能持续占用 PostgreSQL 磁盘。
• latest-offset 和 committed-offset 跳过初始快照，启动位置之前的数据不会补读。
• snapshot 只读取快照，完成后退出，不持续读取 WAL。
• PostgreSQL 多维数组不受支持，仅支持一维数组。
• TIME 固定映射为 TIME(3)；高精度时间和 Decimal 模式变更可能影响下游类型。
• Sequence 和 Large Object 不属于行级 WAL 数据同步范围。

语法结构 #

source:
  type: postgres
  name: PostgreSQL Source 
  hostname: <postgres_host>
  port: 5432
  username: ${secret_values.postgres-username}
  password: ${secret_values.postgres-password}
  tables: <database>.<schema>.<table_pattern>
  slot.name: <replication_slot_name>
  decoding.plugin.name: pgoutput
  schema-change.enabled: true
  scan.startup.mode: initial

sink:
  type: values
  name: Values Sink

pipeline:
  name: PostgreSQL CDC Pipeline
  parallelism: 2
  local-time-zone: Asia/Shanghai
  schema.change.behavior: lenient

参数说明 #

连接器配置项 #

以下是与 Slot、Publication 生命周期相关的常用 Debezium 透传参数。

类型映射 #

使用示例 #

读取 PostgreSQL 全量和增量数据 #

以下示例读取一个 PostgreSQL database 中匹配表的初始快照和后续 WAL 变更，并写入 Values Sink 进行验证。运行前请替换地址、表名、Slot 和加密变量名称。

source:
  type: postgres
  name: PostgreSQL Source 
  hostname: <postgres_host>
  port: 5432
  username: ${secret_values.postgres-username}
  password: ${secret_values.postgres-password}
  tables: <database>.<schema>.<table_pattern>
  tables.exclude: <database>.<schema>.<excluded_table_pattern>
  slot.name: <replication_slot_name>
  decoding.plugin.name: pgoutput
  schema-change.enabled: true
  scan.startup.mode: initial
  metadata.list: op_ts,table_name,database_name,schema_name
  heartbeat.interval: 30s

sink:
  type: values
  name: Values Sink

pipeline:
  name: PostgreSQL CDC Pipeline
  parallelism: 2
  local-time-zone: Asia/Shanghai
  schema.change.behavior: lenient

预期结果： 作业先输出匹配表的已有记录，然后持续输出 INSERT、UPDATE、DELETE 事件。快照记录的 op_ts 为 0，增量记录包含实际变更时间。

运行机制 #

快照与增量衔接 #

initial 模式将表拆分为多个 Chunk 并行读取。快照期间同时记录 WAL 位置；所有快照 Chunk 完成后，作业继续从对应位置消费增量变更，避免快照和增量之间出现缺口。

Replication Slot 与 LSN #

Replication Slot 保存逻辑消费位置并阻止所需 WAL 被提前清理。scan.lsn-commit.checkpoints-num-delay 控制 Checkpoint 成功后延迟提交 LSN 的数量。Source 基于 Checkpoint 保存快照进度和 WAL 读取位置，可以在故障恢复后继续读取；端到端一致性仍取决于下游 Sink 的交付语义。
Slot 不存在且 Connector 用户具有 REPLICATION 等必要权限时，Connector 会自动创建；默认停止作业不会删除 Slot。多个作业不能共用同一个 Slot。
使用 pgoutput 时，Connector 可以根据 Debezium Publication 配置自动创建 Publication；若 Publication 已存在，则按现有定义使用。自动创建需要数据库 CREATE 权限以及目标表所需权限。生产环境可以提前创建 Slot 和 Publication，以便采用更严格的权限和生命周期管理。

Schema Change #

schema-change.enabled: true 且解码插件为 pgoutput 时，Source 会比较 pgoutput Relation 消息与缓存 Schema，推断加列、删列、重命名列和修改列类型事件。Pipeline 默认使用 lenient 模式，以避免数据丢失的方式转换 Schema Change；实际能否在下游执行仍取决于 Sink 的 Schema Evolution 能力。TRUNCATE TABLE 和 DROP TABLE 事件会被忽略。

验证与排障 #

验证方法 #

1. 验证 PostgreSQL wal_level、Replication Slot、Publication、用户权限和网络连通性。
2. 提交作业并确认进入运行状态，观察快照相关指标逐步完成。
3. 对测试表执行 INSERT、UPDATE、DELETE，确认下游收到对应事件。
4. 在测试表执行加列、修改列类型等 Schema Change，确认下游行为符合 schema.change.behavior 和 Sink 能力。
5. 等待成功 Checkpoint 后执行受控重启，检查快照或 WAL 读取位置是否正确恢复。
6. 查询 pg_replication_slots，确认 Slot 处于预期状态，confirmed_flush_lsn 持续推进。
7. 检查 PostgreSQL WAL 磁盘占用，确认作业停止或异常时没有持续无界增长。

常见问题 #

注意事项 #

• 每个并行运行的作业使用独立 Replication Slot，避免 Slot 被多个消费者争用。
• Slot 和 Publication 默认不会随作业停止自动删除。作业长期停止前应评估 Slot 对 WAL 保留和磁盘容量的影响；不要直接删除仍被使用或仍需用于状态恢复的 Slot。
• latest-offset、committed-offset 会跳过快照，使用前确认历史数据已经同步。
• 生产表变更前应评估 schema.change.behavior 和下游 Sink 的兼容性；类型转换在部分模式下可能改变字段结构或造成数据损失。
• 开启 schema-change.enabled 时必须使用 pgoutput，并建议先在测试表验证目标 DDL 和下游 Sink 行为。
• TRUNCATE TABLE 和 DROP TABLE 事件会被忽略，不会对下游执行清表或删表。
• debezium.*、jdbc.properties.* 会改变类型、快照和连接行为，仅使用当前版本已验证的参数。
• 不要在 YAML 中填写明文密码，应使用 Flink 加密变量。

相关文档 #

• Apache Flink CDC master：PostgreSQL Pipeline Connector
• Apache Flink CDC 3.4：Schema Evolution
• PostgreSQL：逻辑复制配置
• PostgreSQL：Publication 与 Replica Identity
• Debezium：PostgreSQL Connector
• 火山引擎流式计算 Flink 版：Postgres CDC