设置 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) 。

流参数¶

Start with setting the parameters of the PostgreSQL Source Parameters context, then the PostgreSQL Destination Parameters context. Once this is done, you can enable the connector, and it should connect both to PostgreSQL and Snowflake and start running. However, it will not replicate any data until any tables are explicitly added to its configuration.

要为复制配置特定的表，请编辑 PostgreSQL 引入参数上下文。在对复制参数上下文应用更改后不久，连接器会检测到这些配置，并为每张表启动复制生命周期。

从 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() 函数强制主服务器将有关正在运行的事务的信息发送到副本服务器。

Restart table replication¶

A table in FAILED state --- for example, due to a missing primary key or unsupported schema change --- does not restart automatically. If a table enters a FAILED state or you need to restart replication from scratch, use the following procedure to remove and re-add the table to replication.

1. Remove the table from flow parameters: In the Ingestion Parameters context, either remove the table from the Included Table Names or modify the Included Table Regex so the table is no longer matched.

2. Verify the table has been removed:

   1. In the Openflow runtime canvas, right-click a processor group and choose Controller Services.

   2. In the table listing controller services, locate the Table State Store row, click the three vertical dots on the right side of the row, then choose View State.

   重要

   You must wait until the table's state is fully removed from this list before proceeding. Do not continue until this configuration change has completed.

3. Clean up the destination: Once the table's state shows as fully removed, manually DROP the destination table in Snowflake. Note that the connector will not overwrite an existing destination table during the snapshot phase; if the table still exists, replication will fail again. Optionally, the journal table and stream can also be removed if they are no longer needed.

4. Re-add the table: Update the Included Table Names or Included Table Regex parameters to include the table again.

5. Verify the restart: Check the Table State Store using the instructions given previously. The state of the table should appear with the status NEW, then transition to SNAPSHOT_REPLICATION, and finally INCREMENTAL_REPLICATION.

复制表中列的子集¶

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

要对列应用筛选器，请在复制参数上下文中修改 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.

When a table is removed from replication, then added back, the <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 处理器的状态，您可以查看给定连接器实例使用的是哪个复制槽。