跳到主要内容

Postgres Connector

Postgres CDC Pipeline 连接器允许从 Postgres 数据库读取快照数据和增量数据，并提供端到端的整库数据同步能力。本文描述了如何设置 Postgres CDC Pipeline 连接器。

示例​

从 Postgres 读取数据同步到 Fluss 的 Pipeline 可以定义如下：

source:
   type: postgres
   name: Postgres Source
   hostname: 127.0.0.1
   port: 5432
   username: admin
   password: pass
   # 需要确保所有的表来自同一个database
   tables: adb.\.*.\.*
   decoding.plugin.name:  pgoutput
   slot.name: pgtest
   schema-change.enabled: true

sink:
  type: fluss
  name: Fluss Sink
  bootstrap.servers: localhost:9123
  # Security-related properties for the Fluss client
  properties.client.security.protocol: sasl
  properties.client.security.sasl.mechanism: PLAIN
  properties.client.security.sasl.username: developer
  properties.client.security.sasl.password: developer-pass

pipeline:
   name: Postgres to Fluss Pipeline
   parallelism: 4
   schema.change.behavior: lenient

连接器配置项​

Option Required Default Type Description
hostname required (none) String Postgres 数据库服务器的 IP 地址或主机名。
port optional 5432 Integer Postgres 数据库服务器的整数端口号。
username required (none) String 连接到 Postgres 数据库服务器时要使用的 Postgres 用户的名称。
password required (none) String 连接 Postgres 数据库服务器时使用的密码。
tables required (none) String 需要监视的 Postgres 数据库的表名。表名支持正则表达式，以监视满足正则表达式的多个表。
需要确保所有的表来自同一个数据库。
需要注意的是，点号（.）被视为数据库、模式和表名的分隔符。如果需要在正则表达式中使用点（.）来匹配任何字符，必须使用反斜杠对点进行转义。
例如，bdb.user_schema_[0-9].user_table_[0-9]+, bdb.schema_\.*.order_\.*
slot.name required (none) String 为从特定插件以流式传输方式获取某个数据库/模式的变更数据，所创建的 Postgre 逻辑解码槽（logical decoding slot）的名称。服务器使用这个槽（slot）将事件流式传输给你要配置的连接器（connector）。
复制槽名称必须符合 PostgreSQL 复制插槽的命名规则, 其规则如下："Each replication slot has a name, which can contain lower-case letters, numbers, and the underscore character."
decoding.plugin.name optional (none) String The name of the Postgres logical decoding plug-in installed on the server. Supported values are decoderbufs and pgoutput.
tables.exclude optional (none) String 需要排除的 Postgres 数据库的表名，参数会在 tables 参数后发生排除作用。表名支持正则表达式，以排除满足正则表达式的多个表。
用法和 tables 参数相同
server-time-zone optional (none) String 数据库服务器中的会话时区，例如："Asia/Shanghai"。 它控制 Postgres 中的时间戳类型如何转换为字符串。 更多请参考 这里。 如果没有设置，则使用 ZoneId.systemDefault() 来确定服务器时区。
scan.incremental.snapshot.chunk.size optional 8096 Integer 表快照的块大小（行数），读取表的快照时，捕获的表被拆分为多个块。
scan.snapshot.fetch.size optional 1024 Integer 读取表快照时每次读取数据的最大条数。
scan.startup.mode optional initial String Postgres CDC 消费者可选的启动模式， 合法的模式为 "initial"，"latest-offset"，"committed-offset" 和 ""snapshot"。
scan.incremental.snapshot.backfill.skip optional false Boolean 是否在快照读取阶段跳过 backfill。
如果跳过 backfill，快照阶段捕获表的更改将在稍后的 wal log 读取阶段被回放，而不是合并到快照中。
警告：跳过 backfill 可能会导致数据不一致，因为快照阶段发生的某些 wal log 事件可能会被重放（仅保证 at-least-once）。 例如，更新快照阶段已更新的值，或删除快照阶段已删除的数据。这些重放的 wal log 事件应进行特殊处理。
scan.lsn-commit.checkpoints-num-delay optional 3 Integer 在开始提交 LSN 偏移量之前，允许的检查点延迟次数。
检查点的 LSN 偏移量将以滚动方式提交，最早的那个检查点标识符将首先从延迟的检查点中提交。
connect.timeout optional 30s Duration 连接器在尝试连接到 Postgres 数据库服务器后超时前应等待的最长时间。该时长不能少于 250 毫秒。
connect.max-retries optional 3 Integer 连接器应重试以建立 Postgres 数据库服务器连接的最大重试次数。
connection.pool.size optional 20 Integer 连接池大小。
jdbc.properties.* optional 20 String 传递自定义 JDBC URL 属性的选项。用户可以传递自定义属性，如 'jdbc.properties.useSSL' = 'false'。
heartbeat.interval optional 30s Duration 用于跟踪最新可用 wal committed offset 偏移的发送心跳事件的间隔。
debezium.* optional (none) String 将 Debezium 的属性传递给 Debezium 嵌入式引擎，该引擎用于从 Postgres 服务器捕获数据更改。 例如：'debezium.snapshot.mode' = 'never'。 查看更多关于 Debezium 的 Postgres 连接器属性
scan.incremental.close-idle-reader.enabled optional false Boolean 是否在快照结束后关闭空闲的 Reader。此特性需要 flink 版本大于等于 1.14 并且 'execution.checkpointing.checkpoints-after-tasks-finish.enabled' 需要设置为 true。
若 flink 版本大于等于 1.15，'execution.checkpointing.checkpoints-after-tasks-finish.enabled' 默认值变更为 true，可以不用显式配置 'execution.checkpointing.checkpoints-after-tasks-finish.enabled' = true。
chunk-meta.group.size optional 1000 String 分块元数据的组大小，如果元数据大小超过该组大小，则元数据将被划分为多个组。
metadata.list optional (none) String 源记录中可读取的元数据列表，将传递给下游并在转换模块中使用，各字段以逗号分隔。可用的可读元数据包括：op_ts、table_name、database_name、schema_name。详见支持的元数据列。
scan.incremental.snapshot.unbounded-chunk-first.enabled optional false String 在快照读取阶段，是否优先分配无界分块。
这有助于降低在对最大无界分块进行快照时，TaskManager 发生内存溢出（OOM）错误的风险。
此为实验性选项，默认值为 false。
table-id.include-database optional false Boolean 是否在生成的 Table ID 中包含数据库名称。
如果设置为 true，Table ID 的格式为（数据库，模式，表）。
如果设置为 false，Table ID 的格式为（模式，表）。
默认值为 false。
scan.newly-added-table.enabled optional false Boolean 是否扫描新增的表。默认值为 false。 该选项仅在从 savepoint 或 checkpoint 恢复时生效， 会启用 SnapshotSplitAssigner#captureNewlyAddedTables() 代码路径， 以发现在 savepoint 时刻未被捕获但匹配 source tables 模式的新表。
scan.pre-epoch-timestamp.wall-clock-conversion.enabled optional false Boolean 是否对 1970-01-01（epoch）之前的 PostgreSQL "timestamp without time zone" 列值，按照数据库中存储的日期时间（wall clock）进行转换。
开启后，该值不会受运行任务的 JVM 时区影响，这在存在历史时区偏移的时区下很重要：当 JVM 时区为 Asia/Shanghai 时， 数据库中存储的 "1900-01-01 00:00:00.123" 否则会被读取为 "1900-01-01 00:05:43.123"。
关闭（默认值）时保持原有的转换行为。1970-01-01 及之后的值无论是否开启该选项都会被同样地转换， 且该选项不会改变列的数据类型。
schema-change.enabled optional false Boolean 是否开启 Postgres 源的 Schema 变更推导。开启后，连接器会通过对比 pgoutput Relation 消息与缓存的 Schema 来推导 Schema 变更事件（新增列、删除列、重命名列、修改列类型）。
需要将 decoding.plugin.name 设置为 pgoutput。
默认值为 false。

