流计算 Oceanus Postgres 连接器

Postgres 连接器支持从 PostgreSQL 数据库读取快照数据与增量 WAL 数据，并提供整库同步能力。通过 YAML 作业定义即可完成一条数据同步链路。
﻿
前置准备
使用 Postgres Source 前，需要确保 PostgreSQL 实例满足下列条件：
﻿
逻辑解码已开启，并已安装逻辑解码插件。
数据库账号具备连接数据库、创建逻辑复制槽、读取快照所需的权限，通常使用具备REPLICATION 属性的超级用户或具备相应权限的角色。
已为作业创建专用的逻辑复制槽，并通过 slot.name 指定。
每个作业应使用独立的复制槽，多个作业共用同一个槽会导致数据读取异常。
tables 中的所有表必须来自同一个 database，跨库同步需要拆分为多个作业。
﻿
版本说明
Flink 版本
说明
1.20
基于 Flink-cdc 社区 release-3.6 开发，JDK 版本为 11，支持 PostgreSQL 数据库
使用范围
可作为 Source（type: postgres）使用，不支持作为 Sink。
﻿
暂不支持同步表结构变更。PostgreSQL 的 WAL 日志中不包含可解析的表结构变更记录，因此上游发生 DDL 变更时不会传递到下游。
﻿
快速示例
Postgres 同步到 Doris
source:
  type: postgres
  name: Postgres Source
  hostname: 127.0.0.1
  port: 5432
  username: postgres
  password: pass
  tables: adb.\\.*.\\.*
  decoding.plugin.name: pgoutput
  slot.name: pgtest
﻿
sink:
  type: doris
  name: Doris Sink
  fenodes: 127.0.0.1:8030
  username: root
  password: pass
﻿
pipeline:
  name: Postgres to Doris Pipeline
  parallelism: 4
