| import ChangeLog from '../changelog/connector-cdc-postgres.md'; |
| |
| # PostgreSQL CDC |
| |
| > PostgreSQL CDC 源连接器 |
| |
| ## 支持的引擎 |
| |
| > SeaTunnel Zeta<br/> |
| > Flink <br/> |
| |
| ## 主要特性 |
| |
| - [ ] [批处理](../../introduction/concepts/connector-v2-features.md) |
| - [x] [流处理](../../introduction/concepts/connector-v2-features.md) |
| - [x] [精确一次](../../introduction/concepts/connector-v2-features.md) |
| - [ ] [列投影](../../introduction/concepts/connector-v2-features.md) |
| - [x] [并行性](../../introduction/concepts/connector-v2-features.md) |
| - [x] [支持用户定义的拆分](../../introduction/concepts/connector-v2-features.md) |
| |
| ## 描述 |
| |
| Postgre CDC 连接器允许从 Postgre 数据库读取快照数据和增量数据。本文件描述了如何设置 Postgre CDC 连接器，以便对 Postgre 数据库执行 SQL 查询。 |
| |
| ## 支持的数据源信息 |
| |
| | 数据源 | 支持的版本 | 驱动 | Url | Maven | |
| |------------|-----------------------------------------------------|---------------------|---------------------------------------|--------------------------------------------------------------------------| |
| | PostgreSQL | 不同的依赖版本有不同的驱动类。 | org.postgresql.Driver | jdbc:postgresql://localhost:5432/test | [下载](https://mvnrepository.com/artifact/org.postgresql/postgresql) | |
| | PostgreSQL | 如果您想在 PostgreSQL 中操作 GEOMETRY/GEOGRAPHY 类型。 | org.postgresql.Driver | jdbc:postgresql://localhost:5432/test | [下载](https://mvnrepository.com/artifact/net.postgis/postgis-jdbc) | |
| |
| ## 使用依赖 |
| |
| ### 安装 Jdbc 驱动 |
| |
| #### 对于 Spark/Flink 引擎 |
| |
| > 1. 您需要确保 [jdbc 驱动 jar 包](https://mvnrepository.com/artifact/org.postgresql/postgresql) 已放置在目录 `${SEATUNNEL_HOME}/plugins/` 中。 |
| |
| #### 对于 SeaTunnel Zeta 引擎 |
| |
| > 1. 您需要确保 [jdbc 驱动 jar 包](https://mvnrepository.com/artifact/org.postgresql/postgresql) 已放置在目录 `${SEATUNNEL_HOME}/lib/` 中。 |
| |
| 请下载并将 PostgreSQL 驱动放入 `${SEATUNNEL_HOME}/lib/` 目录。例如：cp postgresql-xxx.jar `$SEATUNNEL_HOME/lib/` |
| |
| > 以下是启用 PostgreSQL 中的 CDC（变化数据捕获）的步骤： |
| |
| 1. 确保 wal_level 设置为 logical：通过在 postgresql.conf 配置文件中添加 "wal_level = logical" 来修改，重启 PostgreSQL 服务器以使更改生效。 |
| 或者，您可以使用 SQL 命令直接修改配置： |
| |
| ```sql |
| ALTER SYSTEM SET wal_level TO 'logical'; |
| SELECT pg_reload_conf(); |
| ``` |
| |
| 2. 将指定表的 REPLICA 策略更改为 FULL，除非 `require-replica-identity-full` 设置为 `false`。 |
| |
| ```sql |
| ALTER TABLE your_table_name REPLICA IDENTITY FULL; |
| ``` |
| |
| ## 数据类型映射 |
| |
| | PostgreSQL 数据类型 | SeaTunnel 数据类型 | |
| |-----------------------------------------------------------------------------------------|------------------------------------------------------------------------------------------------------------------------------------------------| |
| | BOOL<br/> | BOOLEAN | |
| | _BOOL<br/> | ARRAY<BOOLEAN> | |
| | BYTEA<br/> | BYTES | |
| | _BYTEA<br/> | ARRAY<TINYINT> | |
| | INT2<br/>SMALLSERIAL<br/>INT4<br/>SERIAL<br/> | INT | |
| | _INT2<br/>_INT4<br/> | ARRAY<INT> | |
| | INT8<br/>BIGSERIAL<br/> | BIGINT | |
| | _INT8<br/> | ARRAY<BIGINT> | |
| | FLOAT4<br/> | FLOAT | |
| | _FLOAT4<br/> | ARRAY<FLOAT> | |
| | FLOAT8<br/> | DOUBLE | |
| | _FLOAT8<br/> | ARRAY<DOUBLE> | |
| | NUMERIC(指定列的列大小>0) | DECIMAL(指定列的列大小, 获取指定列小数点右侧的位数) | |
| | NUMERIC(指定列的列大小<0) | DECIMAL(38, 18) | |
| | BPCHAR<br/>CHARACTER<br/>VARCHAR<br/>TEXT<br/>GEOMETRY<br/>GEOGRAPHY<br/>JSON<br/>JSONB | STRING | |
| | _BPCHAR<br/>_CHARACTER<br/>_VARCHAR<br/>_TEXT | ARRAY<STRING> | |
| | TIMESTAMP<br/> | TIMESTAMP | |
| | TIME<br/> | TIME | |
| | DATE<br/> | DATE | |
| | 其他数据类型 | 尚不支持 | |
| |
| ## 源选项 |
| |
| | 名称 | 类型 | 必需 | 默认 | 描述 | |
| |-------------------------------------------|----------|------|----------|----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------| |
| | url | String | 是 | - | JDBC 连接的 URL。参考案例：`jdbc:postgresql://localhost:5432/postgres_cdc?loggerLevel=OFF`。 | |
| | username | String | 是 | - | 连接到数据库服务器时使用的数据库名称。 | |
| | password | String | 是 | - | 连接到数据库服务器时使用的密码。 | |
| | database-names | List | 否 | - | 需要监控的数据库名称。 | |
| | table-names | List | 是 | - | 需要监控的数据库表名称。表名称需要包含数据库名称，例如：`database_name.table_name`。 | |
| | table-names-config | List | 否 | - | 表配置列表。例如： [{"table": "db1.schema1.table1","primaryKeys": ["key1"],"snapshotSplitColumn": "key2"}]。 snapshotSplitColumn 选项必须配置为唯一键(主键或唯一索引)。 如果指定了非唯一列，该配置将被忽略，SeaTunnel 会在内部自动选择合适的拆分列。 | |
| | startup.mode | List | 否 | INITIAL | PostgreSQL CDC 消费者的可选启动模式，有效枚举为 `initial`、`earliest` 和 `latest`。<br/> `initial`: 启动时同步历史数据，然后同步增量数据。<br/> `earliest`: 从可能的最早偏移量启动。<br/> `latest`: 从最新偏移量启动。 | |
| | snapshot.split.size | Integer | 否 | 8096 | 表快照的拆分大小（行数），捕获的表在读取表快照时被拆分成多个拆分。 | |
| | snapshot.fetch.size | Integer | 否 | 1024 | 读取表快照时每次轮询的最大获取大小。 | |
| | slot.name | String | 否 | - | 为特定数据库/模式创建的用于流式传输更改的 PostgreSQL 逻辑解码槽的名称。服务器使用此槽将事件流式传输到您正在配置的连接器。默认值为 seatunnel。 | |
| | decoding.plugin.name | String | 否 | pgoutput | 安装在服务器上的 Postgres 逻辑解码插件的名称，支持的值有 decoderbufs、wal2json、wal2json_rds、wal2json_streaming、wal2json_rds_streaming 和 pgoutput。 | |
| | server-time-zone | String | 否 | UTC | 数据库服务器中的会话时区。如果未设置，则使用 ZoneId.systemDefault() 来确定服务器时区。 | |
| | connect.timeout.ms | Duration | 否 | 30000 | 连接器在尝试连接到数据库服务器后应等待的最大时间，以防超时。 | |
| | connect.max-retries | Integer | 否 | 3 | 连接器应重试建立数据库服务器连接的最大重试次数。 | |
| | connection.pool.size | Integer | 否 | 20 | JDBC 连接池大小。 | |
| | chunk-key.even-distribution.factor.upper-bound | Double | 否 | 100 | 块键分布因子的上限。此因子用于确定表数据是否均匀分布。如果计算出的分布因子小于或等于此上限（即 (MAX(id) - MIN(id) + 1) / 行数），则将优化表块以实现均匀分布。否则，如果分布因子更大，则将认为该表分布不均匀，并且如果估计的分片数量超过 `sample-sharding.threshold` 指定的值，则将使用基于采样的分片策略。默认值为 100.0。 | |
| | chunk-key.even-distribution.factor.lower-bound | Double | 否 | 0.05 | 块键分布因子的下限。此因子用于确定表数据是否均匀分布。如果计算出的分布因子大于或等于此下限（即 (MAX(id) - MIN(id) + 1) / 行数），则将优化表块以实现均匀分布。否则，如果分布因子更小，则将认为该表分布不均匀，并且如果估计的分片数量超过 `sample-sharding.threshold` 指定的值，则将使用基于采样的分片策略。默认值为 0.05。 | |
| | sample-sharding.threshold | Integer | 否 | 1000 | 此配置指定触发采样分片策略的估计分片数量阈值。当分布因子超出由 `chunk-key.even-distribution.factor.upper-bound` 和 `chunk-key.even-distribution.factor.lower-bound` 指定的范围，且估计的分片数量（计算为近似行数 / 块大小）超过此阈值时，将使用采样分片策略。这可以帮助更有效地处理大数据集。默认值为 1000 个分片。 | |
| | inverse-sampling.rate | Integer | 否 | 1000 | 在采样分片策略中使用的采样率的倒数。例如，如果此值设置为 1000，则意味着在采样过程中应用 1/1000 的采样率。此选项提供了控制采样粒度的灵活性，从而影响最终的分片数量。在处理非常大数据集时，较低的采样率尤为有用。默认值为 1000。 | |
| | split.allow-sampling | Boolean | 否 | true | 是否允许基于采样的分片策略。当设置为 false 时，无论预估分片数是否超过阈值，系统都将回退到非均匀分片方式（迭代查询方式）。默认值为 true。 | |
| | exactly_once | Boolean | 否 | false | 启用精确一次语义。 | |
| | format | Enum | 否 | DEFAULT | PostgreSQL CDC 的可选输出格式，有效枚举为 `DEFAULT`、`COMPATIBLE_DEBEZIUM_JSON`。 | |
| | require-replica-identity-full | Boolean | 否 | true | 要求表具有 REPLICA IDENTITY FULL。设置为 false 时，允许表使用其他副本标识设置，但 UPDATE/DELETE 事件可能不包含之前的状态。此选项仅应用于仅追加的表（例如 outbox 模式）。默认为 true 以保持向后兼容性。 | |
| | debezium | Config | 否 | - | 将 [Debezium 的属性](https://github.com/debezium/debezium/blob/v1.9.8.Final/documentation/modules/ROOT/pages/connectors/postgresql.adoc#connector-configuration-properties) 传递给用于捕获 PostgreSQL 服务器数据更改的 Debezium 嵌入式引擎。 | |
| | common-options | | 否 | - | 源插件的公共参数，请参阅 [源公共选项](../common-options/source-common-options.md) 获取详细信息。 | |
| |
| ## 任务示例 |
| |
| ### 简单 |
| |
| > 支持多表读取 |
| |
| ``` |
| |
| |
| env { |
| # You can set engine configuration here |
| execution.parallelism = 1 |
| job.mode = "STREAMING" |
| checkpoint.interval = 5000 |
| read_limit.bytes_per_second=7000000 |
| read_limit.rows_per_second=400 |
| } |
| |
| source { |
| Postgres-CDC { |
| plugin_output = "customers_Postgre_cdc" |
| username = "postgres" |
| password = "postgres" |
| database-names = ["postgres_cdc"] |
| schema-names = ["inventory"] |
| table-names = ["postgres_cdc.inventory.postgres_cdc_table_1,postgres_cdc.inventory.postgres_cdc_table_2"] |
| url = "jdbc:postgresql://postgres_cdc_e2e:5432/postgres_cdc?loggerLevel=OFF" |
| } |
| } |
| |
| transform { |
| |
| } |
| |
| sink { |
| jdbc { |
| plugin_input = "customers_Postgre_cdc" |
| url = "jdbc:postgresql://postgres_cdc_e2e:5432/postgres_cdc?loggerLevel=OFF" |
| driver = "org.postgresql.Driver" |
| username = "postgres" |
| password = "postgres" |
| |
| generate_sink_sql = true |
| # You need to configure both database and table |
| database = postgres_cdc |
| schema = "inventory" |
| tablePrefix = "sink_" |
| primary_keys = ["id"] |
| } |
| } |
| ``` |
| |
| ### 支持自定义表的主键 |
| |
| ``` |
| source { |
| Postgres-CDC { |
| plugin_output = "customers_mysql_cdc" |
| username = "postgres" |
| password = "postgres" |
| database-names = ["postgres_cdc"] |
| schema-names = ["inventory"] |
| table-names = ["postgres_cdc.inventory.full_types_no_primary_key"] |
| url = "jdbc:postgresql://postgres_cdc_e2e:5432/postgres_cdc?loggerLevel=OFF" |
| decoding.plugin.name = "decoderbufs" |
| exactly_once = false |
| table-names-config = [ |
| { |
| table = "postgres_cdc.inventory.full_types_no_primary_key" |
| primaryKeys = ["id"] |
| } |
| ] |
| } |
| } |
| ``` |
| |
| ## 常见问题 |
| |
| ### PostgreSQL CDC 需要哪些权限？ |
| |
| CDC 用户需要具备 `REPLICATION` 角色以及对监控表的 `SELECT` 权限： |
| |
| ```sql |
| CREATE USER replication_user REPLICATION LOGIN PASSWORD 'password'; |
| GRANT SELECT ON ALL TABLES IN SCHEMA public TO replication_user; |
| ``` |
| |
| 同时需要在 `postgresql.conf` 中设置 `wal_level = logical`，并在 `pg_hba.conf` 中添加允许该用户复制连接的条目。 |
| |
| ### 支持哪些逻辑解码插件？ |
| |
| SeaTunnel PostgreSQL CDC 支持 `pgoutput`（PostgreSQL 10 起内置）、`wal2json` 和 `decoderbufs`，默认使用 `pgoutput`。通过 `decoding.plugin.name` 参数指定插件。 |
| |
| ### SeaTunnel 能从 PostgreSQL 备库读取 CDC 数据吗？ |
| |
| 不能。PostgreSQL 逻辑复制槽必须在主库上创建和消费，SeaTunnel 无法直接从备库读取逻辑复制槽，需将 CDC 连接器指向主库实例。 |
| |
| ### PostgreSQL CDC 是否支持无主键表？ |
| |
| 默认需要主键。如果表有可作为唯一标识的列，可以通过 `table-names-config` 中的 `primaryKeys` 字段自定义主键。 |
| |
| ### 复制槽如何管理？ |
| |
| SeaTunnel 在任务启动时会创建或复用 `slot.name` 指定的复制槽。未使用的复制槽会持续占用 |
| 磁盘上的 WAL 段，导致 WAL 持续增长。当 CDC 任务永久下线时，应在 PostgreSQL 侧手动删除 |
| 不再使用的复制槽。 |
| |
| ### PostgreSQL CDC 为什么会滞后？ |
| |
| 滞后可能由逻辑解码插件处理慢或 WAL sender 负载过高引起。可通过监控 `pg_replication_slots` 中的 `confirmed_flush_lsn` 漂移情况来排查。确保 CDC 任务持续消费事件，并保持 SeaTunnel 与 PostgreSQL 之间的网络低延迟。 |
| |
| ## 另请参阅 |
| |
| 若需要一份面向生产的端到端实践指南，涵盖全量 + 增量同步生命周期、2PC sink 配置、Schema 演进与常见故障排查，请参阅 [CDC 生产实战手册](../cdc-production-cookbook.md)。 |
| |
| ## 变更日志 |
| |
| <ChangeLog /> |
