Implementing Snowflake Streams and Tasks for Lightweight CDC and Automated Transformation Without Airflow

Introduction

You've got a source table in Snowflake getting hammered with new orders, status updates, and cancellations every few minutes. Your dbt models downstream need to reflect that reality, but you don't want to run a full table scan every hour just to pick up the 400 rows that actually changed. And your team doesn't want to spin up Airflow — or pay for a managed orchestrator — just to trigger a few SQL transformations on a schedule.

This is exactly the problem that Snowflake Streams and Tasks were designed to solve. Together, they give you a native, warehouse-internal CDC (Change Data Capture) mechanism and a lightweight scheduler — no external orchestration layer required. Streams track which rows were inserted, updated, or deleted since you last consumed the stream. Tasks execute SQL or stored procedures on a schedule or in response to a dependency chain completing. Combine them, and you get an event-driven, incremental processing pipeline that lives entirely inside Snowflake.

By the end of this lesson, you'll understand how streams work at the offset-pointer level, how to design task DAGs that mirror what you'd normally build in Airflow, and how to handle the tricky edge cases that trip up even experienced engineers — like stream staleness, task retries, and the interaction between streams and transactions.

What you'll learn:

• How Snowflake Streams implement CDC using offset tracking and change data metadata columns
• The difference between standard, append-only, and insert-only streams, and when to use each
• How to build multi-step task trees with predecessor dependencies and conditional execution
• How to combine streams and tasks to build a production-grade incremental pipeline without Airflow
• Advanced patterns: idempotent stream consumption, error handling in tasks, and monitoring task execution history

---

Prerequisites

You should be comfortable with:

• Core Snowflake concepts: databases, schemas, tables, warehouses, roles — if you need a foundation here, the Cloud Data Warehouses: Snowflake vs BigQuery vs Redshift Complete Comparison Guide is a solid starting point
• SQL DDL and DML, including MERGE statements
• Basic understanding of CDC concepts — what it means to track inserts, updates, and deletes at the row level
• Familiarity with incremental data processing; you'll see parallels to incremental dbt models

---

How Snowflake Streams Actually Work Under the Hood

Before writing a single line of SQL, let's establish the mental model that makes streams predictable rather than magical.

Snowflake tables are stored as immutable micro-partitions — columnar files that are never modified in place. When you update or delete a row, Snowflake writes new micro-partitions and marks the old ones for eventual removal. This is key: the old data doesn't disappear immediately, and that's what makes streams possible.

A stream is essentially a change tracking cursor — a pointer into the table's transaction log. When you create a stream on a table, Snowflake records the current transaction offset (the internal timestamp/sequence number). From that moment forward, any DML that touches the table — inserts, updates, deletes — is tracked relative to that offset.

When you query a stream, Snowflake computes the diff between the stream's current offset and the latest committed state of the table. The result is a set of rows augmented with three system metadata columns:

Metadata Column	Description
METADATA$ACTION	INSERT or DELETE
METADATA$ISUPDATE	TRUE when this row is part of an update pair
METADATA$ROW_ID	Unique, immutable row identifier

An UPDATE is represented as two rows: a DELETE of the old value and an INSERT of the new value, both with METADATA$ISUPDATE = TRUE. This design is intentional — it gives you maximum flexibility to handle updates however your use case demands.