注意：

  1. 配置选项 tables 指定 Postgres CDC 需要采集的表，格式为 db.schema1.tabe1,db.schema2.table2，其中所有的 db 需要为同一个 db，这是因为 postgres 链接 url 中需要指定 dbname，目前 cdc 只支持链接一个 db。

启动模式​

配置选项 scan.startup.mode 指定 Postgres CDC 使用者的启动模式。有效枚举包括：

  • initial（默认）：在连接器首次启动时，会对被监控的数据库表执行一次初始快照（snapshot），然后继续读取复制插槽中的变更事件。。
  • latest-offset：在首次启动时不执行快照操作，而是直接从复制插槽的末尾开始读取，即只获取连接器启动之后发生的变更。
  • committed-offset：跳过快照阶段，从复制插槽中已确认的 confirmed_flush_lsn 偏移量开始读取事件，即从上次提交的位置继续读取变更。
  • snapshot: 只进行快照阶段，跳过增量阶段，快照阶段读取结束后退出。

可用的指标​

指标系统能够帮助了解分片分发的进展，下面列举出了支持的 Flink 指标 Flink metrics：

GroupNameTypeDescription
namespace.schema.tableisSnapshottingGauge表是否在快照读取阶段
namespace.schema.tableisStreamReadingGauge表是否在增量读取阶段
namespace.schema.tablenumTablesSnapshottedGauge已经被快照读取完成的表的数量
namespace.schema.tablenumTablesRemainingGauge还没有被快照读取的表的数据
namespace.schema.tablenumSnapshotSplitsProcessedGauge正在处理的分片的数量
namespace.schema.tablenumSnapshotSplitsRemainingGauge还没有被处理的分片的数量
namespace.schema.tablenumSnapshotSplitsFinishedGauge已经处理完成的分片的数据
namespace.schema.tablesnapshotStartTimeGauge快照读取阶段开始的时间
namespace.schema.tablesnapshotEndTimeGauge快照读取阶段结束的时间