﻿
Source 配置项
参数
是否必填
默认值
类型
说明
hostname
required
(none)
String
PostgreSQL 数据库服务器的 IP 地址或主机名。
port
optional
5432
Integer
PostgreSQL 数据库服务器的端口号。
username
required
(none)
String
连接 PostgreSQL 数据库服务器时使用的用户名。
password
required
(none)
String
连接 PostgreSQL 数据库服务器时使用的密码。
tables
required
(none)
String
需要捕获的表名，格式为 库名.模式名.表名，支持正则表达式，多条规则用英文逗号分隔。
注意：点号（.）是库名、模式名与表名的分隔符，如果需要在正则表达式中用点号匹配任意字符， 必须用反斜杠转义。
例如：db0.\\.*.\\.*, db1.public.user_table_[0-9]+
所有表名必须使用同一个 database，否则作业启动校验失败。
若按该规则匹配不到任何表，作业启动时会直接报错。
slot.name
required
(none)
String
为流式读取变更而在服务端创建的逻辑复制槽名称。每个作业应使用独立的复制槽。
decoding.plugin.name
optional
pgoutput
String
服务端已安装的逻辑解码插件名称，可取 pgoutput 或 decoderbufs。
tables.exclude
optional
(none)
String
需要排除的表名，在 tables 匹配结果之上做二次排除，用法与 tables 相同。
排除后若无剩余表，作业启动校验失败。
server-time-zone
optional
(none)
String
数据库服务器的会话时区，例如 Asia/Shanghai。未配置时使用作业运行环境的系统时区， 对时间相关字段可能造成数据不一致，建议显式配置。
table-id.include-database
optional
false
Boolean
生成的 Table ID 是否包含 database 名。
为 true 时格式为 (database, schema, table)；为 false 时为 (schema, table)。
scan.startup.mode
optional
initial
String
启动模式，可取 initial、latest-offset、 snapshot、committed-offset，详见「启动模式」章节。
取值大小写不敏感。
scan.incremental.snapshot.chunk.size
optional
8096
Integer
快照阶段单个 chunk 的行数。快照读取时表会被切分为多个 chunk。
scan.snapshot.fetch.size
optional
1024
Integer
读取表快照时每次拉取的最大行数。
scan.incremental.snapshot.backfill.skip
optional
true
Boolean
快照阶段是否跳过 backfill。跳过之后，快照期间发生的变更会在增量阶段回放， 而不是合并进快照结果。
注意：跳过 backfill 可能带来数据不一致，只能保证 at-least-once 语义。
scan.lsn-commit.checkpoints-num-delay
optional
3
Integer
开始提交 LSN 位点前延迟的 checkpoint 次数。
增大该值后，全局复制槽消费到的位点会在多个 checkpoint 延迟之后才提交， 有利于在流式阶段持续回收日志文件。
schema-change.enabled
optional
false
Boolean
是否在处理 pgoutput 的 Relation 消息时推断 CDC 列类型。
metadata.list
optional
(none)
String
需要额外透传到下游的元数据列表，英文逗号分隔， 可取 op_ts、table_name、database_name、schema_name， 详见「支持的元数据列」章节。
connect.timeout
optional
30s
Duration
连接 PostgreSQL 服务器的超时时间。
connect.max-retries
optional
3
Integer
建立 PostgreSQL 连接的最大重试次数。
connection.pool.size
optional
20
Integer
连接池大小。
heartbeat.interval
optional
30s
Duration
发送心跳事件的间隔，用于跟踪最新可用的 WAL 位点。
chunk-meta.group.size
optional
1000
Integer
chunk 元数据的分组大小，超过该大小时元数据会被拆成多个分组。
chunk-key.even-distribution.factor.upper-bound
optional
1000.0
Double
chunk 切分键分布因子的上界。分布因子按 (MAX(id) - MIN(id) + 1) / rowCount 计算， 用于判断表数据是否均匀分布。取值必须大于等于 1.0。
chunk-key.even-distribution.factor.lower-bound
optional
0.05
Double
chunk 切分键分布因子的下界，取值必须落在 [0, 1] 区间内。
scan.incremental.close-idle-reader.enabled
optional
false
Boolean
快照结束后是否关闭空闲的 Reader。依赖 FLIP-147（Checkpoints After Tasks Finished）， Flink 版本需大于等于 1.14。
scan.incremental.snapshot.unbounded-chunk-first.enabled
optional
false
Boolean
快照阶段是否优先分配 unbounded chunk，有助于降低 TaskManager 在快照阶段发生 OOM 的风险。
jdbc.properties.*
optional
(none)
String
透传给 JDBC URL 的自定义属性，例如 jdbc.properties.ssl: 'true'。
debezium.*
optional
(none)
String
透传给 Debezium Embedded Engine 的属性。
启动模式
scan.startup.mode 决定作业从哪里开始读取数据，取值如下（大小写不敏感）：
﻿
initial（默认）：首次启动时先对捕获的表做全量快照，快照完成后继续读取最新 WAL。
latest-offset：跳过快照，只从 WAL 末尾开始读取，即只消费作业启动之后产生的变更。
snapshot：只做快照阶段，不进入增量阶段，快照读取完成后作业退出。
committed-offset：跳过快照，从复制槽中已确认提交的位点开始读取。
﻿
支持的元数据列
通过 metadata.list 可以指定透传到下游的元数据，可在 transform 中直接引用。
Key
类型
说明
op_ts
TIMESTAMP_LTZ(3) NOT NULL
该行数据在数据库中发生变更的时间。
如果该记录读取自表的快照而非变更流，则该值始终为 0。
table_name
STRING NOT NULL
包含该行数据的表名。
schema_name
STRING NOT NULL
包含该行数据的 schema 名。
database_name
STRING NOT NULL
包含该行数据的 database 名。
配置示例：
source:
  type: postgres
  metadata.list: op_ts,table_name,database_name,schema_name
