Snowflake Streams

SDC 7.7 and later

The Snowflake Streams source processes changed (CDC) data from a Snowflake stream for a standard Snowflake table. The Snowflake Streams source generates a record for each change record that it receives from the specified stream. For information about supported versions, see Supported systems and versions.

You might use this source to perform database replication. You can use a separate flow with the Snowflake Bulk source to read existing data. Then start a flow with the Snowflake Streams source to process subsequent changes.

When you configure the source, you define Snowflake connection information. You can specify a Snowflake private link to use. You can also use a connection to configure the source.

The Snowflake Streams source creates transient tables while processing data from the Snowflake stream. You can specify a table name prefix for the transient tables. You can also configure the maximum amount of time that the transient tables are accessible after processing.

The source includes the CRUD operation type in a record header attribute so generated records can be easily processed by CRUD-enabled targets. For an overview of Data Collector changed data processing and a list of CRUD-enabled targets, see Processing changed data.

For more information about Snowflake streams, see the Snowflake documentation.

Snowflake prerequisites

Before you use the Snowflake Streams source, complete the following prerequisite tasks:
Create the stream
Create the Snowflake stream for the standard table that you want to process.
Assign privileges

The source requires a Snowflake role that grants all privileges that are needed to process data.

The Snowflake role must include the following privileges:
Object Privilege
Warehouse USAGE
Database USAGE
Schema USAGE
Stream SELECT
Schema for transient tables CREATE TABLE, CREATE TEMPORARY TABLE

Retaining access to transient tables

You can configure how long you can use Snowflake to access data from transient tables after the Snowflake Streams source processes data from those tables.

The Snowflake Streams source stores data in transient tables for processing and removes the tables after it completes processing that data. You can use the Snowflake UNDROP TABLE command to access data in the removed transient tables for a configurable period of time.

By default, data in removed transient tables is accessible for one day. You can increase the retention period to up to 90 days. You might increase the time to allow more time for disaster recovery. Retaining data for an extended time period can increase storage costs.

To set the data retention time, configure the Data retention time (days) property on the Stream tab of the source.

Record header attributes

The Snowflake Streams source includes the following record header attributes in generated records:
CRUD operation header attribute
The source evaluates the operation type associated with each entry that it processes. When appropriate, it writes the operation type to the sdc.operation.type record header attribute.
The source uses the following values in the attribute to represent the operation type:
  • 1 for INSERT
  • 2 for DELETE
When an update occurs, the Snowflake stream provides a pair of delete and insert records, which the source processes as they are received.
If you use a CRUD-enabled target in the flow, such as the Snowflake target, the target can use the operation type to write to the target system. When necessary, you can use an Expression Evaluator processor or any scripting processor to manipulate the value in the header attribute. For an overview of Data Collector change data processing and a list of CRUD-enabled targets, see Processing changed data.
CDC and additional attributes
The source also provides the following CDC record header attributes for each record:
  • snowflake.stream.action - CRUD operation type from Snowflake, such as INSERT or DELETE.
  • snowflake.stream.is_update - Indicator for update operations: true for update operations and false for other operations.
  • snowflake.stream.row_id - Unique row identifier assigned by Snowflake.
  • snowflake.db - Snowflake database where the source table resides.
  • snowflake.schema - Snowflake schema where the source table resides.
  • snowflake.table - Name of the transient table used for the record.
  • snowflake.stream - Name of the originating stream.
  • jdbc.primaryKey.specification - JSON specification of primary key columns.
  • jdbc.<columnName>.precision - Original precision of the specified column.
  • jdbc.<columnName>.scale - Original scale of the specified column, when applicable.

You can use the record:attribute or record:attributeOrDefault functions to access the information in the attributes. For more information about working with record header attributes, see Working with header attributes.

Snowflake data types

The Snowflake Streams source converts Snowflake data types to Data Collector data types. The source supports the following data types:

Snowflake Data Type Data Collector Data Type
Array List
Bigint Long
Binary Byte Array
Boolean Boolean
Byteint Long
Char String
Date Date
Datetime Datetime
Decimal Long
Double Double
Double Precision Double
Float Double
Geography String
Int Long
Integer Long
Number Long

Number(<precision>, 0)

 Long

Number(<precision>, <scale>)

 Long
Nchar String
Nvarchar2 String
Real Double
Smallint Long
Time Time
Timestamp_LZT Zoned Datetime
Timestamp_NTZ Datetime
Timestamp_TZ Zoned Datetime
Tinyint Long
Varchar String
Variant Conversion occurs based on content

Configuring a Snowflake Streams source

About this task

Configure a Snowflake Streams source to process changed data from a Snowflake stream. Before you use the source, complete the Snowflake prerequisites.

Procedure

  1. In the Properties panel, on the General tab configure the general stage properties.
    For more information, see the following topic:
  2. On the Snowflake tab, select a Snowflake connection to provide connection information or configure connection details such as the organization, account, and authentication method.
  3. On the Streams tab, configure the main properties for the source, such as the Snowflake stream and the data retention time to use.