注意：

  1. Group 名称是 namespace.schema.table，这里的 namespace 是实际的数据库名称，schema 是实际的 schema 名称，table 是实际的表名称。

支持的元数据列​

PostgreSQL CDC 连接器支持从源记录中读取元数据列。这些元数据列可以在转换操作中使用或传递给下游 Sink。

注意： 部分元数据信息也可以通过 Transform 表达式获取（例如 __namespace_name__、__schema_name__、__table_name__）。主要区别如下：

  • op_ts：仅可通过 metadata.list 获取 - 提供数据库中实际的操作时间戳。
  • table_name、database_name、schema_name：可通过 metadata.list 或 Transform 表达式获取。使用 metadata.list 可以直接将这些值传递给下游 Sink，无需编写转换规则，对于基本用例更加简单。

要启用元数据列，请使用逗号分隔的元数据列名称列表配置 metadata.list 选项：

source:
  type: postgres
  # ... 其他配置
  metadata.list: op_ts,table_name,database_name,schema_name

支持以下元数据列：

元数据列 数据类型 描述
op_ts BIGINT NOT NULL 数据变更事件在数据库中发生的时间戳（自纪元以来的毫秒数）。对于快照记录，此值为 0。
table_name STRING NOT NULL 包含变更行的表名称。替代方案：在 Transform 表达式中使用 __table_name__。
database_name STRING NOT NULL 包含变更行的数据库名称。替代方案：在 Transform 表达式中使用 __namespace_name__。
schema_name STRING NOT NULL 包含变更行的 Schema 名称。这是 PostgreSQL 特有的。替代方案：在 Transform 表达式中使用 __schema_name__。

使用示例：

source:
  type: postgres
  hostname: localhost
  port: 5432
  username: postgres
  password: postgres
  tables: mydb.public.orders
  slot.name: flink_slot
  metadata.list: op_ts,table_name,schema_name

transform:
  - source-table: mydb.public.orders
    projection: order_id, customer_id, op_ts, table_name, schema_name
    description: 在输出中包含元数据列

数据类型映射​

PostgreSQL type CDC type
BOOLEAN
BIT(1)
BOOLEAN
BIT( > 1) BYTES
SMALLINT
INT2
SMALLSERIAL
SERIAL2		 SMALLINT
INTEGER
SERIAL		 INT
BIGINT
BIGSERIAL
OID
 BIGINT
REAL
FLOAT4 		FLOAT
NUMERIC DECIMAL(38, 0)
DOUBLE PRECISION
FLOAT8 		DOUBLE
CHAR[(M)]
VARCHAR[(M)]
CHARACTER[(M)]
BPCHAR[(M)]
CHARACTER VARYING[(M)] 		STRING
TIMESTAMPTZ
TIMESTAMP WITH TIME ZONE		 ZonedTimestampType
INTERVAL [P] BIGINT
INTERVAL [P] STRING(when debezium.interval.handling.mode is set to string)
BYTEA BYTES or STRING (when debezium.binary.handling.mode is set to base64 or base64-url-safe or hex)
JSON
JSONB
XML
UUID
POINT
LTREE
CITEXT
INET
INT4RANGE
INT8RANGE
NUMRANGE
TSRANGE
DATERANGE
ENUM 		STRING

注意：由于 Debezium 版本不支持多维数组，目前只支持 PostgreSQL 的单维数组，如：'ARRAY[1,2,3]'，'int[]'。

Temporal types Mapping​

除了包含时区信息的 PostgreSQL 的 TIMESTAMPTZ 数据类型之外，其他时间类型如何映射取决于连接器配置属性 debezium.time.precision.mode 的值。以下各节将描述这些映射关系：

  • debezium.time.precision.mode=adaptive
  • debezium.time.precision.mode=adaptive_time_microseconds
  • debezium.time.precision.mode=connect

注意：受限当前 CDC 对时间类型 Time 的精度为 3，debezium.time.precision.mode 为 adaptive 或 adaptive_time_microseconds 或 connect Time 类型都转化为 Time(3) 类型。

debezium.time.precision.mode=adaptive

当 debezium.time.precision.mode 属性设置为默认的 adaptive（自适应）时，TIME 的精度为 3，TIMESTAMP 的精度为 6。

PostgreSQL type CDC type
DATE DATE
TIME([P]) TIME(3)
TIMESTAMP([P]) TIMESTAMP([P])

