Set up the Openflow Connector for PostgreSQL¶

Configure wal_level¶

Openflow Connector for PostgreSQL requires wal_level to be set to logical.

Depending on where your PostgreSQL server is hosted, you can configure the wal_level as follows:

Create a publication¶

Openflow Connector for PostgreSQL requires a 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. Log in as a user with the CREATE privilege in the database and execute following query:

CREATE PUBLICATION <publication name>;

Copy

1. Define tables that the database agent will be able to see using:

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

Copy

For more information see ALTER PUBLICATION.

5. As a Snowflake account administrator, perform the following tasks:

   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.

      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. Create a pair of secure keys (public and private). Store the private key for the user in a file to supply to the connector’s configuration. Assign the public key to the Snowflake service user:

      ALTER USER openflow_user SET RSA_PUBLIC_KEY = 'thekey';
      

      Copy

      For more information, see pair of keys.

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

Replicate a set of tables in real-time¶

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

2. Populate the required parameter values as described in Flow parameters.

Remove and re-add a table to replication¶

To remove a table from replication, ensure that it is removed from the Included Table Names or Included Table Regex parameters in the Replication Parameters context.

If you want to re-add the table to replication later, first delete the corresponding destination table in Snowflake. Afterward, add the table back to the Included Table Names or Included Table Regex parameters. This ensures that the replication process starts fresh for the table.

This approach can also be used to recover from a failed table replication scenario.

Replicate a subset of columns in a table¶

The connector can filter the data replicated per table to a subset of configured columns.

To apply filters to columns, modify the Column Filter property in the Replication Parameters context, adding an array of configurations, one entry for every table you wish to apply a filter to.

Columns can be included or excluded by name or pattern. You can apply a single condition per table, or combine multiple conditions, with exclusions always taking precedence over inclusions.

The following example shows the fields that are available. schema and table are mandatory, and then one or more of included, excluded, includedPattern, excludedPattern is required.

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

Track data changes in tables¶

The connector replicates not only the current state of data from the source tables, but also every state of every row from every changeset. This data is stored in journal tables created in the same schema as the destination table.

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. This means a source table that undergoes 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 it only makes active use of the latest journal for every replicated source table. If you wish to reclaim the storage, you can safely drop the journal tables related to source tables that were removed from replication, and all but the latest generation ones 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

Configure scheduling of merge tasks¶

The connector uses a warehouse to merge change data capture (CDC) data into destination tables. This operation is triggered by the MergeSnowflakeJournalTable processor. If there are no new changes or if no new flow files are waiting in the MergeSnowflakeJournalTable queue, no merge is triggered and the warehouse auto-suspends.

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.

Stop or delete the connector¶

When stopping or removing the connector, you have to consider the replication slot 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 is not 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.

When the connector is removed, whether by deleting it from the Openflow canvas, or any other means, such as deleting the whole Openflow instance, the replication slot remains in place, and must be dropped manually.

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.