Snowpipe Streaming¶

Snowpipe Streaming API vs Snowpipe¶

The API is intended to complement Snowpipe, not replace it. Use the Snowpipe Streaming API in streaming scenarios where data is streamed via rows (e.g. Apache Kafka topics) rather than written to files. The API fits into an ingest workflow that includes an existing custom Java application that produces or receives records. The API removes the need to create files to load data into Snowflake tables, and enables the automatic, continuous loading of data streams into Snowflake as the data becomes available.

The following table describes the differences between Snowpipe Streaming and Snowpipe:

Software Requirements¶

Channels¶

The API ingests rows via one or more channels. A channel represents a logical, named streaming connection to Snowflake for loading data into a table. A single channel maps to exactly one table in Snowflake; however, multiple channels can point to the same table. The Client SDK has the ability to open multiple channels to multiple tables, however the SDK cannot open channels across accounts. The ordering of rows and their corresponding offset tokens are preserved within a channel, but not across channels that point to the same table.

You can run SHOW CHANNELS command to list the channels for which you have access privileges. For more information, refer to SHOW CHANNELS.

Offset Tokens¶

An offset token is a string that a client can include in insertRow (single row) or insertRows (set of rows) method requests to track ingestion progress on a per-channel basis. The token is initialized to NULL on channel creation and is updated when the rows with a provided offset token are committed to Snowflake through an asynchronous process. Clients can periodically make getLatestCommittedOffsetToken method requests to get the latest committed offset token for a particular channel and use that to reason about ingestion progress. Note that this token is not used by Snowflake to perform de-duplication; however, clients can use this token to perform de-duplication using your custom code.

When a client re-opens a channel, the latest persisted offset token is returned. The client can reset its position in the data source using the token to avoid sending the same data twice. Note that when a channel re-open event occurs, any data buffered in Snowflake is discarded to avoid committing it.

You can use the latest committed offset token to perform the following common use cases:

• Tracking the ingestion progress.

• Checking whether a specific offset has been committed or not by comparing it with the latest committed offset token.

• Advancing the source offset and purging the data that has already been committed.

• Enabling de-duplication and ensuring exactly-once delivery of data.

For example, the Kafka connector could read an offset token from the topic such as <partition>:<offset>, or simply <offset> if the partition is encoded in the channel name. Consider the following scenario:

1. The Kafka connector comes online and opens a channel corresponding to Partition 1 in Kafka topic T with the channel name of T:P1.

2. The connector begins reading records from the Kafka partition. The connector calls the API, making an insertRows method request with the offset associated with the record as the offset token. For example, the offset token could be 10, referring to the 10th record in the Kafka partition.

3. The connector periodically makes getLatestCommittedOffsetToken method requests to determine the ingest progress.

If the Kafka connector crashes, then the following flow could be completed to resume reading records from the correct offset for the Kafka partition:

1. The Kafka connector comes back online and re-opens the channel using the same name as earlier.

2. The connector calls the API, making a getLatestCommittedOffsetToken method request to get the latest committed offset for the partition. For example, assume the latest persisted offset token is 20.

3. The connector uses the Kafka read APIs to reset a cursor corresponding to the offset plus 1: 21 in this example.

4. The connector resumes reading records. No duplicate data is retrieved after the read cursor is repositioned successfully.

As another example, suppose you have an application that reads logs from a directory and exports those logs to Snowflake using the Snowpipe Streaming Client SDK. You could build a log export application that does the following:

1. List files in the log directory. Assume the logging framework generates log files that can be ordered lexicographically and that new log files are positioned at the end of this ordering.

2. Reads a log file line by line and calls the API, making insertRows method requests with an offset token corresponding to the log file name and the line count or byte position. For instance, an offset token could be messages_1.log:20, where messages_1.log is the name of the log file and 20 is the line number.

If the application crashes or needs to be restarted, it would then call the API, making a getLatestCommittedOffsetToken method request to retrieve an offset token corresponding to the last exported log file and line. Continuing with the example, this could be messages_1.log:20. The application would then open messages_1.log and seek line 21 to prevent the same log line from being ingested twice.

Exactly-Once Delivery Best Practices¶

Achieving exactly-once delivery can be challenging, and adherence to the following principles in your custom code is critical:

• To ensure appropriate recovery from exceptions, failures, or crashes, you must always reopen the channel and restart ingestion using the latest committed offset token.

• While your application may maintain its own offset, it’s crucial to use the latest committed offset token provided by Snowflake as the source of truth and reset your own offset accordingly.

• The only instance in which your own offset should be treated as the source of truth is when the offset token from Snowflake is set or reset to NULL. A NULL offset token usually means one of the following:

  • This is a new channel, so no offset token has been set.

  • The target table is dropped and recreated, so the channel is considered new.

  • If there is no ingestion activity through a channel for 30 days, the channel is automatically cleaned up, which results in the loss of offset token information.

• If necessary, you can periodically purge the source data that has already been committed based on the latest committed offset token and advance your own offset.

For more information about how the Kafka connector with Snowpipe Streaming achieves exactly-once delivery, refer to Exactly-Once Semantics.

Migration to Optimized Files¶

The API writes the rows from channels into blobs in cloud storage, which are then committed to the target table. Initially, the streamed data written to a target table is stored in a temporary intermediate file format. At this stage, the table is considered a “mixed table,” because partitioned data is stored in a mixture of native and intermediary files. An automated, background process migrates data from the active intermediate files to native files optimized for query and DML operations as needed.

Replication¶

Snowpipe streaming supports the replication and failover of Snowflake tables populated by Snowpipe Streaming and its associated channel offsets from a source account to a target account in different regions and across cloud platforms.

For more information, see Replication and Snowpipe Streaming.

Insert-only Operations¶

The API is currently limited to inserting rows. To modify, delete, or combine data, write the “raw” records to one or more staging tables. Merge, join, or transform the data using continuous data pipelines to insert modified data into destination reporting tables.

Classes and Interfaces¶

For documentation on the classes and interfaces see Snowflake Ingest SDK API.

Supported Java Data Types¶

The following table summarizes which Java data types are supported for ingestion into Snowflake columns:

Required Access Privileges¶

Calling the Snowpipe Streaming API requires a role with the following privileges:

Limitations¶

• Tables with any one of the following column settings are not supported:

  • AUTOINCREMENT or IDENTITY

  • Default column value that is not NULL.

• The GEOGRAPHY and GEOMETRY data types are not supported.

• Columns with collations are not supported.

• Snowpipe Streaming only supports using 256-bit AES keys for data encryption.

• TRANSIENT or TEMPORARY tables are not supported.