The critical operational implication: consuming a stream is a transaction that advances the offset pointer. You must consume stream data inside a DML transaction (INSERT, UPDATE, MERGE, DELETE). If you just SELECT from a stream without wrapping it in DML, the offset does not advance. This is both a feature (you can query a stream multiple times before consuming it) and a footgun (forgetting to wrap consumption in DML means you'll reprocess the same changes repeatedly).

---

Three Types of Streams and When to Choose Each

Snowflake offers three stream types, and picking the wrong one is a common source of unnecessary compute cost or missing data.

Standard Streams

Standard streams track all DML: inserts, updates, and deletes. They're the most flexible and the most expensive because Snowflake must track full row-level history. Use these when:

• Your source table has updates and deletes that your downstream models need to reflect
• You're maintaining SCD Type 2 history or any pattern that requires knowing what changed, not just what was added
• Your source is an operational database replicated via Fivetran or Airbyte

CREATE OR REPLACE STREAM raw.orders_stream
  ON TABLE raw.orders
  SHOW_INITIAL_ROWS = FALSE;

SHOW_INITIAL_ROWS = FALSE means the stream starts empty — only changes after creation are captured. Set it to TRUE if you want the stream to initially return all existing rows as inserts, which is useful for bootstrapping a new pipeline.

Append-Only Streams

Append-only streams track only inserts. If a row is updated after insertion, the stream ignores those updates. If a row is deleted, the stream ignores the delete. This is a significant constraint, but it comes with a meaningful performance benefit — Snowflake can optimize the change tracking because it only needs to identify net-new rows.

CREATE OR REPLACE STREAM raw.events_stream
  ON TABLE raw.events
  APPEND_ONLY = TRUE;

Use append-only streams when:

• Your source table is truly append-only (event logs, clickstream data, IoT sensor readings)
• You're feeding an analytics aggregate where late updates don't matter
• You need maximum throughput on high-volume tables

Insert-Only Streams

Insert-only streams work on external tables — tables that read data from files in cloud storage (S3, Azure Blob, GCS). They track when new files land in the stage that backs the external table, treating each new file as a set of inserts. Updates and deletes are not supported because external tables don't support DML.

CREATE OR REPLACE STREAM raw.sensor_readings_stream
  ON EXTERNAL TABLE raw.sensor_readings_ext
  INSERT_ONLY = TRUE;

This pattern is powerful for event-driven file ingestion: new files land in S3, the stream captures them, a task processes them into an internal table. It's a lightweight alternative to Snowpipe with more control over transformation logic.

---

Managing Stream Staleness

Stream staleness is the operational gotcha that bites teams months after they deploy their pipeline. Here's the deal: a stream's change tracking relies on Time Travel data for the source table. If a stream's offset falls outside the table's data retention period, the stream becomes stale — it cannot return accurate change data, and querying it throws an error.

You can check a stream's staleness with:

SHOW STREAMS LIKE 'orders_stream' IN SCHEMA raw;
-- Look at the STALE column and STALE_AFTER column

The STALE_AFTER column tells you exactly when your stream will expire if not consumed. For a table with a 1-day data retention period (the default on the Standard edition), your stream will go stale if it isn't consumed within that window.

Prevention strategies:

1. Increase table data retention. For tables with streams, set DATA_RETENTION_TIME_IN_DAYS = 7 (or higher if you're on Enterprise). This extends the stream's survival window.

ALTER TABLE raw.orders SET DATA_RETENTION_TIME_IN_DAYS = 7;

2. Monitor stream staleness proactively. Query INFORMATION_SCHEMA.STREAMS or SHOW STREAMS in a scheduled monitoring task.

3. Design tasks to run at least once per data retention window. If your task runs every 5 minutes, staleness is essentially never a concern. If your task runs weekly, you need a longer retention window.

---

Building Your First Task

Tasks are Snowflake's native scheduler. A task can execute a single SQL statement, a stored procedure call, or a scripted block (using Snowflake Scripting). Tasks can run on a fixed schedule (cron or interval) or as part of a dependency chain (triggered by a predecessor task completing successfully).

Basic Task Anatomy

CREATE OR REPLACE TASK transform.process_new_orders
  WAREHOUSE = TRANSFORM_WH
  SCHEDULE = '5 MINUTE'
  COMMENT = 'Process new and updated orders from the CDC stream'
AS
  MERGE INTO transform.orders_fact AS target
  USING (
    SELECT
      order_id,
      customer_id,
      order_status,
      total_amount,
      updated_at,
      METADATA$ACTION,
      METADATA$ISUPDATE
    FROM raw.orders_stream
    WHERE METADATA$ACTION = 'INSERT'
  ) AS source
  ON target.order_id = source.order_id
  WHEN MATCHED THEN UPDATE SET
    target.order_status = source.order_status,
    target.total_amount = source.total_amount,
    target.updated_at = source.updated_at
  WHEN NOT MATCHED THEN INSERT (
    order_id, customer_id, order_status, total_amount, updated_at
  ) VALUES (
    source.order_id, source.customer_id, source.order_status,
    source.total_amount, source.updated_at
  );

A few things to understand about this definition:

Warehouse assignment: Every task needs a warehouse. This is the compute that runs the task's SQL. Use an appropriately sized warehouse — XSMALL is usually fine for lightweight stream processing. The task will auto-suspend the warehouse if no other work is running, so you only pay while the task is actually executing.

Schedule: '5 MINUTE' means "run every 5 minutes." You can also use cron syntax for more complex schedules:

SCHEDULE = 'USING CRON 0 6 * * MON-FRI America/New_York'

Task state: Tasks are created in a suspended state. You must explicitly resume them:

ALTER TASK transform.process_new_orders RESUME;

Suspend them with:

ALTER TASK transform.process_new_orders SUSPEND;

---

Designing a Multi-Step Task Tree (The Airflow Alternative)

A single task running a single MERGE covers simple cases. Real pipelines have dependencies: load raw data, validate it, transform it, aggregate it. In Airflow, you'd model this as a DAG. In Snowflake Tasks, you model it as a task tree — a directed acyclic graph where each task specifies its predecessor.

Let's build a realistic example. Imagine an e-commerce platform where:

1. Orders land in raw.orders via a Fivetran replication
2. We need to standardize and validate them into staging.orders
3. Then compute daily revenue metrics into analytics.daily_revenue
4. Then refresh a customer lifetime value table in analytics.customer_ltv

The raw stream captures changes. The staging transformation applies business rules and type casting. The analytics models aggregate. Steps 3 and 4 can run in parallel after step 2 completes.

Step 1: Root Task (Scheduled)

-- Root task: runs on schedule, checks if stream has data
CREATE OR REPLACE TASK pipeline.check_orders_stream
  WAREHOUSE = TRANSFORM_XS_WH
  SCHEDULE = '5 MINUTE'
  WHEN SYSTEM$STREAM_HAS_DATA('raw.orders_stream')
AS
  -- Root task body: can be a lightweight sentinel or actual work
  -- We'll use it to trigger the pipeline
  CALL pipeline.log_pipeline_start('orders');

The WHEN SYSTEM$STREAM_HAS_DATA(...) clause is the key to efficiency. Without it, the task fires every 5 minutes regardless of whether there's anything to process. With it, Snowflake checks whether the stream has unconsumed changes before spinning up the warehouse. If the stream is empty, the task skips execution entirely and pays nothing.

Step 2: Staging Transformation (Predecessor: Root)

CREATE OR REPLACE TASK pipeline.transform_orders_to_staging
  WAREHOUSE = TRANSFORM_S_WH
  AFTER pipeline.check_orders_stream
AS
  MERGE INTO staging.orders AS target
  USING (
    SELECT
      order_id,
      customer_id,
      UPPER(TRIM(order_status))                          AS order_status,
      CAST(raw_amount AS NUMBER(18, 2))                  AS order_amount,
      TRY_TO_TIMESTAMP(order_created_at)                 AS order_created_at,
      TRY_TO_TIMESTAMP(order_updated_at)                 AS order_updated_at,
      CURRENT_TIMESTAMP()                                AS _loaded_at,
      METADATA$ACTION,
      METADATA$ISUPDATE,
      METADATA$ROW_ID
    FROM raw.orders_stream
    WHERE METADATA$ACTION = 'INSERT'
      -- For updates: INSERT row contains the new values, DELETE has old values
      -- We only want new values, so filtering INSERT is correct
  ) AS source
  ON target.order_id = source.order_id
  WHEN MATCHED THEN UPDATE SET
    target.order_status    = source.order_status,
    target.order_amount    = source.order_amount,
    target.order_updated_at = source.order_updated_at,
    target._loaded_at      = source._loaded_at
  WHEN NOT MATCHED THEN INSERT VALUES (
    source.order_id,
    source.customer_id,
    source.order_status,
    source.order_amount,
    source.order_created_at,
    source.order_updated_at,
    source._loaded_at
  );

Notice that this task consumes the stream — the MERGE is a DML statement that advances the stream's offset. Once this task completes successfully, the stream pointer moves forward. Any downstream task reading from raw.orders_stream in the same pipeline run would see an empty stream. This is by design: stream consumption is transactional.

Step 3a and 3b: Parallel Analytics Tasks

-- Task 3a: Daily revenue aggregation
CREATE OR REPLACE TASK pipeline.refresh_daily_revenue
  WAREHOUSE = TRANSFORM_S_WH
  AFTER pipeline.transform_orders_to_staging
AS
  INSERT OVERWRITE INTO analytics.daily_revenue
  SELECT
    DATE_TRUNC('DAY', order_created_at)  AS revenue_date,
    COUNT(DISTINCT order_id)             AS order_count,
    SUM(order_amount)                    AS total_revenue,
    AVG(order_amount)                    AS avg_order_value
  FROM staging.orders
  WHERE order_status NOT IN ('CANCELLED', 'REFUNDED')
  GROUP BY 1;

-- Task 3b: Customer LTV refresh
CREATE OR REPLACE TASK pipeline.refresh_customer_ltv
  WAREHOUSE = TRANSFORM_S_WH
  AFTER pipeline.transform_orders_to_staging
AS
  MERGE INTO analytics.customer_ltv AS target
  USING (
    SELECT
      customer_id,
      COUNT(DISTINCT order_id)  AS total_orders,
      SUM(order_amount)         AS lifetime_value,
      MIN(order_created_at)     AS first_order_at,
      MAX(order_created_at)     AS last_order_at
    FROM staging.orders
    WHERE order_status NOT IN ('CANCELLED', 'REFUNDED')
    GROUP BY customer_id
  ) AS source
  ON target.customer_id = source.customer_id
  WHEN MATCHED THEN UPDATE SET
    target.total_orders    = source.total_orders,
    target.lifetime_value  = source.lifetime_value,
    target.last_order_at   = source.last_order_at
  WHEN NOT MATCHED THEN INSERT VALUES (
    source.customer_id,
    source.total_orders,
    source.lifetime_value,
    source.first_order_at,
    source.last_order_at
  );

Both tasks 3a and 3b have the same predecessor (transform_orders_to_staging), so they run in parallel once the staging transformation completes. Snowflake's task scheduler handles this automatically — no additional configuration needed.

Activating the Task Tree

Tasks with predecessors must be resumed in reverse dependency order: leaf nodes first, root last.

-- Resume in dependency order (leaves first)
ALTER TASK pipeline.refresh_daily_revenue RESUME;
ALTER TASK pipeline.refresh_customer_ltv RESUME;
ALTER TASK pipeline.transform_orders_to_staging RESUME;
ALTER TASK pipeline.check_orders_stream RESUME;  -- Root last

This ordering matters because resuming the root task starts the schedule. If downstream tasks aren't resumed yet, the root might fire and find its children suspended — causing failures.

---

The Transaction Boundary Problem: A Deep Dive

Here's where many engineers get tripped up: stream consumption and transactional safety.

When a task executes a DML statement that reads from a stream, that entire statement runs in a single transaction. The stream offset advances only when the transaction commits. If the statement fails mid-execution (say, the warehouse hits a timeout), the transaction rolls back, the stream offset does not advance, and the same changes will be available on the next run. This is exactly what you want — it gives you at-least-once processing semantics automatically.

But there's a subtlety: you can only consume a stream once per transaction. If you write two separate SQL statements in a single task (using Snowflake Scripting), and both try to SELECT from the same stream, the second SELECT will see an empty stream because the first DML already advanced the offset.

Here's how this breaks in practice:

-- BROKEN: Two separate DML statements in a scripted task
BEGIN
  -- First statement consumes the stream
  INSERT INTO staging.orders
  SELECT * FROM raw.orders_stream WHERE METADATA$ACTION = 'INSERT';
  
  -- Second statement sees an EMPTY stream - offset already advanced!
  INSERT INTO audit.order_changes
  SELECT * FROM raw.orders_stream;  -- Returns 0 rows
END;

The fix is to materialize the stream into a temporary table first, then reference the temp table in multiple statements:

-- CORRECT: Materialize stream, then process from temp table
BEGIN
  CREATE OR REPLACE TEMPORARY TABLE pipeline.orders_stream_buffer AS
  SELECT * FROM raw.orders_stream;
  
  -- Now process from the buffer as many times as needed
  INSERT INTO staging.orders
  SELECT * FROM pipeline.orders_stream_buffer WHERE METADATA$ACTION = 'INSERT';
  
  INSERT INTO audit.order_changes
  SELECT * FROM pipeline.orders_stream_buffer;
END;

---

Implementing Deletes: The CDC Pattern for SCD Type 2

Tracking deletes is where stream-based CDC really earns its complexity budget. Consider a slowly changing dimension for customer records. When a customer is deleted from the source system, you don't want to hard-delete them from your warehouse — you want to close out their current record by setting an is_active = FALSE flag and recording the deletion timestamp.

Here's how to handle the full insert/update/delete lifecycle in a single MERGE:

MERGE INTO staging.customers AS target
USING (
  SELECT
    customer_id,
    email,
    full_name,
    country,
    created_at,
    updated_at,
    METADATA$ACTION,
    METADATA$ISUPDATE,
    METADATA$ROW_ID
  FROM raw.customers_stream
) AS source
ON target.customer_id = source.customer_id
   AND target.is_current = TRUE  -- Only match against current records

-- Handle net-new inserts (new customer)
WHEN NOT MATCHED AND source.METADATA$ACTION = 'INSERT'
  THEN INSERT (
    customer_id, email, full_name, country,
    created_at, updated_at, is_current, deleted_at
  ) VALUES (
    source.customer_id, source.email, source.full_name, source.country,
    source.created_at, source.updated_at, TRUE, NULL
  )

-- Handle updates (customer record changed)
WHEN MATCHED AND source.METADATA$ACTION = 'INSERT' AND source.METADATA$ISUPDATE = TRUE
  THEN UPDATE SET
    target.email      = source.email,
    target.full_name  = source.full_name,
    target.country    = source.country,
    target.updated_at = source.updated_at

-- Handle deletes (soft delete)
WHEN MATCHED AND source.METADATA$ACTION = 'DELETE' AND source.METADATA$ISUPDATE = FALSE
  THEN UPDATE SET
    target.is_current  = FALSE,
    target.deleted_at  = CURRENT_TIMESTAMP();

Notice the logic for distinguishing a "real" delete from the DELETE half of an update pair: METADATA$ISUPDATE = FALSE means it's a standalone delete, not part of an update. Without this filter, every update would trigger a soft delete — a subtle but devastating bug.

---

Error Handling and Task Retry Logic

Production tasks fail. Warehouses time out. Source tables get locked. You need a strategy.

Checking Task Execution History

-- Last 10 task run results
SELECT
  NAME,
  STATE,
  SCHEDULED_TIME,
  COMPLETED_TIME,
  ERROR_CODE,
  ERROR_MESSAGE,
  QUERY_ID
FROM TABLE(INFORMATION_SCHEMA.TASK_HISTORY(
  TASK_NAME    => 'TRANSFORM_ORDERS_TO_STAGING',
  SCHEDULED_TIME_RANGE_START => DATEADD('hour', -24, CURRENT_TIMESTAMP()),
  RESULT_LIMIT => 10
))
ORDER BY SCHEDULED_TIME DESC;

The STATE column will be SUCCEEDED, FAILED, or SKIPPED. A SKIPPED state on the root task means SYSTEM$STREAM_HAS_DATA() returned false — no data, nothing to do.

Automatic Retry on Failure

As of Snowflake's recent releases, tasks support native retry configuration:

CREATE OR REPLACE TASK pipeline.transform_orders_to_staging
  WAREHOUSE = TRANSFORM_S_WH
  AFTER pipeline.check_orders_stream
  ERROR_INTEGRATION = my_notification_integration  -- optional: send alerts on failure
  SUSPEND_TASK_AFTER_NUM_FAILURES = 5              -- auto-suspend after 5 consecutive failures
AS
  -- task body

SUSPEND_TASK_AFTER_NUM_FAILURES is a critical safety valve. Without it, a consistently failing task will keep firing and potentially causing unintended side effects (or at minimum, burning warehouse credits). After N failures, Snowflake auto-suspends the task tree, and you get paged to investigate.

For notification on failure, wire up an error integration that sends to email or SNS:

CREATE OR REPLACE NOTIFICATION INTEGRATION task_failure_alerts
  TYPE = EMAIL
  ENABLED = TRUE
  ALLOWED_RECIPIENTS = ('data-team@yourcompany.com');

Idempotent Task Design

Stream-based tasks are inherently idempotent for inserts (MERGE handles duplicates) and updates (MERGE overwrites with latest values). But for aggregation tasks like refresh_daily_revenue, you're doing a full INSERT OVERWRITE which is also idempotent — running it twice gives the same result.

The pattern to avoid is straight INSERT INTO without deduplication logic. If a task runs twice due to a retry, you'll get double-counted rows. Always use MERGE or INSERT OVERWRITE for task bodies.

---

Monitoring and Observability Without an External Orchestrator

One thing Airflow gives you out of the box is a UI for pipeline monitoring. Without it, you need to build your own observability layer. This doesn't have to be complicated.

A Simple Pipeline Audit Table

CREATE TABLE IF NOT EXISTS pipeline.task_audit_log (
  run_id          STRING    DEFAULT UUID_STRING(),
  task_name       STRING,
  pipeline_name   STRING,
  status          STRING,   -- STARTED, SUCCEEDED, FAILED
  rows_processed  INTEGER,
  started_at      TIMESTAMP_NTZ DEFAULT CURRENT_TIMESTAMP(),
  completed_at    TIMESTAMP_NTZ,
  error_message   STRING
);

Instrument your tasks to write to this table using Snowflake Scripting:

CREATE OR REPLACE TASK pipeline.transform_orders_to_staging
  WAREHOUSE = TRANSFORM_S_WH
  AFTER pipeline.check_orders_stream
AS
DECLARE
  v_rows_processed INTEGER DEFAULT 0;
  v_run_id STRING DEFAULT UUID_STRING();
BEGIN
  -- Log start
  INSERT INTO pipeline.task_audit_log (run_id, task_name, pipeline_name, status)
  VALUES (:v_run_id, 'transform_orders_to_staging', 'orders_pipeline', 'STARTED');

  -- Do the work
  MERGE INTO staging.orders AS target
  USING (SELECT * FROM raw.orders_stream WHERE METADATA$ACTION = 'INSERT') AS source
  ON target.order_id = source.order_id
  WHEN MATCHED THEN UPDATE SET target.order_status = source.order_status
  WHEN NOT MATCHED THEN INSERT VALUES (source.order_id, source.customer_id,
    source.order_status, source.order_amount, source.order_created_at,
    source.order_updated_at, CURRENT_TIMESTAMP());

  -- Capture rows affected
  v_rows_processed := SQLROWCOUNT;

  -- Log success
  UPDATE pipeline.task_audit_log
  SET status = 'SUCCEEDED',
      rows_processed = :v_rows_processed,
      completed_at = CURRENT_TIMESTAMP()
  WHERE run_id = :v_run_id;

EXCEPTION WHEN OTHER THEN
  UPDATE pipeline.task_audit_log
  SET status = 'FAILED',
      completed_at = CURRENT_TIMESTAMP(),
      error_message = :sqlerrm
  WHERE run_id = :v_run_id;
  RAISE;  -- Re-raise so Snowflake marks the task as FAILED
END;

This gives you a queryable audit history without needing Datadog, Monte Carlo, or any external tool. You can build a simple BI dashboard on top of pipeline.task_audit_log to track pipeline health over time.

---

When Streams + Tasks Make Sense (And When They Don't)

This architecture isn't the right answer for every problem. Let's be honest about the trade-offs.

Streams + Tasks shine when:

• Your transformation logic is SQL-expressible (MERGE, INSERT OVERWRITE, window functions)
• You want sub-5-minute latency without paying for a streaming infrastructure
• Your team wants to minimize operational surface area — no Airflow cluster to manage, no dbt Cloud subscription needed
• You're working in a Snowflake-centric stack and want to stay warehouse-native

Streams + Tasks struggle when:

• You need complex branching logic or dynamic task generation (Airflow's TaskFlow API handles this more elegantly)
• Your transformation requires Python, dbt macros, or non-SQL logic
• You need rich retry policies with exponential backoff and dead letter queues
• You need cross-system dependencies (e.g., "wait for the Kafka topic to drain before running")
• Your observability requirements demand tool-native integration with Datadog, PagerDuty, or similar

This last point is worth emphasizing. If you're already using Airflow in your organization for other pipelines, adding a parallel Streams + Tasks system creates an operational split-brain: some pipelines in Airflow's UI, some invisible to it. You end up with two orchestration systems that don't know about each other, which complicates incident response and onboarding.

---

Streams on Views and Dynamic Tables: Expanding the Pattern

As of recent Snowflake releases, you can also create streams on views (with restrictions) and use Dynamic Tables as a higher-level abstraction that embeds stream-like behavior.

Streams on Views

You can create a stream on a view, but only if the view is "simple" — no aggregations, no UNION, no DISTINCT. The stream resolves to the underlying base table's change tracking.

CREATE VIEW raw.orders_us AS
SELECT * FROM raw.orders WHERE country_code = 'US';

CREATE STREAM raw.orders_us_stream ON VIEW raw.orders_us;

This is useful when you want to partition change tracking by a logical subset without creating separate tables.

Dynamic Tables: Streams + Tasks Abstracted

Dynamic Tables (GA as of 2024) are worth understanding as a conceptual alternative. A Dynamic Table is defined by a SELECT query, and Snowflake automatically refreshes it when the upstream source changes — essentially a materialized view with CDC semantics baked in.

CREATE OR REPLACE DYNAMIC TABLE staging.orders
  TARGET_LAG = '5 minutes'
  WAREHOUSE = TRANSFORM_XS_WH
AS
SELECT
  order_id,
  customer_id,
  UPPER(TRIM(order_status)) AS order_status,
  CAST(raw_amount AS NUMBER(18,2)) AS order_amount,
  TRY_TO_TIMESTAMP(order_created_at) AS order_created_at
FROM raw.orders;

Dynamic Tables handle the stream creation, task scheduling, and MERGE logic internally. The trade-off: less control over execution logic, no ability to add audit instrumentation, and limited support for complex multi-table MERGE patterns.

For most incremental transformation patterns in the medallion architecture (bronze → silver → gold), Dynamic Tables are simpler and less error-prone for teams that don't need fine-grained control. Streams + Tasks are the right choice when you need that control.

---

Integrating with dbt: Streams as Sources, Tasks as Schedulers

If you're running a dbt project, you can combine the two tools elegantly. Use Streams + Tasks to handle the high-frequency incremental layer (the raw → staging hop), and use dbt for the staging → mart transformation layer where you want model testing, documentation, and lineage.

The integration pattern:

1. A Snowflake Task runs every 5 minutes, consuming the raw stream and writing to staging.orders (fully controlled by SQL)
2. A separate Task, scheduled less frequently (hourly or on-demand), calls a stored procedure that invokes dbt run --select orders_mart
3. The dbt models read from staging.orders as a source, treating it as a stable, pre-cleaned input

You can invoke dbt from a task using Snowflake's Git integration and Python-based stored procedures — but this introduces complexity. For most teams, the simpler architecture is: Tasks handle the stream consumption, dbt handles the downstream analytics models on a separate schedule. The interface between them is a clean, well-structured staging table.

---

Hands-On Exercise

Build a complete, production-ready stream + task pipeline for a hypothetical SaaS application. Your setup:

Source table: raw.subscriptions — receives inserts (new subscribers), updates (plan changes, renewals), and deletes (cancellations) every few minutes via a Fivetran replication job.

Target tables:

• staging.subscriptions — cleaned, type-cast, with soft deletes
• analytics.subscription_metrics — daily active subscribers and MRR (monthly recurring revenue)

Your tasks:

1. Create the source table and stream:

CREATE TABLE IF NOT EXISTS raw.subscriptions (
  subscription_id   STRING        NOT NULL,
  customer_id       STRING        NOT NULL,
  plan_name         STRING,
  plan_amount_usd   FLOAT,
  status            STRING,       -- 'active', 'cancelled', 'paused'
  started_at        TIMESTAMP_NTZ,
  cancelled_at      TIMESTAMP_NTZ,
  updated_at        TIMESTAMP_NTZ
);

CREATE OR REPLACE STREAM raw.subscriptions_stream
  ON TABLE raw.subscriptions
  SHOW_INITIAL_ROWS = FALSE;

2. Build the staging task that handles inserts, updates, and soft deletes using a full MERGE with METADATA$ column filtering.

3. Build the analytics task that computes daily active subscriber count and MRR from staging.subscriptions, refreshing with INSERT OVERWRITE.

4. Wire up the task tree with the root task using SYSTEM$STREAM_HAS_DATA() to skip execution when no changes are present.

5. Add an audit log using Snowflake Scripting in the staging task.

6. Test your pipeline by inserting a few rows, updating one, and deleting one, then manually executing the root task with EXECUTE TASK pipeline.check_subscriptions_stream and verifying the downstream results.

Bonus challenge: Modify the staging task to use the temp table materialization pattern so you can write both to staging.subscriptions and to a separate audit.subscription_changes table in a single task run without double-consuming the stream.

---

Common Mistakes & Troubleshooting

"Stream has no data" but rows definitely exist

Check whether your stream was consumed by a previous query that wrapped in DML. Query SHOW STREAMS and check the STALE and ROWS columns. If the stream was recently consumed, it will show as empty. Also check whether another task or manual query advanced the offset unexpectedly.

Task runs but processes 0 rows

Your stream might have been consumed by a different process (a manual test query wrapped in a BEGIN/COMMIT, or a competing task). Check TASK_HISTORY to see if another execution overlapped. Also verify your METADATA$ column filters — if you wrote WHERE METADATA$ACTION = 'INSERT' and the only recent changes were deletes, you'll process zero rows on purpose.

Task tree fires but child tasks don't execute

Verify that all child tasks are in RESUMED state. SHOW TASKS returns state for all tasks in the schema — look for state = 'suspended' on any child task.

MERGE fails with "table is locked"

If another session has an open transaction on the target table, your MERGE will block or timeout. This happens when Fivetran or Airbyte is loading into a staging table that your task is simultaneously reading. Stagger your load and transform schedules, or use task scheduling to run transformations after the ingestion window closes.

Stream goes stale on a low-volume table

Low-volume tables may not be consumed frequently enough. If your stream runs weekly and your table has a 1-day retention, the stream goes stale after day 1. Increase DATA_RETENTION_TIME_IN_DAYS on the table or increase task frequency.

"Invalid stream" error after a table was recreated

CREATE OR REPLACE TABLE invalidates all existing streams on that table — it's treated as a new table. If your upstream data loading process uses CREATE OR REPLACE TABLE instead of TRUNCATE + INSERT, you'll lose your stream. Always use TRUNCATE for tables with streams, never CREATE OR REPLACE.

---

Summary & Next Steps

Snowflake Streams and Tasks give you a genuinely capable CDC and orchestration layer without leaving the warehouse. You've seen how streams use offset-based change tracking to identify new, modified, and deleted rows; how the three stream types (standard, append-only, insert-only) map to different source characteristics; and how to build task trees that execute SQL pipelines on schedule or in response to upstream changes.

The patterns that separate production-grade implementations from naive ones:

• Always gate root tasks with SYSTEM$STREAM_HAS_DATA() to avoid wasteful warehouse consumption
• Materialize streams to a temp table when you need to reference change data in multiple statements
• Use the full METADATA$ column semantics to correctly distinguish inserts, updates, and deletes
• Build audit logging directly into your task stored procedures using Snowflake Scripting
• Monitor STALE_AFTER and configure adequate data retention to prevent silent stream staleness

Where to go from here:

If your transformation needs have outgrown pure SQL and you want test coverage and documentation alongside your incremental models, the natural next step is integrating what you've built here with a multi-layer dbt project where dbt handles the staging → mart layer while streams + tasks own the raw → staging hop.

If you're running this at scale and want to understand how Snowflake's underlying storage engine affects your micro-partition scanning costs, How Data Warehouses Actually Store Data goes deep on columnar storage, clustering, and pruning — all directly relevant to making your MERGE operations fast.

And if you're thinking about where streams + tasks fit in the broader question of real-time vs. batch processing for your organization, Real-Time Data: When to Use Streaming vs Batch Processing will help you frame that architectural decision with clarity.

The primitives you've mastered here — offset-based change tracking, transactional stream consumption, dependency-chained task execution — are the same concepts that underpin every CDC system from Debezium to Flink. Understanding them at this level makes you a sharper architect regardless of which tools you're working with.