Streams & Tasks

Snowflake Streams and Tasks are the backbone of automated, incremental data pipelines. Streams implement Change Data Capture (CDC), recording exactly what changed in a table. Tasks schedule SQL or Snowpark logic to process those changes on a defined cadence. Together they form a serverless ELT pipeline inside Snowflake — no external orchestrators required.

1. What is a Stream?

A Snowflake Stream is a CDC object attached to a source table (or external table, or certain views). It does not store a copy of the data — it stores an offset: a pointer to the last transaction that was consumed. Every time you query the stream inside a DML statement, Snowflake computes the delta between the stored offset and the current state of the table, returns those changed rows, and then advances the offset.

This design means streams are lightweight: no data is duplicated, and the cost of maintaining a stream is negligible. What you pay for is the compute used when you actually read the stream.

2. Stream Mechanics — The Offset Model

When you create a stream on a table, Snowflake records the current transaction timestamp as the initial offset. From that point forward, any DML applied to the source table is tracked. The stream becomes a view over those changes.

Source Table: ORDERS
    ↓  (INSERT row 101)
    ↓  (UPDATE row 99)
    ↓  (DELETE row 55)

Stream: ORDERS_STREAM
    ├── offset = T0  ← stream created here
    └── delta since T0:
          ROW 101  │ METADATA$ACTION = INSERT │ METADATA$ISUPDATE = FALSE
          ROW 99   │ METADATA$ACTION = DELETE │ METADATA$ISUPDATE = TRUE   (old value)
          ROW 99   │ METADATA$ACTION = INSERT │ METADATA$ISUPDATE = TRUE   (new value)
          ROW 55   │ METADATA$ACTION = DELETE │ METADATA$ISUPDATE = FALSE

Reading the stream inside an INSERT…SELECT advances the offset to the current moment, clearing the visible delta for the next read.

Stream Offset Lifecycle

A stream is created at T0. Changes accumulate. A DML read at T1 consumes the delta and advances the offset. New changes accumulate from T1 onwards.

3. Stream Types

Snowflake offers three stream types, each suited to different source objects and workloads.

Insert-Only Streams are a third type, used exclusively on External Tables. Because external files can only be added (not modified via Snowflake DML), the insert-only stream captures newly loaded files registered by Snowpipe or directory listing refresh.

1-- Standard stream (default) — tracks INSERT, UPDATE, DELETE
2CREATE OR REPLACE STREAM orders_stream
3ON TABLE orders;
4
5-- Append-only stream — tracks INSERT only (efficient for logs)
6CREATE OR REPLACE STREAM events_stream
7ON TABLE web_events
8APPEND_ONLY = TRUE;
9
10-- Insert-only stream — for external tables only
11CREATE OR REPLACE STREAM ext_stream
12ON EXTERNAL TABLE ext_orders
13INSERT_ONLY = TRUE;
14
15-- Verify stream properties
16SHOW STREAMS LIKE 'orders_stream';
17-- Key columns: name, type, stale, stale_after, source_name
18
19-- Describe a specific stream
20DESCRIBE STREAM orders_stream;

4. Stream Metadata Columns

Every row returned by a stream query includes three system-generated metadata columns that are critical for the exam:

How to read UPDATE rows: An UPDATE generates two rows in a standard stream:

1. A DELETE row with the old column values — METADATA$ISUPDATE = TRUE
2. An INSERT row with the new column values — METADATA$ISUPDATE = TRUE

To isolate only net-new inserts that are not part of an update:

SELECT * FROM orders_stream
WHERE METADATA$ACTION = 'INSERT'
  AND METADATA$ISUPDATE = FALSE;

Metadata Columns for an UPDATE Operation

When a single row is updated in the source table, the standard stream produces two rows: a DELETE of the old value (METADATA$ISUPDATE=TRUE) and an INSERT of the new value (METADATA$ISUPDATE=TRUE). Plain INSERTs and DELETEs produce one row each with METADATA$ISUPDATE=FALSE.

5. Consuming a Stream

Consuming a stream means reading it inside a DML statement. This is the only way to advance the stream offset.

