Set up the Openflow Connector for SQL Server¶

This topic describes how to set up the Openflow Connector for SQL Server.

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

Prerequisites¶

Before setting up the connector, ensure that you have completed the following prerequisites:

1. Ensure that you have reviewed About Openflow Connector for SQL Server.

2. Ensure that you have reviewed Supported SQL Server versions.

3. Ensure that you have set up your runtime deployment. For more information, see the following topics:

   • Set up Openflow - BYOC
   • Set up Openflow - Snowflake Deployments.

4. If you use Openflow - Snowflake Deployments, ensure that you have reviewed configuring required domains and have granted access to the required domains for the SQL Server connector.

Set up your SQL Server instance¶

Before setting up the connector, perform the following tasks in your SQL Server environment:

1. Enable change tracking on the databases (https://learn.microsoft.com/en-us/sql/relational-databases/track-changes/enable-and-disable-change-tracking-sql-server?view=sql-server-ver16#enable-change-tracking-for-a-database) and tables (https://learn.microsoft.com/en-us/sql/relational-databases/track-changes/enable-and-disable-change-tracking-sql-server?view=sql-server-ver16#enable-change-tracking-for-a-table) that you plan to replicate, as shown in the following SQL Server example:

   Copy code

   ALTER DATABASE <database>
     SET CHANGE_TRACKING = ON
     (CHANGE_RETENTION = 5 DAYS, AUTO_CLEANUP = ON);
   
   ALTER TABLE <schema>.<table>
     ENABLE CHANGE_TRACKING
     WITH (TRACK_COLUMNS_UPDATED = OFF);
   

   Note

   Run the ALTER DATABASE command only once per database to enable change tracking on that database. Run the ALTER TABLE command for every table that you plan to replicate in that database.

   Important

   Keep the column tracking feature off by setting TRACK_COLUMNS_UPDATED = OFF (the default) when you enable change tracking on a table. The connector doesn’t use column-level change information, so enabling TRACK_COLUMNS_UPDATED = ON adds unnecessary storage and per-DML overhead on the source without providing any benefit to replication.

   Each table must have change tracking enabled before the connector can replicate it. Enable change tracking on every table you want to replicate before you start replication.

   Important

   Set CHANGE_RETENTION to a window that’s long enough to keep change data available for data continuity. SQL Server removes tracked changes once they’re older than the retention period, so if the connector can’t read changes before they expire, it can’t replicate them incrementally and the affected tables require a full reload.

   A longer retention window, such as 5 days, gives you time to detect and react to problems, such as a paused connector, a network outage, or a failed replication cycle, and to mitigate them without reloading entire tables. Choose a value that covers the longest interruption you expect to recover from, balanced against the additional storage that retained change data consumes in SQL Server.

2. Create a login for the SQL Server instance:

   Copy code

   CREATE LOGIN <user_name> WITH PASSWORD = '<password>';
   

   This login is used to create users for the databases you plan to replicate.

3. Create a user for each database you are replicating by running the following SQL Server command in each database:

   Copy code

   USE <source_database>;
   CREATE USER <user_name> FOR LOGIN <user_name>;
   

4. Grant the SELECT and VIEW CHANGE TRACKING permissions to the user for each database that you are replicating:

   Copy code

   GRANT SELECT ON <database>.<schema>.<table> TO <user_name>;
   GRANT VIEW CHANGE TRACKING ON <database>.<schema>.<table> TO <user_name>;
   

   Run these commands in each database for every table that you plan to replicate. These permissions must be granted to the user of each database that you created in a previous step.

5. (Optional) Grant the VIEW DEFINITION privilege on the User Defined Data Types (UDDT).

   If your tables contain columns that use User Defined Data Types (UDDT), and the UDDT is owned by a different user than the connector user, you must grant the VIEW DEFINITION permission to the connector user as shown in the following SQL Server example:

   Copy code

   GRANT VIEW DEFINITION TO <user_name>;
   

   Without this permission, columns using UDDT are silently excluded from replication.

6. (Optional) Configure SSL connection.

   If you use an SSL connection to connect SQL Server, create the root certificate for your database server. This is required when configuring the connector.

Set up your Snowflake environment¶

As a Snowflake administrator, perform the following tasks:

1. Create a destination database in Snowflake to store the replicated data:

   Copy code

   CREATE DATABASE <destination_database>;
   

2. Create a Snowflake service user:

   Copy code

   CREATE USER <openflow_user>
     TYPE = SERVICE
     COMMENT='Service user for automated access of Openflow';
   

3. Create a Snowflake role for the connector and grant the required privileges:

   Copy code

   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>;
   

   Use this role to manage the connector’s access to the Snowflake database.

   To create objects in the destination database, you must grant the USAGE and CREATE SCHEMA privileges on the database to the role used to manage access.

4. Create a Snowflake warehouse for the connector and grant the required privileges:

   Copy code

   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>;
   

   Snowflake recommends starting with a XSMALL warehouse size, then experimenting with size depending on the number of tables being replicated and the amount of data transferred. Large numbers of tables typically scale better with multi-cluster warehouses, rather than a larger warehouse size. For more information, see multi-cluster warehouses.

5. Set up the public and private keys for key pair authentication:

   1. Create a pair of secure keys (public and private).

   2. Store the private key for the user in a file to supply to the connector’s configuration.

   3. Assign the public key to the Snowflake service user:

      Copy code

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

      For more information, see Key-pair authentication and key-pair rotation.

Install the connector¶

Runtime sizing¶

For migration instructions, see Reinstall the connector.

Configure 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:

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

Start by setting the parameters of the SQLServer Source Parameters context, then the SQLServer Destination Parameters context. After you complete this, enable the connector. The connector connects to both SQLServer and Snowflake and starts running. However, the connector doesn’t replicate any data until any tables to be replicated are explicitly added to its configuration.

To configure specific tables for replication, edit the SQLServer Ingestion Parameters context. After you apply the changes to the SQLServer Ingestion Parameters context, the configuration is picked up by the connector, and the replication lifecycle starts for every table.

SQLServer Source Parameters¶

For failover behavior, see Always On Availability Groups and source failover.

SQLServer Destination Parameters¶

SQLServer Ingestion Parameters¶

Read the source under SNAPSHOT isolation¶

The connector reads the source tables during both snapshot and incremental replication. Under SQL Server’s default READ COMMITTED isolation level, these reads acquire shared locks that can deadlock with concurrent writes from other database clients. To avoid these deadlocks without affecting the isolation level that other applications use, configure the connector to read under SNAPSHOT isolation (https://learn.microsoft.com/en-us/dotnet/framework/data/adonet/sql/snapshot-isolation-in-sql-server). For background, see Source database locking behavior.

Enable SNAPSHOT isolation for the connector in two steps:

1. On each source database, allow snapshot isolation:

   Copy code

   ALTER DATABASE <database> SET ALLOW_SNAPSHOT_ISOLATION ON;
   

2. Add a dynamic property named Use Snapshot Isolation with the value true to the connector’s processors. Because the connector reads the source tables during both snapshot and incremental replication, add this dynamic property to both the MultiDatabaseFetchTableSnapshot processor and the MultiDatabaseCaptureChangeSqlServer processor.

The connector checks each source database when it starts and uses SNAPSHOT isolation only for databases that have ALLOW_SNAPSHOT_ISOLATION enabled. For a database that doesn’t have it enabled, the connector falls back to the default isolation level. Because this check runs at startup, restart the processors after you change ALLOW_SNAPSHOT_ISOLATION.

Replicate tables from a SQL Server replica server¶

The connector can ingest data from a primary server or from a subscriber server using transactional replication (https://learn.microsoft.com/en-us/sql/relational-databases/replication/transactional/transactional-replication). Before configuring the connector to connect to a SQL Server replica, ensure that replication between the primary and replica nodes works correctly. For instructions on setting up transactional replication, see Tutorial: Configure transactional replication (https://learn.microsoft.com/en-us/sql/relational-databases/replication/tutorial-replicating-data-between-continuously-connected-servers). When investigating issues with missing data in the connector, first ensure that missing rows and change tracking events are present in the replica server used by the connector.

To configure the connector to read from a subscriber server instead of the publisher, specify the subscriber server URL in the SQLServer Connection URL parameter.

Restart table replication¶

Replicate a subset of columns in a table¶

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

[
    {
        "database": "my_db",
        "schema": "dbo",
        "table": "orders",
        "included": ["account_id", "status", "created_at"]
    }
]

[
    {
        "database": "my_db",
        "schema": "dbo",
        "table": "orders",
        "excluded": ["internal_note", "debug_flag"]
    }
]

[
    {
        "database": "my_db",
        "schema": "dbo",
        "table": "contacts",
        "includedPattern": ".*_email",
        "excluded": ["admin_email"]
    }
]

[
    {
        "databasePattern": "prod_.*",
        "schema": "dbo",
        "table": "customers",
        "excluded": ["internal_note"]
    }
]

[
    {"database": "my_db", "schema": "dbo", "table": "orders", "included": ["account_id", "status"]},
    {"database": "my_db", "schema": "dbo", "table": "customers", "excludedPattern": ".*_internal"}
]

Replicate a partitioned table¶

The connector supports replication of partitioned tables. A SQL Server partitioned table is replicated into Snowflake as a single destination table, containing data from all partitions.

To replicate a partitioned table, ensure that change tracking is enabled on the partitioned table, as described in Set up your SQL Server instance.

Performance on partitioned tables¶

The connector supports partitioned tables. When the connector reads from a partitioned table, it places the partition column first in the queries it sends to SQL Server. This lets SQL Server use partition elimination to quickly narrow the read to the relevant partitions instead of scanning every partition, which significantly reduces the time and resources required to replicate large partitioned tables.

This behavior applies whether or not the partition column is part of the table’s primary key. You don’t need to configure anything to benefit from it. The connector applies the optimization automatically for all partitioned tables.

Track data changes in tables¶

The connector replicates the current state of data from the source tables, as well as detected changes from each polling interval. 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. As a result, source tables that undergo schema changes will have multiple journal tables.

When you remove a table from replication, then add it back, the <timestamp> value changes, and <schema_generation> starts again from 1.

The connector never drops journal tables, but uses 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

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.

Use the CRON expression in the Merge task Schedule CRON parameter to limit the warehouse cost and limit merges to only scheduled time. 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).

Run the flow¶

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.