Streams for change data capture in Snowflake - Commands & Configuration

Introduction

When you want to track changes like inserts, updates, and deletes in a table automatically, streams help by capturing these changes so you can use them later without changing your original table.

When you want to keep a log of all changes made to a sales data table for auditing.

When you need to update a reporting table only with new or changed records from a source table.

When you want to build a data pipeline that reacts to changes in your main data without scanning the entire table.

When you want to synchronize changes from one table to another in near real-time.

When you want to implement incremental data processing to save time and resources.

Config File - create_stream.sql

create_stream.sql

CREATE OR REPLACE TABLE sales_data (
  id INT,
  product STRING,
  quantity INT,
  price FLOAT,
  updated_at TIMESTAMP
);

CREATE OR REPLACE STREAM sales_data_stream ON TABLE sales_data;

This SQL file first creates a table named sales_data with columns for id, product, quantity, price, and update time.

Then it creates a stream called sales_data_stream on the sales_data table. This stream will track all changes made to the table after the stream is created.

Commands

This command inserts a new record into the sales_data table to create a change that the stream will capture.

Terminal

snowsql -q "INSERT INTO sales_data (id, product, quantity, price, updated_at) VALUES (1, 'Widget', 10, 9.99, CURRENT_TIMESTAMP)"

Expected Output

1 row inserted.

This command queries the stream to see the changes captured since the stream was created or last consumed.

Terminal

snowsql -q "SELECT * FROM sales_data_stream"

Expected Output

ID | PRODUCT | QUANTITY | PRICE | UPDATED_AT | METADATA$ACTION

1  | Widget  | 10       | 9.99  | 2024-06-01 12:00:00 | INSERT

This command deletes the record from the sales_data table, creating another change for the stream to capture.

Terminal

snowsql -q "DELETE FROM sales_data WHERE id = 1"

Expected Output

1 row deleted.

Query the stream again to see the new delete action captured by the stream.

Terminal

snowsql -q "SELECT * FROM sales_data_stream"

Expected Output

ID | PRODUCT | QUANTITY | PRICE | UPDATED_AT | METADATA$ACTION

1  | Widget  | 10       | 9.99  | 2024-06-01 12:00:00 | DELETE

Key Concept

If you remember nothing else from this pattern, remember: a stream in Snowflake automatically tracks all changes to a table so you can process only what changed.

Common Mistakes

Querying the stream without making any changes to the base table first.

The stream will be empty because it only shows changes made after its creation.

Make sure to insert, update, or delete rows in the base table before querying the stream.

Not consuming or querying the stream regularly.

Streams keep track of changes until consumed; if ignored, they can grow large and cause confusion.

Query the stream regularly to process and clear the captured changes.

Creating a stream on a table that does not exist or has no data.

The stream will not capture any meaningful changes if the table is missing or empty.

Create the base table first and ensure it has data or will receive data changes.

Summary

Create a stream on a table to automatically capture inserts, updates, and deletes.

Make changes to the base table to generate change data the stream can capture.

Query the stream to see and process only the changed rows since the last query.