1-- Consume stream: insert new orders into a target table
2INSERT INTO orders_processed
3SELECT
4  order_id,
5  customer_id,
6  amount,
7  CURRENT_TIMESTAMP() AS processed_at
8FROM orders_stream
9WHERE METADATA$ACTION = 'INSERT'
10AND METADATA$ISUPDATE = FALSE;
11
12-- MERGE pattern: handle inserts, updates, and deletes in one statement
13MERGE INTO orders_target AS t
14USING (
15  SELECT *,
16         METADATA$ACTION,
17         METADATA$ISUPDATE
18  FROM orders_stream
19) AS s
20ON t.order_id = s.order_id
21WHEN MATCHED AND s.METADATA$ACTION = 'DELETE' AND s.METADATA$ISUPDATE = FALSE THEN
22  DELETE
23WHEN MATCHED AND s.METADATA$ACTION = 'INSERT' AND s.METADATA$ISUPDATE = TRUE THEN
24  UPDATE SET t.amount = s.amount, t.status = s.status
25WHEN NOT MATCHED AND s.METADATA$ACTION = 'INSERT' THEN
26  INSERT (order_id, customer_id, amount, status)
27  VALUES (s.order_id, s.customer_id, s.amount, s.status);

6. Stream Staleness

A stream becomes stale if it is not consumed within the source table’s Data Retention Period (default 1 day on Standard, up to 90 days on Enterprise). Snowflake cannot guarantee the accuracy of the delta once the underlying micro-partition data required to compute it has been cleaned up by Fail-Safe expiry.

• SHOW STREAMS exposes the STALE (boolean) and STALE_AFTER (timestamp) columns.
• A stale stream must be dropped and recreated — you cannot “un-stale” it.
• To avoid staleness, ensure your Tasks run more frequently than the retention window.

-- Check for stale streams
SHOW STREAMS;

-- Identify if a specific stream is stale
SELECT SYSTEM$STREAM_HAS_DATA('orders_stream');
-- Returns TRUE if unconsumed data exists; FALSE if empty (or stale)

7. Tasks — Scheduled Automation Inside Snowflake

A Task is a Snowflake object that executes a single SQL statement (or a call to a Stored Procedure / Snowpark function) on a user-defined schedule. Think of it as a cron job that lives inside your Snowflake account — no external scheduler, no Airflow connection required.

Task Schedule Options

Tasks support two schedule formats:

1. Interval-based: SCHEDULE = '5 MINUTE' — runs every N minutes (1 to 11,520 minutes / 8 days max).
2. Cron-based: SCHEDULE = 'USING CRON 0 9 * * MON-FRI UTC' — full cron expression. The timezone identifier is mandatory (e.g., UTC, Europe/London).

1-- Basic task: virtual warehouse, interval schedule
2CREATE OR REPLACE TASK process_new_orders
3WAREHOUSE = COMPUTE_WH
4SCHEDULE = '5 MINUTE'
5AS
6INSERT INTO orders_processed
7SELECT order_id, customer_id, amount
8FROM orders_stream
9WHERE METADATA$ACTION = 'INSERT';
10
11-- Task with cron schedule (weekdays at 09:00 UTC)
12CREATE OR REPLACE TASK daily_report_task
13WAREHOUSE = REPORTING_WH
14SCHEDULE = 'USING CRON 0 9 * * MON-FRI UTC'
15AS
16CALL generate_daily_report();
17
18-- Serverless task (Snowflake manages compute)
19CREATE OR REPLACE TASK serverless_ingest
20USER_TASK_MANAGED_INITIAL_WAREHOUSE_SIZE = 'XSMALL'
21SCHEDULE = '10 MINUTE'
22AS
23INSERT INTO target SELECT * FROM orders_stream WHERE METADATA$ACTION = 'INSERT';
24
25-- MUST resume the task before it runs
26ALTER TASK process_new_orders RESUME;
27ALTER TASK daily_report_task RESUME;
28ALTER TASK serverless_ingest RESUME;
29
30-- Suspend a task
31ALTER TASK process_new_orders SUSPEND;

8. Task Trees — DAGs

Snowflake tasks can be chained into a Directed Acyclic Graph (DAG). The root task carries the schedule; child tasks carry an AFTER clause naming their parent. No task can appear in a cycle.

Rules for task DAGs:

• Only the root task has a SCHEDULE.
• Child tasks use AFTER parent_task_name instead of a schedule.
• To enable the whole tree, resume child tasks first, then the root task — or use SYSTEM$TASK_DEPENDENTS_ENABLE('root_task').
• A task tree can have up to 1,000 tasks in a single DAG.

