设置 Openflow Connector for PostgreSQL¶

配置 wal_level¶

Openflow Connector for PostgreSQL 要求将 wal_level (https://www.postgresql.org/docs/current/runtime-config-wal.html#GUC-WAL-LEVEL) 设置为 logical。

根据 PostgreSQL 服务器的托管位置，您可以按如下方式配置 wal_level：

创建发布¶

在开始复制之前，Openflow Connector for PostgreSQL 要求您在 PostgreSQL 中创建并配置一个 发布 (https://www.postgresql.org/docs/current/logical-replication-publication.html#LOGICAL-REPLICATION-PUBLICATION)。您可以为所有表或表的子集以及仅具有指定列的特定表创建发布。确保您计划复制的每张表和列都包含在发布中。您也可以稍后在连接器运行时修改发布。要创建和配置发布，请执行以下操作：

1. 以具有 CREATE 权限的用户身份登录数据库，然后运行以下查询：

   • For PostgreSQL 13 and later:

     CREATE PUBLICATION <publication name> WITH (publish_via_partition_root = true);
     

     Copy

     The additional publish_via_partition_root is needed for correct replication of partitioned tables. To learn more about ingestion of partitioned tables see Replicate a partitioned table.

   • For PostgreSQL versions earlier than 13:

     CREATE PUBLICATION <publication name>;
     

     Copy

2. 使用以下命令定义数据库代理可见的表：

ALTER PUBLICATION <publication name> ADD TABLE <table name>;

Copy

For partitioned tables, it's enough to just add the root partition table to the publication. See Replicate a partitioned table for more details.

重要

** PostgreSQL 15 及更高版本** 支持为表列的指定子集配置发布。为使连接器正确支持此功能，必须使用 列筛选设置 来包含与发布上设置的相同的列。

如果没有此设置，连接器的行为将如下所示：

• 在目标表中，未包含在筛选器中的列将以 __DELETED 作为后缀。在快照阶段复制的所有数据都将保留下来。

• After you add new columns to the publication, the table will be permanently failed, and you will need to restart its replication.

有关更多信息，请参阅 ALTER PUBLICATION (https://www.postgresql.org/docs/current/sql-alterpublication.html) 。

实时复制一组表¶

1. 右键点击导入的进程组并选择 Parameters。

2. 按照 流参数 中所述填充所需的参数值。

从 PostgreSQL 副本服务器复制表¶

连接器可以使用 逻辑复制 (https://www.postgresql.org/docs/current/logical-replication.html) 从主服务器、热备用副本 (https://www.postgresql.org/docs/current/hot-standby.html)，或订阅者服务器引入数据。在配置连接器以连接到 PostgreSQL 副本之前，确保主节点和副本节点之间的复制正常工作。在调查连接器中缺失数据的问题时，首先确保连接器使用的副本服务器中存在缺失的行。

连接到备用副本时的其他注意事项：

• 仅支持连接到热备用副本。请注意，在提升为主实例之前，热备用副本无法接受来自客户端的连接。

• 服务器的 PostgreSQL 版本必须 >= 16。

• 连接器所需的 发布 必须在主服务器上创建，而不是在备用服务器上创建。备用服务器是只读的，不允许创建发布。

如果您连接到热备用实例并看到 Trying to create the replication slot '<replication slot>' timed out.If connecting to a standby instance, ensure there is some traffic on the primary PostgreSQL instance, otherwise the call to create a replication slot will never return. 错误，或者 Read PostgreSQL CDC Stream 处理器未启动，请登录主 PostgreSQL 实例并执行以下查询：

SELECT pg_log_standby_snapshot();

当主服务器中没有数据变更时，会发生此错误。因此，在副本服务器上创建复制槽时，连接器可能会出现卡顿。这是因为副本服务器需要有关主服务器正在运行的事务的信息，才能创建复制槽。主服务器在空闲时不会发送信息。pg_log_standby_snapshot() 函数强制主服务器将有关正在运行的事务的信息发送到副本服务器。

从复制中移除并重新添加表¶

要从复制中移除表，请确保在复制参数上下文中的 包括表名 或 包括表正则表达式 参数里移除。

如果您想稍后将该表重新添加到复制中，请先在 Snowflake 中删除相应的目标表。然后，再将该表添加回 包括表名 或 包括表正则表达式 参数中。这样可以确保表的复制过程重新开始。

此方法还可用于从失败的表复制场景中恢复复制。

复制表中列的子集¶

连接器可以将每张表复制的数据筛选到已配置列的子集。

要对列应用筛选器，请在复制参数上下文中修改 Column Filter 属性，添加一个配置数组，每个条目对应一张需要筛选列的表。

可以按名称或模式包含或排除列。您可以对每张表应用一个条件，也可以合并多个条件，排除项的优先级始终高于包含项。

以下示例显示了可用的字段。schema 和 table 是必填项，且必须至少填写以下其中一个字段：included、excluded、includedPattern、excludedPattern。

[
    {
        "schema": "<source table schema>",
        "table" : "<source table name>",
        "included": ["<column name>", "<column name>"],
        "excluded": ["<column name>", "<column name>"],
        "includedPattern": "<regular expression>",
        "excludedPattern": "<regular expression>",
    }
]

Replicate a partitioned table¶

The connector supports replication of partitioned tables for PostgreSQL servers with version >= 15. A PostgreSQL partitioned table will be replicated into Snowflake as a single destination table.

For example, if you have a partitioned table orders, with sub-partitions orders_2023, orders_2024, and configured the connector to ingest all tables matching orders.* pattern, then only the orders table will be replicated to Snowflake, and it will include data from all sub-partitions.

To support replication of partitioned tables, ensure that the publication created in PostgreSQL has the publish_via_partition_root option set to true.

Ingestion of partitioned tables has currently the following limitations:

• When a table is attached as a partition to a partitioned table after ingestion was started, the connector won't fetch data that existed in the partition table before attaching.

• When a sub-partition table is detached from the partitioned table after ingestion was started, the connector won't mark the data from this sub-partition as deleted in the root partition table.

• Truncate operation on subpartitions will not mark affected records as deleted.

跟踪表中的数据变化¶

连接器不仅复制源表中数据的当前状态，还复制每个变更集中每行的每个状态。这些数据存储在与目标表相同的架构中创建的日志表中。

The journal table names are formatted as: <source table name>_JOURNAL_<timestamp>_<schema generation> where <timestamp> is the value of epoch seconds when the source table was added to replication, and <schema generation> is an integer increasing with every schema change on the source table. As a result, source tables that undergo schema changes will have multiple journal tables.

当表从复制中移除后又重新添加时，<timestamp> value will change, and <schema generation> will start again from 1。

The connector never drops journal tables, but does make use of the latest journal for every replicated source table, only reading append-only streams on top of journals. To reclaim the storage, you can:

• Truncate all journal tables at any time.

• Drop the journal tables related to source tables that were removed from replication.

• Drop all but the latest generation journal tables for actively replicated tables.

例如，如果您的连接器设置为主动复制源表 orders，并且您之前已将表 customers 从复制中移除，则可能存在以下日志表：在这种情况下，您可以把它们全部删除，仅保留 orders_5678_2。

customers_1234_1
customers_1234_2
orders_5678_1
orders_5678_2

配置合并任务的调度¶

连接器使用仓库将变更数据捕获 (CDC) 数据合并到目标表中。此操作由 MergeSnowflakeJournalTable 处理器触发。如果没有新的更改，或者 MergeSnowflakeJournalTable 队列中没有新的待处理的 FlowFile，则不会触发合并，仓库会自动暂停。

要限制仓库成本并将合并仅限于预定时间，可以在合并任务计划 CRON 参数中使用 CRON 表达式。它限制了流向 MergeSnowflakeJournalTable 处理器的 FlowFile，并且只能在特定的时间段内触发合并。有关调度的更多信息，请参阅 调度策略 (https://nifi.apache.org/docs/nifi-docs/html/user-guide.html#scheduling-strategy)。

停止或删除连接器¶

停止或移除连接器时，需要注意连接器使用的 复制槽 (https://www.postgresql.org/docs/current/warm-standby.html#STREAMING-REPLICATION-SLOTS)。

连接器会创建自己的复制槽，其名称以 snowflake_connector_ 开头，后跟随机后缀。当连接器读取复制流时会推进复制槽的位置，使 PostgreSQL 能够清理 WAL 日志并释放磁盘空间。

当连接器暂停时，复制槽不会被推进，对源数据库的更改会不断增加 WAL 日志大小。因此，不建议长时间暂停连接器，尤其是在高流量数据库中。

移除连接器时，无论是将其从 Openflow 画布中删除，还是通过任何其他方式（例如删除整个 Openflow 实例），复制槽都将保持原位，必须手动删除。

如果您有多个连接器实例从同一个 PostgreSQL 数据库进行复制，则每个实例将创建自己的唯一命名的复制槽。手动删除复制槽时，请确保删除的是正确的复制槽。通过检查 CaptureChangePostgreSQL 处理器的状态，您可以查看给定连接器实例使用的是哪个复制槽。