﻿
数据类型映射
Postgres 作为 Source 时，按下表把读取到的 PostgreSQL 字段类型转换为 CDC 类型。
PostgreSQL 类型
CDC 类型
说明
BOOLEAN
BOOLEAN
-
BIT(1)
VARBIT(1)
BOOLEAN
-
BIT(n)，n > 1
VARBIT(n)，n > 1
BINARY(n)
-
SMALLINT
INT2
SMALLSERIAL
SERIAL2
SMALLINT
-
INTEGER
SERIAL
INT
-
BIGINT
BIGSERIAL
OID
BIGINT
-
REAL
FLOAT4
FLOAT
-
DOUBLE PRECISION
FLOAT8
DOUBLE
-
NUMERIC(p, s)
DECIMAL(p, s)
DECIMAL(p, s) / DOUBLE / STRING
取决于 Debezium 的 decimal 处理模式：
precise 模式下，精度在 (18, 38] 区间内映射为 DECIMAL(p, s)， 否则（含未显式声明精度的情况）映射为 DECIMAL(38, 18)；
double 模式映射为 DOUBLE；string 模式映射为 STRING。
MONEY
DECIMAL(38, s)
s 由 Debezium 的 money 小数位数配置决定。
CHAR(n)
CHARACTER(n)
BPCHAR(n)
CHAR(n)
-
VARCHAR(n)
CHARACTER VARYING(n)
VARCHAR(n)
-
TEXT
POINT
UUID
JSON
JSONB
XML
INET
CIDR
MACADDR
MACADDR8
INT4RANGE
NUMRANGE
INT8RANGE
TSRANGE
TSTZRANGE
DATERANGE
LTREE
GEOMETRY
GEOGRAPHY
CITEXT
枚举类型
STRING
PostGIS 的 GEOMETRY、GEOGRAPHY 类型以文本形式读出。
HSTORE
STRING 或 MAP(STRING, STRING)
取决于 Debezium 的 hstore 处理模式：json 模式映射为 STRING， map 模式映射为 MAP(STRING, STRING)。
BYTEA
BYTES 或 STRING
取决于 Debezium 的 binary 处理模式：bytes 模式映射为 BYTES， base64 / hex 模式映射为 STRING。
TIMESTAMP [(p)]
TIMESTAMP WITHOUT TIME ZONE [(p)]
TIMESTAMP [(p)]
-
TIMESTAMPTZ
TIMESTAMP WITH TIME ZONE [(p)]
TIMESTAMP_LTZ [(p)]
由 Debezium 归一化到 UTC 后按带本地时区的时间戳表示。
TIME [(p)]
TIME WITHOUT TIME ZONE [(p)]
TIME [(p)]
-
DATE
DATE
-
INTERVAL
BIGINT 或 STRING
取决于 Debezium 的 interval 处理模式：numeric 模式映射为 BIGINT（微秒数）， string 模式映射为 STRING。
数组类型（如 INT4[]、TEXT[]）
ARRAY(<元素类型>)
元素类型按上表规则转换，例如 INT4[] 映射为 ARRAY(INT)， TEXT[] 映射为 ARRAY(STRING)。
未在上表中出现的字段类型暂不支持，同步包含此类字段的表会导致作业失败。
﻿

Flink 版本	说明
1.20	基于 Flink-cdc 社区 release-3.6 开发，JDK 版本为 11，支持 PostgreSQL 数据库

