Streams and Tasks: Automate Real-Time Data Pipelines

In modern data platforms, real-time data processing is critical for timely insights. Snowflake provides two powerful features—streams and tasks—that enable automated, incremental data pipelines without external tools. Streams track changes (inserts, updates, deletes) on a table, while tasks execute SQL statements on a schedule or based on dependencies. Together, they allow you to build event-driven workflows that process data as it arrives, reducing latency and manual effort.

Imagine you have a raw sales table that receives thousands of new records every minute. You need to transform and load that data into a reporting table. Without automation, you'd run periodic batch jobs, which could miss recent changes or overload the system. With streams and tasks, you can capture only the new or changed rows and process them immediately, keeping your reporting table up-to-date with minimal overhead.

This tutorial will guide you through creating streams and tasks, writing efficient DML statements, handling common pitfalls, and debugging production issues. By the end, you'll be able to build robust, automated pipelines that scale with your data.

What are Snowflake Streams?

A Snowflake stream is an object that records data manipulation language (DML) changes made to a table, including inserts, updates, and deletes. It provides a change data capture (CDC) mechanism that tracks row-level changes with before and after images. Streams are append-only and store the change records for a configurable retention period (default 14 days).

When you create a stream on a table, it captures all DML changes that occur after the stream is created. You can query the stream like a table to see the changes. Each row in the stream includes metadata columns: METADATA$ACTION (INSERT, DELETE, UPDATE), METADATA$ISUPDATE (true for updates), and METADATA$ROW_ID.

Streams are ideal for incremental processing because they only show new or changed rows since the last time the stream was consumed. Consumption happens when you execute a DML statement that reads from the stream (e.g., INSERT INTO target SELECT ... FROM stream). After consumption, the stream offset advances, and the next query will show only subsequent changes.

-- Create a stream on the raw_sales table
CREATE STREAM raw_sales_stream ON TABLE raw_sales;

-- Query the stream to see changes
SELECT * FROM raw_sales_stream;

-- Check the stream offset timestamp
SELECT SYSTEM$STREAM_GET_TABLE_TIMESTAMP('raw_sales_stream');

What are Snowflake Tasks?

A Snowflake task is a scheduled object that executes a single SQL statement or a call to a stored procedure. Tasks can be used to automate data transformation, loading, and other maintenance operations. They support scheduling using cron expressions or simple intervals (e.g., '5 MINUTE'). Tasks can also be chained together in a DAG (Directed Acyclic Graph) using task dependencies.

When you create a task, you define its schedule and the SQL to execute. The task runs as the role that owns it (or the role specified with EXECUTE AS). You can also set a task to run after another task completes, enabling complex workflows.

Tasks are commonly used with streams to create automated pipelines: a task runs on a schedule, reads from a stream, and performs DML to consume the stream and load data into target tables.

-- Create a task that runs every 5 minutes
CREATE TASK process_sales_task
  WAREHOUSE = my_wh
  SCHEDULE = '5 MINUTE'
AS
  INSERT INTO sales_report (sale_id, amount, sale_date)
  SELECT sale_id, amount, sale_date
  FROM raw_sales_stream;

-- Resume the task to start execution
ALTER TASK process_sales_task RESUME;

Combining Streams and Tasks for Incremental Processing

The real power of Snowflake comes from combining streams and tasks. A typical pattern is: 1. Create a stream on a source table. 2. Create a task that runs on a schedule (e.g., every minute). 3. The task executes a DML statement that reads from the stream and writes to a target table. 4. The DML consumes the stream, advancing its offset.

This pattern enables near-real-time data pipelines without external orchestration. For example, you can process sales data as it arrives, transform it, and load it into a reporting table.

To handle updates and deletes, use MERGE instead of INSERT. MERGE can insert new rows, update existing ones, and delete rows based on the stream's METADATA$ACTION.

Example: A task that runs every minute to process changes from raw_sales_stream into sales_report, handling all three DML operations.

-- Create a task that uses MERGE to handle all changes
CREATE TASK merge_sales_task
  WAREHOUSE = my_wh
  SCHEDULE = '1 MINUTE'