CREATE OR REPLACE TASK root_task
  WAREHOUSE = COMPUTE_WH
  SCHEDULE = '15 MINUTE'
AS
INSERT INTO stage_raw SELECT * FROM orders_stream WHERE METADATA$ACTION = 'INSERT';

CREATE OR REPLACE TASK child_task_1
  WAREHOUSE = COMPUTE_WH
  AFTER root_task
AS
INSERT INTO stage_clean SELECT * FROM stage_raw WHERE amount > 0;

CREATE OR REPLACE TASK child_task_2
  WAREHOUSE = COMPUTE_WH
  AFTER child_task_1
AS
CALL aggregate_daily_totals();

-- Option A: resume individually (children first)
ALTER TASK child_task_2 RESUME;
ALTER TASK child_task_1 RESUME;
ALTER TASK root_task RESUME;

-- Option B: system function enables all children, then resume root
SELECT SYSTEM$TASK_DEPENDENTS_ENABLE('root_task');
ALTER TASK root_task RESUME;

-- Session-level history
SELECT *
FROM TABLE(INFORMATION_SCHEMA.TASK_HISTORY(
    SCHEDULED_TIME_RANGE_START => DATEADD('hour', -1, CURRENT_TIMESTAMP()),
    TASK_NAME => 'root_task'
))
ORDER BY SCHEDULED_TIME DESC;

-- Account-level (90-day retention)
SELECT name, state, error_message, scheduled_time, completed_time
FROM SNOWFLAKE.ACCOUNT_USAGE.TASK_HISTORY
WHERE SCHEDULED_TIME > DATEADD('day', -7, CURRENT_TIMESTAMP())
ORDER BY SCHEDULED_TIME DESC;

Task DAG Structure

The root task runs every 15 minutes on a WAREHOUSE schedule. On success, child_task_1 runs immediately after. On success of child_task_1, child_task_2 runs. Each task uses the AFTER clause to express the dependency. The DAG terminates if any task fails — subsequent dependents do not run.

9. Serverless Tasks vs. Warehouse Tasks

10. Stream + Task Pipeline Pattern

The canonical Snowflake incremental pipeline combines a stream and a task:

1. Source table receives new data (via Snowpipe, COPY, application INSERT).
2. Stream records the delta.
3. Task fires on schedule, reads the stream inside a DML statement, writes to target.
4. Stream offset advances; the next task run sees only the next batch of changes.

-- Full stream + task pipeline example
-- Step 1: Stream on source
CREATE OR REPLACE STREAM raw_orders_stream ON TABLE raw.orders;

-- Step 2: Task to process stream every 5 minutes
CREATE OR REPLACE TASK process_orders_task
  USER_TASK_MANAGED_INITIAL_WAREHOUSE_SIZE = 'SMALL'
  SCHEDULE = '5 MINUTE'
WHEN
  SYSTEM$STREAM_HAS_DATA('raw_orders_stream')
AS
INSERT INTO curated.orders (order_id, customer_id, amount, inserted_at)
SELECT
    order_id,
    customer_id,
    amount,
    CURRENT_TIMESTAMP()
FROM raw_orders_stream
WHERE METADATA$ACTION = 'INSERT'
  AND METADATA$ISUPDATE = FALSE;

-- Step 3: Resume the task
ALTER TASK process_orders_task RESUME;

Note the WHEN SYSTEM$STREAM_HAS_DATA(...) clause — this is a precondition. The task only executes if the stream has unconsumed data, saving compute credits when there is nothing to process.

11. Monitoring & Troubleshooting

-- List all streams and their staleness status
SHOW STREAMS;

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

-- Task history (session context — last 7 days by default)
SELECT *
FROM TABLE(INFORMATION_SCHEMA.TASK_HISTORY())
ORDER BY SCHEDULED_TIME DESC
LIMIT 20;

-- Account Usage task history (90-day retention, may have latency)
SELECT
    name,
    state,
    error_message,
    scheduled_time,
    completed_time,
    DATEDIFF('second', scheduled_time, completed_time) AS duration_seconds
FROM SNOWFLAKE.ACCOUNT_USAGE.TASK_HISTORY
WHERE state = 'FAILED'
ORDER BY scheduled_time DESC;

CheatSheet

Quizzes

Flashcards