Snowflake

Prerequisites

A Snowflake account granted ACCOUNTADMIN system-defined role or custom role with privileges to:

CREATE WAREHOUSE, DATABASE, SCHEMA
CREATE ROLE, USER
CREATE NETWORK POLICY

Snowflake Setup

It’s recommended to create a separate user and role for Streamkap to access your Snowflake database. Below is an example script that does that.

-- We've provided defaults so, change these as required with names for Database Objects in 'UPPERCASE'
SET user_name           = UPPER('STREAMKAP_USER');
SET user_password       = '{password}'; -- IMPORTANT: Make sure to change this!
SET warehouse_name      = UPPER('STREAMKAP_WH'); -- Used for automatic QA, UPSERT mode and optional views
SET database_name       = UPPER('STREAMKAPDB');
SET schema_name         = UPPER('STREAMKAP');
SET role_name           = UPPER('STREAMKAP_ROLE');
SET network_policy_name = UPPER('STREAMKAP_NETWORK_ACCESS');

-- If your Snowflake account uses custom roles to grant privileges, change these values below
SET sysadmin_role       = UPPER('SYSADMIN');
SET securityadmin_role  = UPPER('SECURITYADMIN');
SET accountadmin_role   = UPPER('ACCOUNTADMIN');

-- Create a warehouse with defaults:
-- Standard, X-Small, No Scaling, Auto-Suspend after 1 Minute
USE ROLE IDENTIFIER($sysadmin_role);
CREATE WAREHOUSE IF NOT EXISTS IDENTIFIER($warehouse_name) AUTO_SUSPEND =1;

-- Create a database and schema for Streamkap
USE WAREHOUSE IDENTIFIER($warehouse_name);
CREATE DATABASE IF NOT EXISTS IDENTIFIER($database_name);
USE DATABASE IDENTIFIER($database_name);
CREATE SCHEMA IF NOT EXISTS IDENTIFIER($schema_name);

-- Create a Snowflake role with privileges for the Streamkap connector
USE ROLE IDENTIFIER($securityadmin_role);
CREATE ROLE IF NOT EXISTS IDENTIFIER($role_name);

-- Grant privileges on the warehouse
GRANT USAGE ON WAREHOUSE IDENTIFIER($warehouse_name) TO ROLE IDENTIFIER($role_name);

-- Grant privileges on the database
GRANT USAGE ON DATABASE IDENTIFIER($database_name) TO ROLE IDENTIFIER($role_name);
-- Optional: Permissions for auto schema creation on the database
GRANT CREATE SCHEMA ON DATABASE IDENTIFIER($database_name) TO ROLE IDENTIFIER($role_name);

-- Grant privileges on the database schema
USE ROLE IDENTIFIER($sysadmin_role);
USE DATABASE IDENTIFIER($database_name);
GRANT USAGE ON SCHEMA IDENTIFIER($schema_name) TO ROLE IDENTIFIER($role_name);
GRANT CREATE TABLE ON SCHEMA IDENTIFIER($schema_name) TO ROLE IDENTIFIER($role_name);
GRANT CREATE FILE FORMAT ON SCHEMA IDENTIFIER($schema_name) TO ROLE IDENTIFIER($role_name);
GRANT CREATE STAGE ON SCHEMA IDENTIFIER($schema_name) TO ROLE IDENTIFIER($role_name);
GRANT CREATE PIPE ON SCHEMA IDENTIFIER($schema_name) TO ROLE IDENTIFIER($role_name);

-- Grant privileges for dynamic table and task creation (Only if auto-creation is enabled)
GRANT CREATE DYNAMIC TABLE ON SCHEMA IDENTIFIER($schema_name) TO ROLE IDENTIFIER($role_name);
GRANT CREATE TASK ON SCHEMA IDENTIFIER($schema_name) TO ROLE IDENTIFIER($role_name);
USE ROLE IDENTIFIER($accountadmin_role);
GRANT EXECUTE TASK ON ACCOUNT TO ROLE IDENTIFIER($role_name);

-- Create a user for Streamkap
USE ROLE IDENTIFIER($securityadmin_role);
CREATE USER IDENTIFIER($user_name) PASSWORD = $user_password DEFAULT_ROLE = $role_name;