debezium.time.precision.mode=adaptive_time_microseconds

当 debezium.time.precision.mode 属性设置为默认的 adaptive_time_microseconds 时，TIME 的精度为 3，TIMESTAMP 的精度为 6。

PostgreSQL type CDC type
DATE DATE
TIME([P]) TIME(3)
TIMESTAMP([P]) TIMESTAMP([P])

debezium.time.precision.mode=connect

当 debezium.time.precision.mode 属性设置为默认的 connect 时，TIME 和 TIMESTAMP 的精度都为 3。

PostgreSQL type CDC type
DATE DATE
TIME([P]) TIME(3)
TIMESTAMP([P]) TIMESTAMP(3)

Decimal types Mapping​

PostgreSQL 连接器配置属性 debezium.decimal.handling.mode 的设置决定了连接器如何映射十进制类型。

当 debezium.decimal.handling.mode 属性设置为 precise（精确）时，连接器会对所有 DECIMAL、NUMERIC 和 MONEY 列使用 Kafka Connect 的 org.apache.kafka.connect.data.Decimal 逻辑类型。这是默认模式。

PostgreSQL type CDC type
NUMERIC[(M[,D])] DECIMAL[(M[,D])]
NUMERIC DECIMAL(38,0)
DECIMAL[(M[,D])] DECIMAL[(M[,D])]
DECIMAL DECIMAL(38,0)
MONEY[(M[,D])] DECIMAL(38,digits)（schema 参数 scale 包含一个整数，表示小数点移动了多少位。scale schema 参数由 money.fraction.digits 连接器配置属性决定。）

当 debezium.decimal.handling.mode 属性设置为 double 时，连接器将所有 DECIMAL、NUMERIC 和 MONEY 值表示为 Java 的 double 值，并按照下表所示进行编码。

PostgreSQL type CDC type
NUMERIC[(M[,D])] DOUBLE
DECIMAL[(M[,D])] DOUBLE
MONEY[(M[,D])] DOUBLE

debezium.decimal.handling.mode 配置属性的最后一个可选设置是 string（字符串）。在这种情况下，连接器将 DECIMAL、NUMERIC 和 MONEY 值表示为其格式化的字符串形式，并按照下表所示进行编码。

PostgreSQL type CDC type
NUMERIC[(M[,D])] STRING
DECIMAL[(M[,D])] STRING
MONEY[(M[,D])] STRING

当 debezium.decimal.handling.mode 的设置为 string 或 double 时，PostgreSQL 支持将 NaN（非数字）作为一个特殊值存储在 DECIMAL/NUMERIC 值中。在这种情况下，连接器会将 NaN 编码为 Double.NaN 或字符串常量 NAN。

HSTORE type Mapping​

PostgreSQL 连接器配置属性 debezium.hstore.handling.mode 的设置决定了连接器如何映射 HSTORE 值。

当 debezium.hstore.handling.mode 属性设置为 json（默认值）时，连接器将 HSTORE 值表示为 JSON 值的字符串形式，并按照下表所示进行编码。当 debezium.hstore.handling.mode 属性设置为 map 时，连接器对 HSTORE 值使用 MAP 模式类型。

PostgreSQL type CDC type
HSTORE STRING(`debezium.hstore.handling.mode`=`string`)
HSTORE MAP(`debezium.hstore.handling.mode`=`map`)

Network address types Mapping​

PostgreSQL 拥有可以存储 IPv4、IPv6 和 MAC 地址的数据类型。使用这些类型来存储网络地址比使用纯文本类型更为合适。网络地址类型提供了输入错误检查以及专用的操作符和函数。

PostgreSQL type CDC type
INET STRING
CIDR STRING
MACADDR STRING
MACADDR8 STRING

PostGIS Types Mapping​

PostgreSQL 通过 PostGIS 扩展支持空间数据类型：

    GEOMETRY(POINT, xx): 在笛卡尔坐标系中表示一个点，其中 EPSG:xx 定义了坐标系。它适用于局部平面计算。
    GEOGRAPHY(MULTILINESTRING): 在基于球面模型的纬度和经度上存储多条线串。它适用于全球范围的空间分析。

前者适用于小范围的平面数据，而后者适用于需要考虑地球曲率的大范围数据。

Spatial data in Postgres Json String converted in Flink
GEOMETRY(POINT, xx) {"coordinates":"[[174.9479, -36.7208]]","type":"Point","srid":3187}"
GEOGRAPHY(MULTILINESTRING) {"coordinates":"[[169.1321, -44.7032],[167.8974, -44.6414]]","type":"MultiLineString","srid":4326}

Copyright © 2026 The Apache Software Foundation. Apache Flink, Flink, and the Flink logo are trademarks of The Apache Software Foundation.