流计算 Oceanus Postgres 连接器

一、前置准备

使用前，需要在 PostgreSQL 侧完成以下 6 步。示例中的 <db>、<schema>、<table>、<user>、<password>、<slot_name> 请替换为实际值。

二、1. 开启逻辑解码

连接器通过逻辑解码从 WAL 中读取行级变更（INSERT / UPDATE / DELETE），需要将 wal_level 设为 logical。

查看当前配置：

SHOW wal_level; — 必须为 logicalSHOW max_replication_slots; — 复制槽总数上限SHOW max_wal_senders; — WAL 发送进程数上限

自建 PostgreSQL：

ALTER SYSTEM SET wal_level = logical;ALTER SYSTEM SET max_replication_slots = 10;ALTER SYSTEM SET max_wal_senders = 10;

修改 wal_level 后必须重启实例才能生效（该操作可能导致服务中断）

云数据库 PostgreSQL：在控制台的实例参数设置中，将 wal_level 改为 logical，然后重启实例（该操作可能导致服务中断）。

max_replication_slots 应不小于本实例上的 CDC 作业数与其他逻辑订阅数之和，并留出余量；max_wal_senders 应不小于 max_replication_slots。

三、2. 确认逻辑解码插件

逻辑解码插件将 WAL 转换为连接器可读取的格式，通过 decoding.plugin.name 指定，支持以下两种：

使用默认的 pgoutput 时无需安装。可通过创建测试复制槽确认插件可用，确认后立即删除：

SELECT pg_create_logical_replication_slot('oceanus_check_slot', 'pgoutput');SELECT pg_drop_replication_slot('oceanus_check_slot');

四、3. 创建同步账号并授权

建议创建专用的同步账号，无需超级用户权限。以下语句由高权限账号（自建库为 superuser，云数据库为控制台创建的高权限账号）执行：

— 创建账号。REPLICATION 用于建立复制连接、创建和读取复制槽CREATE ROLE <user> WITH LOGIN REPLICATION PASSWORD '<password>';– 连接目标库GRANT CONNECT ON DATABASE <db> TO <user>;– 访问 schemaGRANT USAGE ON SCHEMA <schema> TO <user>;– 全量快照读取：授权整个 schema，或只授权要同步的表GRANT SELECT ON ALL TABLES IN SCHEMA <schema> TO <user>;– 或：GRANT SELECT ON <schema>.<table1>, <schema>.<table2> TO <user>;– 可选：之后新建的表自动授予 SELECT（<table_owner> 为建表账号）ALTER DEFAULT PRIVILEGES FOR ROLE <table_owner> IN SCHEMA <schema> GRANT SELECT ON TABLES TO <user>;

验证：

SELECT rolname, rolcanlogin, rolreplication FROM pg_roles WHERE rolname = '<user>';– rolcanlogin 和 rolreplication 都应为 t

五、4. 创建 publication（使用 pgoutput 时必须）

pgoutput 只输出 publication 中所含表的变更。连接器默认使用名为 dbz_publication 的 publication，不存在时会尝试自动创建包含全库所有表的 publication，该操作需要超级用户权限。建议由高权限账号预先创建。

— 推荐：只发布需要同步的表CREATE PUBLICATION dbz_publication FOR TABLE <schema>.<table1>, <schema>.<table2>;– 之后新增同步表ALTER PUBLICATION dbz_publication ADD TABLE <schema>.<new_table>;

FOR TABLE 方式需要执行账号对目标库有 CREATE 权限，且为表的 owner；FOR ALL TABLES 方式需要超级用户（云数据库为 tencentdb_superuser）。

不建议使用 FOR ALL TABLES。库中无主键且未设置 REPLICA IDENTITY 的表，执行 UPDATE / DELETE 时会报错（cannot update table … because it does not have a replica identity and publishes updates），影响源库业务。

publication 需覆盖 tables 匹配到的所有表。未包含的表只能读到全量数据，读不到增量数据。

验证：

SELECT * FROM pg_publication_tables WHERE pubname = 'dbz_publication';

如需让不同作业使用不同的 publication，可配置 debezium.publication.name。

六、5. 设置表的 REPLICA IDENTITY（按需）

REPLICA IDENTITY 决定 UPDATE / DELETE 时 WAL 中记录的旧行内容。

— 需要时设为 FULL。会增加 WAL 体积，大表或高频更新的表需评估ALTER TABLE <schema>.<table> REPLICA IDENTITY FULL;

验证：

SELECT relname, relreplident — d=default, f=full, n=nothing, i=indexFROM pg_classWHERE relnamespace = '<schema>'::regnamespace AND relkind IN ('r', 'p');

七、6. 确认网络连通

·云数据库 PostgreSQL：在安全组中放通 Oceanus 集群网段访问 5432 端口。

·自建 PostgreSQL：在 pg_hba.conf 中允许 Oceanus 网段连接目标数据库，然后执行 SELECT pg_reload_conf(); 重新加载。

八、复制槽（slot.name）

·无需手动创建。作业首次启动时自动创建，账号需具备 REPLICATION 属性。

·名称只能包含小写字母、数字和下划线，不超过 63 个字符。

·每个作业必须使用独立的复制槽。多个作业共用一个复制槽会导致数据丢失或读取异常。

九、版本说明

十、使用范围

可作为 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: pgtestsink: type: doris name: Doris Sink fenodes: 127.0.0.1:8030 username: root password: passpipeline: name: Postgres to Doris Pipeline parallelism: 4

十二、Source 配置项

十三、启动模式

scan.startup.mode 决定作业从哪里开始读取数据，取值如下（大小写不敏感）：

·initial（默认）：首次启动时先对捕获的表做全量快照，快照完成后继续读取最新 WAL。

·latest-offset：跳过快照，只从 WAL 末尾开始读取，即只消费作业启动之后产生的变更。

·snapshot：只做快照阶段，不进入增量阶段，快照读取完成后作业退出。

·committed-offset：跳过快照，从复制槽中已确认提交的位点开始读取。

十四、支持的元数据列

通过 metadata.list 可以指定透传到下游的元数据，可在 transform 中直接引用。

配置示例：

source: type: postgres metadata.list: op_ts,table_name,database_name,schema_name

十五、数据类型映射

Postgres 作为 Source 时，按下表把读取到的 PostgreSQL 字段类型转换为 CDC 类型。

未在上表中出现的字段类型暂不支持，同步包含此类字段的表会导致作业失败。