-- Grant the custom role to the Streamkap user
GRANT ROLE IDENTIFIER($role_name) TO USER IDENTIFIER($user_name);

-- Set the custom role as the default role for the Streamkap user.
-- If you encounter an 'Insufficient privileges' error, verify the '$securityadmin_role' has OWNERSHIP privilege on the '$user_name'.
ALTER USER IDENTIFIER($user_name) SET DEFAULT_ROLE = $role_name;

-- Prevents Snowflake getting confused in an edge-case where a table exists in the public schema and current schema with the same name.
ALTER USER IDENTIFIER($user_name) SET SEARCH_PATH = '$current';

-- Allow the Streamkap user access to the Snowflake account
-- Latest IPs can be found here: https://docs.streamkap.com/docs/streamkap-ip-addresses
-- If you need to edit the network policy, you can use:
-- ALTER NETWORK POLICY STREAMKAP_NETWORK_ACCESS SET ALLOWED_IP_LIST=('52.32.238.100');
CREATE NETWORK POLICY IDENTIFIER($network_policy_name) ALLOWED_IP_LIST=('52.32.238.100');
ALTER USER IDENTIFIER($user_name) SET NETWORK_POLICY = $network_policy_name;

We do not use CREATE OR REPLACE in our scripts. This is to avoid destroying something by mistake that already exists in your Snowflake account.

Warehouse auto-suspend for CDC workloadsThe setup script above sets AUTO_SUSPEND = 1 (1 minute). For CDC workloads — especially in upsert mode where a warehouse must be running — very short auto-suspend timeouts can cause frequent suspend/resume cycles. Each resume incurs provisioning latency (up to several seconds) and additional Snowflake credit charges, since warehouses are billed in 60-second minimum increments per resume.For active CDC pipelines, consider setting AUTO_SUSPEND to 60 seconds or higher and ensure the warehouse size matches your ingestion throughput. In append mode with Snowpipe Streaming, the warehouse is only used for optional operations (QA, dynamic tables, tasks), so a short auto-suspend is acceptable.

Key Pair Authentication

The connector relies on an RSA key pair for authentication which you can generate using OpenSSH. Below are example scripts that do that. You can modify them to suit your security policies, but please ensure the key pair meets these minimum requirements:

RSA 2048-bit
PKCS#8 key format

SSH key generation on WindowsSnowflake does not support keys generated by PuTTY Key Generator.One of the easiest and quickest ways to generate a valid OpenSSL key is via Git Bash which is installed by default with Git for Windows. After installation, you can open a Git Bash prompt by Left Shift + Right Clicking on your Desktop, choosing “Open Git Bash here” and then executing the OpenSSL commands below.If you have any issues following these instructions or are unable to install Git for Windows, please contact us.

# Make sure to change '{passphrase}' to a password of your choice
openssl genrsa 2048 | openssl pkcs8 -topk8 -v2 aes256 -inform PEM -out streamkap_key.p8 -passout pass:{passphrase}

# generates the public key, referencing the private key
# Don't forget to replace '{passphrase}' with the password used in the previous command
openssl rsa -in streamkap_key.p8 -pubout -out streamkap_key.pub -passin pass:{passphrase}

The scripts above should create two files (the key pair), one private (may have an extension e.g. .p8) and the other public (usually has the extension .pub). Store both files in a secure place. Once generated, the public key needs to be assigned to the Snowflake database user created for Streamkap earlier. This command will copy the public key you generated to your clipboard.

egrep -v '^-|^$' ./streamkap_key.pub | pbcopy

Now attach the public key to the user:

-- We've provided a default, so change this as required
SET user_name = UPPER('STREAMKAP_USER');

USE ROLE SECURITYADMIN;

-- Replace '{public key}' below with the public key file contents
-- If you used the previous command to copy the key to your clipboard, use Ctrl+V (Windows)
-- or Cmd+V (MacOS) to replace the '{public key}' placeholder with the key
-- Key part MUST start with 'MII' excluding any headers and footers
ALTER USER IDENTIFIER($user_name) SET RSA_PUBLIC_KEY = '{public key}';