AS
  MERGE INTO sales_report t
  USING (
    SELECT 
      sale_id,
      amount,
      sale_date,
      METADATA$ACTION AS action,
      METADATA$ISUPDATE AS is_update
    FROM raw_sales_stream
  ) s ON t.sale_id = s.sale_id
  WHEN MATCHED AND s.action = 'DELETE' THEN DELETE
  WHEN MATCHED AND s.is_update = 'TRUE' THEN UPDATE SET t.amount = s.amount, t.sale_date = s.sale_date
  WHEN NOT MATCHED AND s.action = 'INSERT' THEN INSERT (sale_id, amount, sale_date) VALUES (s.sale_id, s.amount, s.sale_date);

ALTER TASK merge_sales_task RESUME;

Task Dependencies and DAGs

Snowflake tasks can be organized into a DAG (Directed Acyclic Graph) where one task runs after another. This allows you to build multi-step pipelines. For example, Task A loads raw data, Task B transforms it, and Task C loads the final report.

To create a dependency, use the AFTER keyword when creating a task. The dependent task will start automatically after the predecessor task completes successfully. You can also specify conditions using WHEN to control execution.

DAGs are useful for complex workflows where each step depends on the previous one. However, they can be harder to debug because failures cascade.

Example: Two tasks where the second runs after the first.

-- Create first task
CREATE TASK load_raw_task
  WAREHOUSE = my_wh
  SCHEDULE = '5 MINUTE'
AS
  INSERT INTO raw_sales SELECT * FROM staging_table;

-- Create second task that runs after first
CREATE TASK transform_task
  WAREHOUSE = my_wh
  AFTER load_raw_task
AS
  MERGE INTO sales_report ...;

-- Resume both tasks
ALTER TASK load_raw_task RESUME;
ALTER TASK transform_task RESUME;

Monitoring and Debugging Streams and Tasks

Monitoring is crucial for production pipelines. Snowflake provides several views and functions to track task execution and stream status.

• TASK_HISTORY: Shows task runs, including start time, end time, state (SUCCEEDED, FAILED, SKIPPED), and error messages.
• STREAM_HISTORY: Shows stream consumption events.
• SYSTEM$STREAM_GET_TABLE_TIMESTAMP: Returns the timestamp of the last change captured by the stream. If NULL, the stream has no data.
• SYSTEM$STREAM_HAS_DATA: Returns TRUE if the stream has unprocessed changes.

You can also query the stream directly to see pending changes.

Common debugging steps: 1. Check if the stream has data: SELECT SYSTEM$STREAM_HAS_DATA('stream_name'); 2. Check task history: SELECT * FROM TABLE(INFORMATION_SCHEMA.TASK_HISTORY()) WHERE TASK_NAME = 'my_task' ORDER BY SCHEDULED_TIME DESC; 3. If task fails, examine the error message in TASK_HISTORY. 4. If stream is stale, recreate it.

-- Check if stream has data
SELECT SYSTEM$STREAM_HAS_DATA('raw_sales_stream');

-- View recent task history
SELECT *
FROM TABLE(INFORMATION_SCHEMA.TASK_HISTORY())
WHERE TASK_NAME = 'MERGE_SALES_TASK'
ORDER BY SCHEDULED_TIME DESC
LIMIT 10;

-- Check stream offset timestamp
SELECT SYSTEM$STREAM_GET_TABLE_TIMESTAMP('raw_sales_stream');

Best Practices for Production Pipelines

Building robust pipelines requires following best practices:

1. Use MERGE for idempotency: If a task runs multiple times (e.g., due to retries), MERGE ensures no duplicates.
2. Set appropriate schedule: Choose a schedule that balances latency and cost. For near-real-time, use 1 minute; for batch, use longer intervals.
3. Handle stream staleness: If a stream becomes stale, recreate it and backfill missing data.
4. Use conditional tasks: Add a WHEN clause to skip execution if the stream has no data, saving warehouse credits.
5. Monitor costs: Tasks consume warehouse credits. Use auto-suspend and appropriate warehouse size.
6. Test in a development environment: Before deploying to production, test the pipeline with sample data.
7. Document your pipeline: Keep track of stream and task definitions, dependencies, and schedules.

Example of a conditional task:

CREATE TASK conditional_merge_task
  WAREHOUSE = my_wh
  SCHEDULE = '1 MINUTE'
  WHEN SYSTEM$STREAM_HAS_DATA('raw_sales_stream')
AS
  MERGE INTO sales_report ...;

ALTER TASK conditional_merge_task RESUME;