Set up the Openflow Connector for PostgreSQL¶

This topic describes the steps to set up the Openflow Connector for PostgreSQL.

For details on the incremental load process, see Incremental replication.

For information about restarting table replication for failed tables, see Restart table replication.

先决条件

1. Ensure that you have reviewed About Openflow Connector for PostgreSQL.

2. Ensure that you have reviewed the supported PostgreSQL versions.

3. Recommended: Ensure that you add only one connector instance per runtime.

4. Ensure that you have Set up Openflow - BYOC or Set up Openflow - Snowflake Deployments.

5. If using Openflow - Snowflake Deployments, ensure that you’ve reviewed configuring required domains and have granted access to the required domains for the PostgreSQL connector.

6. 作为数据库管理员，请执行以下任务：

   1. Configure wal_level
   2. Create a publication
   3. Ensure that there is enough disk space on your PostgreSQL server for the WAL. This is because once created, a replication slot causes PostgreSQL to retain the WAL data from the position held by the replication slot, until the connector confirms and advances that position.
   4. Allow at least 1 logical replication slot and 2 WAL senders per Openflow Connector for PostgreSQL connector instance on the server. Set max_replication_slots and max_wal_senders high enough to cover that and all other replication traffic on the instance.
   5. 确保启用复制的每张表都有主键。键可以是单个列或组合。
   6. Set the REPLICA IDENTITY (https://www.postgresql.org/docs/current/sql-altertable.html#SQL-ALTERTABLE-REPLICA-IDENTITY) of tables to DEFAULT. This ensures that the primary keys are represented in the WAL, and the connector can read them.
   7. Create a user for the connector. The connector requires a user with the REPLICATION attribute and permissions to SELECT from every replicated table. Create that user with a password to enter into the connector’s configuration. For more information on replication security, see Security (https://www.postgresql.org/docs/current/logical-replication-security.html).

7. 作为 Snowflake 账户管理员，请执行以下任务：

   1. Create a Snowflake user with the type as SERVICE. Create a database to store the replicated data, and set up privileges for the Snowflake user to create objects in that database by granting the USAGE and CREATE SCHEMA privileges.

      Copy codeExpand code block

      CREATE DATABASE <destination_database>;
      CREATE USER <openflow_user> TYPE=SERVICE COMMENT='Service user for automated access of Openflow';
      CREATE ROLE <openflow_role>;
      GRANT ROLE <openflow_role> TO USER <openflow_user>;
      GRANT USAGE ON DATABASE <destination_database> TO ROLE <openflow_role>;
      GRANT CREATE SCHEMA ON DATABASE <destination_database> TO ROLE <openflow_role>;
      CREATE WAREHOUSE <openflow_warehouse>
        WITH
          WAREHOUSE_SIZE = 'XSMALL'
          AUTO_SUSPEND = 300
          AUTO_RESUME = TRUE;
      GRANT USAGE, OPERATE ON WAREHOUSE <openflow_warehouse> TO ROLE <openflow_role>;

      Show lessSee more

      Scroll to top

   2. Create a pair of secure keys (public and private). Store the private key for the user in a file to use while configuring the connector. Assign the public key to the Snowflake service user:

      Copy codeExpand code block

      ALTER USER <openflow_user> SET RSA_PUBLIC_KEY = 'thekey';

      Show lessSee more

      Scroll to top

      For more information, see key-pair authentication.

   3. Designate a warehouse for the connector to use. Start with the XSMALL warehouse size, then experiment with size depending on the amount of tables being replicated, and the amount of data transferred. Large numbers of tables typically scale better with multi-cluster warehouses, rather than the warehouse size.

配置 wal_level¶

Openflow Connector for PostgreSQL requires wal_level (https://www.postgresql.org/docs/current/runtime-config-wal.html#GUC-WAL-LEVEL) to be set to logical.

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

创建发布

Openflow Connector for PostgreSQL requires a publication (https://www.postgresql.org/docs/current/logical-replication-publication.html#LOGICAL-REPLICATION-PUBLICATION) to be created and configured in PostgreSQL before replication starts. You can create it for all, or a subset of tables, as well as for specific tables with specified columns only. Make sure that every table and column that you plan to have replicated is included in the publication. You can also modify the publication later, while the connector is running. To create and configure a publication, do the following:

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

   • For PostgreSQL 13 and later:

     Copy codeExpand code block

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

     Show lessSee more

     Scroll to top

     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:

     Copy codeExpand code block

     CREATE PUBLICATION <publication name>;

     Show lessSee more

     Scroll to top

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

Copy codeExpand code block

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

Show lessSee more

Scroll to top

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

Important

PostgreSQL 15 and later support configuring publications for a specified subset of table columns. For the connector to support this correctly, you must use the column filtering settings to include the same columns as set on the publication.

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

• In the destination table, columns that aren’t included in the filter will be suffixed with __DELETED. All data replicated during the snapshot phase will be retained.
• After you add new columns to the publication, the table will be permanently failed, and you will need to restart its replication.

For more information, see ALTER PUBLICATION (https://www.postgresql.org/docs/current/sql-alterpublication.html).

安装连接器

Runtime sizing¶

For migration instructions, see Reinstall the connector.

配置连接器

To configure the connector, do the following as a data engineer:

1. Right-click on the imported process group and select Parameters.

2. Populate the required parameter values.

   For more information on the required parameter values, see the following sections:

   • PostgreSQL Source Parameters: Used to establish a connection with PostgreSQL.
   • PostgreSQL Destination Parameters: Used to establish a connection with Snowflake.
   • PostgreSQL Ingestion Parameters: Used to specify the tables to replicate.

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 won’t replicate any data until any tables are explicitly added to its configuration.

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

PostgreSQL Source Parameters¶

PostgreSQL Destination Parameters¶

PostgreSQL Ingestion Parameters¶

从 PostgreSQL 副本服务器复制表¶

The connector can ingest data from a primary server, a hot standby replica (https://www.postgresql.org/docs/current/hot-standby.html), or subscriber server using logical replication (https://www.postgresql.org/docs/current/logical-replication.html). Before configuring the connector to connect to a PostgreSQL replica, ensure that replication between primary and replica nodes works correctly. When investigating issues with missing data in the connector, first ensure that missing rows are present in replica server used by the connector.

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

• PostgreSQL version of the server must be >= 16. Amazon Aurora is not supported because it doesn’t offer logical decoding from read replicas (https://docs.aws.amazon.com/AmazonRDS/latest/AuroraUserGuide/AuroraPostgreSQL.Replication.Logical.html).
• Only connecting to hot standby replica is supported. Note that warm standby replicas can’t accept connections from clients until they are promoted to a primary instance.
• The publication needed by the connector must be created on the primary server, not the standby server. The standby server is read-only and doesn’t allow to create publication.

If you connect to a hot standby instance and see 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. error in the Openflow bulletin, or the Read PostgreSQL CDC Stream processor isn’t starting, log in to the primary PostgreSQL instance and execute the following query:

SELECT pg_log_standby_snapshot();

The error occurs when there are no data changes in the primary server. As such the connector can stall while creating a replication slot on the replica server. This results from the replica server requiring information about running transactions from the primary server to be able to create a replication slot. Primary servers won’t send the information while idle. The pg_log_standby_snapshot() function forces the primary server to send information about running transactions to the replica server.

复制表中列的子集

[
    {
        "schema": "<schema>" | "schemaPattern": "<regex>",
        "table": "<table>" | "tablePattern": "<regex>",
        "included": ["<column>", "<column>"],
        "excluded": ["<column>", "<column>"],
        "includedPattern": "<regex>",
        "excludedPattern": "<regex>"
    }
]

[
  {
    schema: "public",
    table: "orders",
    included: ["account_id", "status", "created_at"],
  },
];

[
  {
    schema: "public",
    table: "orders",
    excluded: ["internal_note", "debug_flag"],
  },
];

[
  {
    schema: "public",
    table: "contacts",
    includedPattern: ".*_email",
    excluded: ["admin_email"],
  },
];

[
  {
    schemaPattern: "data_.*",
    table: "customers",
    excluded: ["internal_note"],
  },
];

[
  { schema: "public", table: "orders", included: ["account_id", "status"] },
  { schema: "public", table: "customers", excludedPattern: ".*_internal" },
];

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 won’t 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.

For example, if your connector is set to actively replicate source table orders, and you have earlier removed table customers from replication, you may have the following journal tables. In this case you can drop all of them except orders_5678_2.

customers_1234_1
customers_1234_2
orders_5678_1
orders_5678_2

配置合并任务的调度

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

To limit the warehouse cost and limit merges to only scheduled time, use the CRON expression in the Merge task Schedule CRON parameter. It throttles the flow files coming to the MergeSnowflakeJournalTable processor and merges are triggered only in a dedicated period of time. For more information about scheduling, see Scheduling strategy (https://nifi.apache.org/docs/nifi-docs/html/user-guide.html#scheduling-strategy).

停止或删除连接器

When stopping or removing the connector, you have to consider the replication slot (https://www.postgresql.org/docs/current/warm-standby.html#STREAMING-REPLICATION-SLOTS) that the connector uses.

The connector creates its own replication slot with a name starting with snowflake_connector_ followed by a random suffix. As the connector reads the replication stream, it advances the slot, so that PostgreSQL can trim its WAL log and free up disk space.

When the connector is paused, the slot isn’t advanced, and changes to the source database keep increasing the WAL log size. You should not keep the connector paused for extended periods of time, especially on high-traffic databases.

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

If you have multiple connector instances replicating from the same PostgreSQL database, each instance will create its own uniquely-named replication slot. When dropping a replication slot manually, make sure it’s the right one. You can see which replication slot is used by a given connector instance by checking the state of the CaptureChangePostgreSQL processor.

运行流

1. Right-click on the plane and select Enable all Controller Services.
2. Right-click on the imported process group and select Start. The connector starts the data ingestion.