Streamkap Setup

Follow these steps to configure your new connector:

1. Create the Destination

Navigate to Add Connectors.
Choose Snowflake.

2. Connection Settings

Name: Enter a name for your connector.
Snowflake URL: The URL for accessing your Snowflake account. This URL must include your account identifier. Note that the protocol (https://) and port number are optional.
Username: User login name for the Snowflake account (Case sensitive).
Private Key: Provide the private key you generated by using the command below.
egrep -v '^-|^$' ./streamkap_key.p8 | pbcopy
- Key secured with passphrase?: If checked (default), provide your SSH key’s passphrase, otherwise, uncheck for SSH keys without passphrase.
  - Private Key Passphrase: The passphrase is used to decrypt the private key.
Database Name: The name of the database to use (Case sensitive).
Schema Name: The name of the schema where tables will be created (Case sensitive).
Snowflake Role: The name of an existing role with necessary privileges (for Streamkap) assigned to the user specified by Username (Case sensitive).

3. Ingestion Settings

Ingestion Mode: How the Connector loads data into the Snowflake tables. See Upsert mode for further details.
Changing ingestion modeappend and upsert modes use different, incompatible methods for loading data into the Snowflake tables. If - for whatever reason - you want to change modes for an existing Snowflake Connector, please create a new Snowflake Destination instead i.e. a separate destination for append, and for upsert.
- appendmode:
  - Use Dynamic Tables: Specifies whether the connector should create Dynamic Tables & Cleanup Tasks. See Dynamic Tables.
    - Custom SQL Template - Dynamic Table Creation: These template queries run for each table the first time a record is streamed for them.
    - Custom SQL Template - Dynamic Table Name: Can be used as {{dynamicTableName}} in dynamic table creation SQL. It can use input JSON data for more complex mappings and logic.
    - Custom SQL Template - Input JSON data: Use {"TABLE_DATA": {"{table_name}": {"{key}": "{value}"}, ...}, ...} to set table specific data. This data will be available in the custom SQL templates e.g. SELECT {{key}}.
    - Auto QA Deduplication Table Mapping: Mapping between the tables that store append-only data and the deduplicated tables. The dedupeTable in mapping will be used for QA scripts. If dedupeSchema is not specified, the deduplicated table will be created in the same schema as the raw table.
- upsertmode:
  - Delete Mode: Specifies whether the connector processes deletions (or tombstone events) and removes the corresponding row from the database.
  - Use Hybrid Tables: Specifies whether the connector should create Hybrid Tables.

Click Save.

Troubleshooting

Dynamic Tables

Snowflake Dynamic Tables are materialized views which consist of the latest records inserted into Snowflake. Streamkap’s Snowflake Connector creates them—if enabled—for each table the first time a record is streamed for them. A Snowflake Task is also created for each dynamic table to clean up older entries periodically. Below is the default template—shown in the Streamkap UI. You can modify it there to suit your requirements.

  CREATE OR REPLACE DYNAMIC TABLE {{table}}_DT
    TARGET_LAG = '15 minutes' -- Minimum is 1 minute
    WAREHOUSE = {{warehouse}}
    AS
  SELECT * EXCLUDE dedupe_id
  FROM (
    SELECT *,
      ROW_NUMBER() OVER (
        PARTITION BY {{primaryKeyColumns}}
        ORDER BY _streamkap_ts_ms DESC, _streamkap_offset DESC
      ) AS dedupe_id
    FROM {{table}}
  )
  WHERE dedupe_id = 1         -- Latest record
    AND __deleted = 'false';  -- Excluding deleted records

  CREATE OR REPLACE TASK {{table}}_CT -- This statement and the `ALTER TASK {{table}}_CT RESUME;` can be removed if you don't want to clean up old records
    WAREHOUSE = {{warehouse}}
    SCHEDULE = '4380 minutes' -- We don't recommend changing this to a very short interval (e.g. 30 minutes) as it can increase Snowflake costs
    TASK_AUTO_RETRY_ATTEMPTS = 3
    ALLOW_OVERLAPPING_EXECUTION = FALSE
    AS
  DELETE FROM {{table}}
  WHERE NOT EXISTS (
    SELECT 1
    FROM (
      SELECT {{primaryKeyColumns}}, MAX(_streamkap_ts_ms) AS max_timestamp
      FROM {{table}}
      GROUP BY {{primaryKeyColumns}}
    ) AS subquery
    WHERE {{{keyColumnsAndCondition}}}
      AND {{table}}._streamkap_ts_ms = subquery.max_timestamp
  );

  ALTER TASK {{table}}_CT RESUME;

Offset Management (Append Mode)

Streamkap retains topic data based on your service’s retention policy (typically 7 days by default). You can only replay messages that are still within the retention window.

Understanding Dual Offset Tracking

When using append mode with Snowflake destinations, there are two, separate offset systems to manage:

Consumer Group Offsets (connect-{connector-id})
- Tracks which messages the connector has consumed from the source topic
- Visible in Streamkap UI under Consumer Groups
Snowflake Channel Offsets
- Tracks which messages have been successfully ingested into Snowflake via Snowpipe Streaming
- Each topic partition creates a Snowflake channel (e.g., TOPIC_0 for partition 0)
- Managed within Snowflake using SYSTEM$SNOWPIPE_STREAMING_UPDATE_CHANNEL_OFFSET_TOKEN

These two offset systems can become misaligned, especially when topics are deleted and recreated.Symptoms of misalignment can include:

New data is expected but not appearing in the destination tables
Lag showing as negative (e.g. -1) or unusually high in Streamkap UI
Snowflake channels showing offset positions that don’t match the Consumer Group offsets (via SHOW CHANNELS command)
Missing data in destination table

Critical requirement: Destinations must be stopped before attempting any offset reset operations.

Replaying Messages to Snowflake

To replay messages (resend data to Snowflake), you must coordinate resets across both offset systems: Before you start:

Ensure you have the required Snowflake privileges (e.g., OWNER on the table)
Note the offset position you want to replay to (earliest, specific offset, or timestamp)
Plan for the replay window within your topic retention period

Step 1: Stop the Destination Connector In Streamkap UI, navigate to the connector and click Stop or Pause. Wait for the connector status to show as stopped. This prevents Snowpipe Streaming from committing offsets while you reset them. Step 2: Reset Consumer Group Offsets Use Streamkap UI to reset the consumer group:

Go to Consumer Groups
Find the consumer group for the destination connector (e.g. connect-{connector-id})
Follow the Consumer Groups Reset Procedure to reset to your desired position

To find the connector ID, navigate to the connector details page. The ID appears in the URL, or by clicking Copy ID in the quick actions menu (three dots) at the top right.

Step 3: Reset Snowflake Channel Offsets Connect to Snowflake (using your preferred client: SnowSQL, DBeaver, or Snowflake web UI) and reset each channel to -1. This tells Snowflake to defer to the Consumer Group offset position.

-- First, view current channels and their offsets
SHOW CHANNELS IN SCHEMA <database>.<schema>;

-- For each channel, reset the offset to -1
-- Replace <DATABASE>, <SCHEMA>, <TABLE_NAME>, and <TOPIC_NAME_0> with your actual values

SELECT SYSTEM$SNOWPIPE_STREAMING_UPDATE_CHANNEL_OFFSET_TOKEN(
    '<DATABASE>.<SCHEMA>.<TABLE_NAME>',
    '<TOPIC_NAME_0>',  -- Channel name (usually TOPIC_PARTITION_NUMBER, e.g., ORDERS_0, ORDERS_1)
    '-1'               -- Reset to -1 so Snowflake uses the Consumer Group offset
);

Example: Reset channel to use Consumer Group offset

-- Reset topic "orders" partition 0 to use Consumer Group offset
SELECT SYSTEM$SNOWPIPE_STREAMING_UPDATE_CHANNEL_OFFSET_TOKEN(
    'MY_DB.MY_SCHEMA.ORDERS_TABLE',
    'ORDERS_0',
    '-1'
);

-- Repeat for each partition (ORDERS_1, ORDERS_2, etc.)

Setting the Snowflake channel offset to -1 simplifies the reset process. Instead of manually aligning offset numbers between the Consumer Group and Snowflake channels, the channel will automatically start from the Consumer Group’s current offset position.

Step 4: Resume the Destination Connector In Streamkap UI, click Resume or Start on the connector. Step 5: Verify the Replay Monitor the connector and destination table to confirm:

Connector status shows as active/running
Consumer lag decreases as messages are re-ingested
Data appears in the destination table

Offset Reset Strategies

The offset position is controlled by the Consumer Group. When resetting offsets, choose one of the following strategies in Streamkap UI (see Consumer Groups Reset Procedure):

Strategy	Description	Use Case
Earliest	Reset to the beginning of the partition	Replay all available messages within retention window
Latest	Reset to the end of the partition	Skip all existing messages and start fresh
Specific Timestamp	Reset to the first offset after a given timestamp	Replay messages from a specific point in time
Specific Offset	Set a custom offset position	Precise control over where to resume

After resetting the Consumer Group offset to your desired position, reset the Snowflake channel offset to -1 so it defers to the Consumer Group position:

SELECT SYSTEM$SNOWPIPE_STREAMING_UPDATE_CHANNEL_OFFSET_TOKEN(
    '<DATABASE>.<SCHEMA>.<TABLE_NAME>',
    '<TOPIC_NAME_0>',
    '-1'
);

Upsert mode

Snowflake destination connector can run in upsert mode. This mode switches off the use of snowpipe streaming and connector uses periodic MERGE INTO statements to upsert data into target snowflake tables. Dynamic tables or other de-duplication mechanisms will not be necessary when using upsert mode.

Snowflake costsCurrently upsert mode requires a warehouse to be running so overall the costs will be higher compared to append mode which uses snowpipe streaming.

Getting the Snowflake URL

You can also run the script below in a Snowflake worksheet to return the Snowflake URL. You need to be logged into Snowflake with an account granted ORGADMIN system-defined role to run this script.

USE ROLE ORGADMIN;

-- Snowflake URL is the 'account_url', not 'account_locator_url'
SHOW ACCOUNTS;

Snowflake Setup scripts failing

There can be many reasons for them to fail, but the scripts below can help you diagnose the issues. You need to be logged into Snowflake with an account granted ACCOUNTADMIN system-defined role or custom role with equivalent privileges to run these scripts. Copy paste the scripts below into Snowflake worksheets. Change the object names at the top as required and run all queries.

-- Replace object names below with the names used by your Snowflake Setup script
SET warehouse_name      = UPPER('STREAMKAP_WH');
SET database_name       = UPPER('STREAMKAPDB');
SET schema_name         = UPPER('STREAMKAP');
SET role_name           = UPPER('STREAMKAP_ROLE');
SET user_name           = UPPER('STREAMKAP_USER');
SET network_policy_name = UPPER('STREAMKAP_NETWORK_ACCESS');

-- If your Snowflake account uses custom roles to grant privileges, change the role name below
SET accountadmin_role   = UPPER('ACCOUNTADMIN');

USE ROLE IDENTIFIER($accountadmin_role);

-- If any of the queries fail or return no results, '$accountadmin_role' doesn't have necessary privileges, or the object doesn't exist
-- Warehouses, databases and schemas
DESC WAREHOUSE IDENTIFIER($warehouse_name);
DESC DATABASE IDENTIFIER($database_name); -- Displays schemas; no need for DESC SCHEMA ... query also

-- Users
DESC USER IDENTIFIER($user_name); -- Shows name, defaults and RSA details; no need for separate queries

-- Network policies
DESC NETWORK POLICY IDENTIFIER($network_policy_name);

If any of the queries return an error or no results:

Check in the top right corner (next to Share and Run buttons) of the Snowflake Worksheets that the role is set to ACCOUNTADMIN, or a custom role with equivalent privileges
Depending on which query failed or returned no results, check the object names at the top of the script are correct
If a query returns "Object does not exist or is not authorized" error, go to the Snowsight UI Admin page and see if the object is showing there. For example, if DESC WAREHOUSE ... failed, go to Admin -> Warehouses page and check if the Warehouse is shown on that page

If the warehouse, database, schema, role and user exists, privileges might be an issue. Run Script #2 and ensure the privileges displayed match or include the following:

Troubleshooting Common Issues

503 / NullPointerException errors

Transient Snowflake Streaming API overload. Do NOT restart the connector — the built-in retry mechanism with exponential backoff handles recovery automatically, typically within 5-30 minutes.If the error persists beyond 30 minutes, contact Streamkap support.

Records going to DLQ (16 MB limit)

Snowflake has a maximum record size of 16 MB. Records exceeding this limit are routed to the dead letter queue (DLQ).Resolution:

Identify the oversized columns in the DLQ message payload
Exclude large columns from replication if they are not needed at the destination
Add a transform to truncate large fields before they reach the destination

VARCHAR overflow — column value exceeds size

When a source column value exceeds the VARCHAR size defined at the destination, the record is routed to the DLQ.Resolution:

Increase the column size at the destination: ALTER TABLE ... ALTER COLUMN ... TYPE VARCHAR(n)
Configure a transform (SMT) to truncate values before delivery if increasing the column size is not feasible

Schema Evolution Permissions

Streamkap’s Snowflake connector supports schema evolution — automatically adding new columns to destination tables when the source schema changes. For this to work, the Snowflake role used by the connector must have the correct privileges on the target tables.

OWNERSHIP privilege required for schema evolutionSchema evolution requires the OWNERSHIP privilege on the target tables — not just ALTER. When Snowpipe Streaming detects new columns in incoming data, it must alter the destination table, which Snowflake restricts to the table owner.If schema evolution fails with an Insufficient privileges error, grant OWNERSHIP on the affected tables (or future tables in the schema) to the Streamkap role:

-- Grant ownership on all existing tables in the schema
GRANT OWNERSHIP ON ALL TABLES IN SCHEMA <database>.<schema>
  TO ROLE STREAMKAP_ROLE COPY CURRENT GRANTS;

-- Grant ownership on future tables so new tables are also covered
GRANT OWNERSHIP ON FUTURE TABLES IN SCHEMA <database>.<schema>
  TO ROLE STREAMKAP_ROLE COPY CURRENT GRANTS;

If you do not want to grant OWNERSHIP, you can pre-create your destination tables with the expected columns and manage schema changes manually.

Ingestion Modes - Understand insert vs. upsert ingestion modes
Delete Handling - Configure how delete events are processed at the destination

Getting Started

App

Troubleshooting

Deployment

Sources

Destinations

Transformation

Usage

Security

Developer Tools

Terraform

Prerequisites

Snowflake Setup

Key Pair Authentication

Streamkap Setup

1. Create the Destination

2. Connection Settings

3. Ingestion Settings

Troubleshooting

Dynamic Tables

Offset Management (Append Mode)

Understanding Dual Offset Tracking

Replaying Messages to Snowflake

Offset Reset Strategies

Upsert mode

Getting the Snowflake URL

Snowflake Setup scripts failing

Troubleshooting Common Issues

Schema Evolution Permissions

Getting Started

App

Troubleshooting

Deployment

Sources

Destinations

Transformation

Usage

Security

Developer Tools

Terraform

​Prerequisites

​Snowflake Setup

​Key Pair Authentication

​Streamkap Setup

​1. Create the Destination

​2. Connection Settings

​3. Ingestion Settings

​Troubleshooting

​Dynamic Tables

​Offset Management (Append Mode)

​Understanding Dual Offset Tracking

​Replaying Messages to Snowflake

​Offset Reset Strategies

​Upsert mode

​Getting the Snowflake URL

​Snowflake Setup scripts failing

​Troubleshooting Common Issues

​Schema Evolution Permissions

​Related Documentation

Prerequisites

Snowflake Setup

Key Pair Authentication

Streamkap Setup

1. Create the Destination

2. Connection Settings

3. Ingestion Settings

Troubleshooting

Dynamic Tables

Offset Management (Append Mode)

Understanding Dual Offset Tracking

Replaying Messages to Snowflake

Offset Reset Strategies

Upsert mode

Getting the Snowflake URL

Snowflake Setup scripts failing

Troubleshooting Common Issues

Schema Evolution Permissions

Related Documentation