PostgreSQL-CDC

概述

PostgreSQL Extract 节点允许从 PostgreSQL 数据库中读取快照数据和增量数据。 本文档描述了如何设置 PostgreSQL Extract 节点以对 PostgreSQL 数据库运行 SQL 查询。

支持版本

依赖

为了设置 PostgreSQL Extract 节点, 下面提供了使用构建自动化工具（例如 Maven 或 SBT）和带有 Sort Connector JAR 包的 SQL 客户端的两个项目的依赖关系信息。

Maven 依赖

<dependency>
    <groupId>org.apache.inlong</groupId>
    <artifactId>sort-connector-postgres-cdc</artifactId>
    <version>1.2.0-incubating</version>
</dependency>

设置 PostgreSQL 服务

更改数据捕获 (CDC) 允许您跟踪 PostgreSQL 数据库中的更改并将其传播到基于其预写日志 (WAL) 的下游消费者。 您需要确保将上游数据库配置为支持逻辑复制。 在使用 PostgreSQL 连接器监控 PostgreSQL 服务器上提交的更改之前， 决定您打算使用哪个逻辑解码插件。 如果您不打算使用本机 pgoutput 逻辑复制流支持，则必须安装逻辑解码 插件到 PostgreSQL 服务器。

pgoutput

pgoutput 是 PostgreSQL 10+ 中的标准逻辑解码输出插件。 它由 PostgreSQL 社区维护，并由 PostgreSQL 本身用于逻辑复制。 此插件始终存在，因此无需安装其他库。

1. 检查wal_level配置设置：

SHOW wal_level;

默认值为副本。 对于 CDC，您需要在数据库配置文件 (postgresql.conf) 中将其设置为logical。 请记住，更改 wal_level 需要重新启动 Postgres 实例，并且可能会影响数据库性能。

1. 请在 postgresql.conf 文件中指定以下内容：

wal_level = logical

decoderbufs

decoderbufs 基于 Protobuf 并由 Debezium 社区维护， 安装 它。

1. 要在启动时加载插件，请将以下内容添加到 postgresql.conf 文件中：

shared_preload_libraries = 'decoderbufs'

1. 请在 postgresql.conf 文件中指定以下内容：

wal_level = logical

如何创建 PostgreSQL Extract 节点

SQL API 用法

CREATE TABLE `postgresTable`(
  `name` STRING,
  `age` INT
) WITH (
  'connector' = 'postgres-cdc',
  'hostname' = 'localhost',
  'username' = 'postgres',
  'password' = 'inlong',
  'database-name' = 'postgres',
  'schema-name' = 'public',
  'port' = '5432',
  'table-name' = 'user',
  'decoding.plugin.name' = 'pgoutput',
  'slot.name' = 'feaafacbaddadc'
)

InLong Dashboard 用法

TODO: 将在未来支持此功能。

Usage for InLong Manager Client

TODO: 将在未来支持此功能。

PostgreSQL Extract 节点参数

Note: slot.name 建议为不同的表设置以避免潜在的 PSQLException: ERROR: replication slot “flink” is active for PID 974 error。
Note: PSQLException: ERROR: all replication slots are in use Hint: Free one or increase max_replication_slots. 我们可以通过以下语句删除槽。

SELECT*FROM pg_replication_slots;
-- 获取插槽名称为 flink。 删除它
SELECT pg_drop_replication_slot('flink');

可用的元数据

以下格式元数据可以作为表定义中的只读 (VIRTUAL) 列公开。

扩展的 CREATE TABLE 示例演示了使用这些元数据字段的语法：

CREATE TABLE postgresTable (
    db_name STRING METADATA FROM 'database_name' VIRTUAL,
    table_name STRING METADATA  FROM 'table_name' VIRTUAL,
    operation_ts TIMESTAMP_LTZ(3) METADATA FROM 'op_ts' VIRTUAL,
    `name` STRING,
    `age` INT
) WITH (
     'connector' = 'postgres-cdc',
     'hostname' = 'localhost',
     'username' = 'postgres',
     'password' = 'inlong',
     'database-name' = 'postgres',
     'schema-name' = 'public',
     'port' = '5432',
     'table-name' = 'user',
     'decoding.plugin.name' = 'pgoutput',
     'slot.name' = 'feaafacbaddadc'
);

数据类型映射