Snowflake Streams & Tasks

You are a Snowflake data engineer. The user wants to set up change data capture (CDC) with Snowflake Streams to track table changes and automate processing with scheduled Tasks.

What to check first

• Run SHOW STREAMS; in your target schema to see existing streams
• Verify the source table exists: DESCRIBE TABLE <table_name>;
• Check your role has CREATE STREAM and CREATE TASK privileges: SHOW GRANTS TO ROLE <role_name>;

Steps

1. Create a standard stream on your source table with CREATE STREAM <stream_name> ON TABLE <table_name>; — this captures INSERT, UPDATE, DELETE changes
2. Query the stream with SELECT * FROM <stream_name>; to inspect the system columns METADATA$ACTION, METADATA$ISUPDATE, and METADATA$ROW_NUMBER
3. Create a target table or staging table that will receive processed changes using CREATE TABLE <target_table> AS SELECT * FROM <source_table> WHERE 1=0;
4. Write a stored procedure that consumes the stream, processes changes, and calls TRUNCATE <stream_name>; to mark changes as consumed — use CREATE OR REPLACE PROCEDURE process_changes() RETURNS STRING LANGUAGE SQL AS $$...$$;
5. Create a scheduled task with CREATE TASK <task_name> WAREHOUSE = <warehouse_name> SCHEDULE = '5 MINUTE' AS CALL process_changes();
6. Resume the task with ALTER TASK <task_name> RESUME; — tasks are created in suspended state
7. Monitor task execution with SELECT * FROM TABLE(INFORMATION_SCHEMA.TASK_HISTORY(TASK_NAME='<task_name>')) ORDER BY QUERY_START_TIME DESC;
8. Set task dependencies if multiple tasks chain together using CREATE TASK child_task AFTER parent_task AS...

Code

-- 1. Create a stream on the source table
CREATE STREAM sales_changes ON TABLE sales_data
  APPEND_ONLY = FALSE;

-- 2. Create a staging table for processed records
CREATE TABLE sales_processed (
  id INT,
  customer_id INT,
  amount DECIMAL(10,2),
  sale_date DATE,
  dml_type VARCHAR(10),
  processed_at TIMESTAMP_NTZ
);

-- 3. Create a stored procedure to consume the stream
CREATE OR REPLACE PROCEDURE consume_sales_stream()
RETURNS STRING
LANGUAGE SQL
AS
$$
DECLARE
  change_count INTEGER DEFAULT 0;
BEGIN
  -- Insert new records from stream
  INSERT INTO sales_processed (id, customer_id, amount, sale_date, dml_type, processed_at)
  SELECT 
    id,
    customer_id,
    amount,
    sale_date,
    METADATA$ACTION AS dml_type,
    CURRENT_TIMESTAMP() AS processed_at
  FROM sales

Note: this example was truncated in the source. See the GitHub repo for the latest full version.

Common Pitfalls

• Treating this skill as a one-shot solution — most workflows need iteration and verification
• Skipping the verification steps — you don't know it worked until you measure
• Applying this skill without understanding the underlying problem — read the related docs first

When NOT to Use This Skill

• When a simpler manual approach would take less than 10 minutes
• On critical production systems without testing in staging first
• When you don't have permission or authorization to make these changes

How to Verify It Worked

• Run the verification steps documented above
• Compare the output against your expected baseline
• Check logs for any warnings or errors — silent failures are the worst kind

Production Considerations

• Test in staging before deploying to production
• Have a rollback plan — every change should be reversible
• Monitor the affected systems for at least 24 hours after the change