设置 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 权限的用户身份登录数据库，并执行以下查询：

CREATE PUBLICATION <publication name>;

Copy

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

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

Copy

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

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

   1. 创建一个类型为 SERVICE 的 Snowflake 用户。创建一个用于存储复制数据的数据库，并为该 Snowflake 用户设置在该数据库中创建对象所需的权限，即授予 USAGE 和 CREATE SCHEMA 权限。

      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 = 'MEDIUM'
                AUTO_SUSPEND = 300
                AUTO_RESUME = TRUE;
      GRANT USAGE, OPERATE ON WAREHOUSE <openflow_warehouse> TO ROLE <openflow_role>;
      

      Copy

   2. 创建安全密钥对（公钥和私钥）。将用户的私钥存储在文件中，以提供给连接器的配置。将公钥分配给 Snowflake 服务用户：

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

      Copy

      有关更多信息，请参阅 密钥对。

   3. 指定一个仓库供连接器使用。从 MEDIUM 仓库规模开始，然后根据要复制的表的数量和传输的数据量尝试规模。相较于扩大仓库规模，采用 多集群仓库 通常能更有效地应对海量表数量的扩展需求。

实时复制一组表¶

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

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

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

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

如果您想稍后将该表重新添加到复制中，请先在 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>",
    }
]

跟踪表中的数据变化¶

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

日志表名称的格式为：<source table name>_JOURNAL_<timestamp>_<schema generation>

其中，<timestamp> is the value of epoch seconds when the source table was added to replication, and <schema generation> 是一个整数，随着源表的每一次架构更改而递增。这意味着经过架构更改的源表将有多个日志表。

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

连接器永远不会删除日志表，但只会为每个复制的源表积极使用最新的日志。如果您想回收存储空间，可以安全地删除与已从复制中移除的源表相关的日志表，以及除最新一代主动复制表之外的所有日志表。

例如，如果您的连接器设置为主动复制源表 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 处理器的状态，您可以查看给定连接器实例使用的是哪个复制槽。