Streams for change data capture in Snowflake - Mini Project: Build & Apply

Streams for change data capture

📖 Scenario: You work as a cloud data engineer. Your team wants to track changes in a customer orders table in Snowflake. They want to capture inserts, updates, and deletes to analyze order trends over time.

🎯 Goal: Create a Snowflake stream on the orders table to capture all changes (inserts, updates, deletes). This stream will help downstream processes detect data changes efficiently.

📋 What You'll Learn

Create a table called orders with columns order_id (integer), customer_id (integer), and order_status (string).

Create a stream called orders_stream on the orders table to capture all change data.

Use the stream to select all changed rows with metadata columns.

Add a final query to consume the stream data.

💡 Why This Matters

🌍 Real World

Streams in Snowflake help track data changes efficiently without scanning entire tables. This is useful for auditing, incremental data pipelines, and real-time analytics.

💼 Career

Data engineers and cloud architects use streams to build scalable, event-driven data workflows and maintain data consistency across systems.

1

Create the orders table

Create a table called orders with columns order_id as INTEGER, customer_id as INTEGER, and order_status as STRING.

Snowflake

# Your code here to create the orders table

Need a hint?

Use CREATE OR REPLACE TABLE orders with the specified columns and types.

2

Create a stream on the orders table

Create a stream called orders_stream on the orders table to capture all changes including inserts, updates, and deletes.

Snowflake

CREATE OR REPLACE TABLE orders (
    order_id INTEGER,
    customer_id INTEGER,
    order_status STRING
);

# Your code here to create the orders_stream stream

Need a hint?

Use CREATE OR REPLACE STREAM orders_stream ON TABLE orders; to create the stream.

3

Select changes from the orders_stream

Write a SELECT query to retrieve all columns and metadata columns from the orders_stream stream.

Snowflake

CREATE OR REPLACE TABLE orders (
    order_id INTEGER,
    customer_id INTEGER,
    order_status STRING
);

CREATE OR REPLACE STREAM orders_stream ON TABLE orders;

# Your code here to select all columns and metadata from orders_stream

Need a hint?

Use SELECT *, METADATA$ACTION, METADATA$ISUPDATE FROM orders_stream; to get all changes with metadata.

4

Consume the stream data

Write a query to insert all changed rows from orders_stream into a new table called orders_changes. Create the orders_changes table first with the same columns as orders plus change_type as STRING.

Snowflake

CREATE OR REPLACE TABLE orders (
    order_id INTEGER,
    customer_id INTEGER,
    order_status STRING
);

CREATE OR REPLACE STREAM orders_stream ON TABLE orders;

SELECT *, METADATA$ACTION, METADATA$ISUPDATE FROM orders_stream;

# Your code here to create orders_changes table and insert changes from orders_stream

Need a hint?

Create orders_changes table with a change_type column, then insert changes from orders_stream using METADATA$ACTION.