参数	是否必填	默认值	类型	说明
hostname	required	(none)	String	PostgreSQL 数据库服务器的 IP 地址或主机名。
port	optional	5432	Integer	PostgreSQL 数据库服务器的端口号。
username	required	(none)	String	连接 PostgreSQL 数据库服务器时使用的用户名。
password	required	(none)	String	连接 PostgreSQL 数据库服务器时使用的密码。
tables	required	(none)	String	需要捕获的表名，格式为库名.模式名.表名，支持正则表达式，多条规则用英文逗号分隔。注意：点号（.）是库名、模式名与表名的分隔符，如果需要在正则表达式中用点号匹配任意字符，必须用反斜杠转义。例如：db0.\\..\\., db1.public.user_table_[0-9]+ 所有表名必须使用同一个 database，否则作业启动校验失败。若按该规则匹配不到任何表，作业启动时会直接报错。
slot.name	required	(none)	String	为流式读取变更而在服务端创建的逻辑复制槽名称。每个作业应使用独立的复制槽。
decoding.plugin.name	optional	pgoutput	String	服务端已安装的逻辑解码插件名称，可取 pgoutput 或 decoderbufs。
tables.exclude	optional	(none)	String	需要排除的表名，在 tables 匹配结果之上做二次排除，用法与 tables 相同。排除后若无剩余表，作业启动校验失败。
server-time-zone	optional	(none)	String	数据库服务器的会话时区，例如 Asia/Shanghai。未配置时使用作业运行环境的系统时区，对时间相关字段可能造成数据不一致，建议显式配置。
table-id.include-database	optional	false	Boolean	生成的 Table ID 是否包含 database 名。为 true 时格式为 (database, schema, table)；为 false 时为 (schema, table)。
scan.startup.mode	optional	initial	String	启动模式，可取 initial、latest-offset、 snapshot、committed-offset，详见「启动模式」章节。取值大小写不敏感。
scan.incremental.snapshot.chunk.size	optional	8096	Integer	快照阶段单个 chunk 的行数。快照读取时表会被切分为多个 chunk。
scan.snapshot.fetch.size	optional	1024	Integer	读取表快照时每次拉取的最大行数。
scan.incremental.snapshot.backfill.skip	optional	true	Boolean	快照阶段是否跳过 backfill。跳过之后，快照期间发生的变更会在增量阶段回放，而不是合并进快照结果。注意：跳过 backfill 可能带来数据不一致，只能保证 at-least-once 语义。
scan.lsn-commit.checkpoints-num-delay	optional	3	Integer	开始提交 LSN 位点前延迟的 checkpoint 次数。增大该值后，全局复制槽消费到的位点会在多个 checkpoint 延迟之后才提交，有利于在流式阶段持续回收日志文件。
schema-change.enabled	optional	false	Boolean	是否在处理 pgoutput 的 Relation 消息时推断 CDC 列类型。
metadata.list	optional	(none)	String	需要额外透传到下游的元数据列表，英文逗号分隔，可取 op_ts、table_name、database_name、schema_name，详见「支持的元数据列」章节。
connect.timeout	optional	30s	Duration	连接 PostgreSQL 服务器的超时时间。
connect.max-retries	optional	3	Integer	建立 PostgreSQL 连接的最大重试次数。
connection.pool.size	optional	20	Integer	连接池大小。
heartbeat.interval	optional	30s	Duration	发送心跳事件的间隔，用于跟踪最新可用的 WAL 位点。
chunk-meta.group.size	optional	1000	Integer	chunk 元数据的分组大小，超过该大小时元数据会被拆成多个分组。
chunk-key.even-distribution.factor.upper-bound	optional	1000.0	Double	chunk 切分键分布因子的上界。分布因子按 (MAX(id) - MIN(id) + 1) / rowCount 计算，用于判断表数据是否均匀分布。取值必须大于等于 1.0。
chunk-key.even-distribution.factor.lower-bound	optional	0.05	Double	chunk 切分键分布因子的下界，取值必须落在 [0, 1] 区间内。
scan.incremental.close-idle-reader.enabled	optional	false	Boolean	快照结束后是否关闭空闲的 Reader。依赖 FLIP-147（Checkpoints After Tasks Finished）， Flink 版本需大于等于 1.14。
scan.incremental.snapshot.unbounded-chunk-first.enabled	optional	false	Boolean	快照阶段是否优先分配 unbounded chunk，有助于降低 TaskManager 在快照阶段发生 OOM 的风险。
jdbc.properties.*	optional	(none)	String	透传给 JDBC URL 的自定义属性，例如 jdbc.properties.ssl: 'true'。
debezium.*	optional	(none)	String	透传给 Debezium Embedded Engine 的属性。

Key	类型	说明
op_ts	TIMESTAMP_LTZ(3) NOT NULL	该行数据在数据库中发生变更的时间。如果该记录读取自表的快照而非变更流，则该值始终为 0。
table_name	STRING NOT NULL	包含该行数据的表名。
schema_name	STRING NOT NULL	包含该行数据的 schema 名。
database_name	STRING NOT NULL	包含该行数据的 database 名。

PostgreSQL 类型	CDC 类型	说明
BOOLEAN	BOOLEAN	-
BIT(1) VARBIT(1)	BOOLEAN	-
BIT(n)，n > 1 VARBIT(n)，n > 1	BINARY(n)	-
SMALLINT INT2 SMALLSERIAL SERIAL2	SMALLINT	-
INTEGER SERIAL	INT	-
BIGINT BIGSERIAL OID	BIGINT	-
REAL FLOAT4	FLOAT	-
DOUBLE PRECISION FLOAT8	DOUBLE	-
NUMERIC(p, s) DECIMAL(p, s)	DECIMAL(p, s) / DOUBLE / STRING	取决于 Debezium 的 decimal 处理模式： precise 模式下，精度在 (18, 38] 区间内映射为 DECIMAL(p, s)，否则（含未显式声明精度的情况）映射为 DECIMAL(38, 18)； double 模式映射为 DOUBLE；string 模式映射为 STRING。
MONEY	DECIMAL(38, s)	s 由 Debezium 的 money 小数位数配置决定。
CHAR(n) CHARACTER(n) BPCHAR(n)	CHAR(n)	-
VARCHAR(n) CHARACTER VARYING(n)	VARCHAR(n)	-
TEXT POINT UUID JSON JSONB XML INET CIDR MACADDR MACADDR8 INT4RANGE NUMRANGE INT8RANGE TSRANGE TSTZRANGE DATERANGE LTREE GEOMETRY GEOGRAPHY CITEXT 枚举类型	STRING	PostGIS 的 GEOMETRY、GEOGRAPHY 类型以文本形式读出。
HSTORE	STRING 或 MAP(STRING, STRING)	取决于 Debezium 的 hstore 处理模式：json 模式映射为 STRING， map 模式映射为 MAP(STRING, STRING)。
BYTEA	BYTES 或 STRING	取决于 Debezium 的 binary 处理模式：bytes 模式映射为 BYTES， base64 / hex 模式映射为 STRING。
TIMESTAMP [(p)] TIMESTAMP WITHOUT TIME ZONE [(p)]	TIMESTAMP [(p)]	-
TIMESTAMPTZ TIMESTAMP WITH TIME ZONE [(p)]	TIMESTAMP_LTZ [(p)]	由 Debezium 归一化到 UTC 后按带本地时区的时间戳表示。
TIME [(p)] TIME WITHOUT TIME ZONE [(p)]	TIME [(p)]	-
DATE	DATE	-
INTERVAL	BIGINT 或 STRING	取决于 Debezium 的 interval 处理模式：numeric 模式映射为 BIGINT（微秒数）， string 模式映射为 STRING。
数组类型（如 INT4[]、TEXT[]）	ARRAY(<元素类型>)	元素类型按上表规则转换，例如 INT4[] 映射为 ARRAY(INT)， TEXT[] 映射为 ARRAY(STRING)。

Postgres 连接器

本页目录：

前置准备

版本说明

使用范围

快速示例

Postgres 同步到 Doris

Source 配置项

启动模式

支持的元数据列

数据类型映射