Lab 12 · Domain 3 · You are Priya

Streams, Tasks & Dynamic Tables

Read every Teach first section before you copy SQL. Snowpipe gets files in. Streams + Tasks move row changes through Alpine’s pipeline. Dynamic Tables are Snowflake’s declarative “keep this SELECT fresh” option — know when you’d still pick dbt.

Business problem LAB.ORDERS changes all day. Kai’s dashboard reads a curated copy that goes stale unless someone re-runs a MERGE by hand. Riley refuses to cron five overlapping jobs that step on each other.

Solution you will build Stream PIPELINE.STR_ORDERS_CDC on LAB.ORDERS, task TASK_MERGE_ORDERS that MERGEs changes into PIPELINE.ORDERS_CURATED, plus a Dynamic Table with TARGET_LAG = '1 minute' (or the commented story if edition blocks).

Before

Stale curated copy

Full reload or forgotten MERGE
No change feed
Three tools confused: pipe / stream / DT

After

CDC + scheduled merge

Stream tracks offsets
Task MERGEs only changes
DT option for declarative lag

Episode 12 · Start here

Three tools. Three jobs. Don’t mash them.

RileyEngineer — pipelines

Morgan keeps saying “just automate it.” Automate what? Landing files? Applying row changes? Refreshing a mart?

DevCoach — explains why

Snowpipe = files in. Stream+Task = “what changed since last time” + run SQL on a schedule. Dynamic Table = declare the SELECT and a lag — Snowflake refreshes it.

If you mix these on the exam, every distractor looks right.

See it — pick the right tool

Lab 11

Snowpipe

New files → table. Not a change log on an existing table.

CSV lands → rows appear in landing

This lab

Stream + Task

Stream = bookmark of changes. Task = scheduled SQL (often MERGE).

ORDER 105 status flip → stream row → MERGE

Declarative

Dynamic Table

You write AS SELECT … + TARGET_LAG. Snowflake manages refresh.

TARGET_LAG = '1 minute'

Outside Snowflake

dbt (when to prefer)

Complex DAGs, tests, CI, multi-tool warehouses. Exam may still ask Snowflake-native DT.

dbt models + tests in Git → warehouse

Contrast — pipe vs stream/task vs DT

Snowpipe Files → table serverless COPY

Stream + Task CDC → MERGE WH runs task SQL

Dynamic Table SELECT stays fresh TARGET_LAG budget

Before Step 1

Kai: “Is a stream a table I query forever?”

KaiAnalyst — still learning

If I SELECT from the stream twice, do I see the same changes both times?

DevCoach

Consuming the stream (in a DML that reads it, like our MERGE task) advances the offset. Think bookmark, not permanent history table. That’s why “stream offset” shows up on exams.

1 Curated table + stream on LAB.ORDERS

What this does: Builds the target table and a change feed. Streams need a source table that already exists (Lab 00 sample data).

SQL · stream

USE ROLE ROLE_CORE_LAB_ADMIN;
USE WAREHOUSE WH_CORE_XS;
USE DATABASE ALPINE_RETAIL;
USE SCHEMA PIPELINE;

CREATE OR REPLACE TABLE PIPELINE.ORDERS_CURATED (
  ORDER_ID NUMBER, CUSTOMER_ID NUMBER,
  ORDER_TS TIMESTAMP_NTZ, AMOUNT_USD NUMBER(10,2),
  STATUS STRING, UPDATED_AT TIMESTAMP_NTZ DEFAULT CURRENT_TIMESTAMP()
);

CREATE OR REPLACE STREAM PIPELINE.STR_ORDERS_CDC
  ON TABLE LAB.ORDERS;

✓ SHOW STREAMS should list STR_ORDERS_CDC

Before Step 2

Make the stream dirty on purpose

PriyaYou

Empty stream means I did nothing wrong — it means nothing changed since the stream was created.

RileyEngineer

Insert and update a couple of orders, then SELECT * from the stream. You should see METADATA$ACTION telling you insert vs delete.

2 Produce changes + peek the stream

What this does: Creates CDC rows so you can see the feed before the task consumes it.

SQL · change + select

INSERT INTO LAB.ORDERS VALUES
  (106, 5, '2026-02-10 09:00:00', 88.00, 'PENDING');
UPDATE LAB.ORDERS SET STATUS = 'SHIPPED' WHERE ORDER_ID = 105;

SELECT * FROM PIPELINE.STR_ORDERS_CDC;
SELECT SYSTEM$STREAM_HAS_DATA('PIPELINE.STR_ORDERS_CDC');

✓ Stream shows change rows (actions / update flags)

3 CREATE TASK + EXECUTE once

What this does: Schedules a MERGE from the stream. Bootstrap granted EXECUTE TASK on the account. Tasks start suspended — resume, execute once for the lab, then suspend so you don’t burn credits hourly.

SQL · task (create)

CREATE OR REPLACE TASK PIPELINE.TASK_MERGE_ORDERS
  WAREHOUSE = WH_CORE_XS
  SCHEDULE = '60 MINUTE'
AS
  MERGE INTO PIPELINE.ORDERS_CURATED t
  USING (
    SELECT * FROM PIPELINE.STR_ORDERS_CDC
  ) s
  ON t.ORDER_ID = s.ORDER_ID
  WHEN MATCHED AND s.METADATA$ACTION = 'DELETE' THEN DELETE
  WHEN MATCHED AND s.METADATA$ISUPDATE THEN UPDATE SET
    t.STATUS = s.STATUS, t.AMOUNT_USD = s.AMOUNT_USD,
    t.UPDATED_AT = CURRENT_TIMESTAMP()
  WHEN NOT MATCHED AND s.METADATA$ACTION = 'INSERT' THEN INSERT
    (ORDER_ID, CUSTOMER_ID, ORDER_TS, AMOUNT_USD, STATUS)
    VALUES (s.ORDER_ID, s.CUSTOMER_ID, s.ORDER_TS,
            s.AMOUNT_USD, s.STATUS);

SQL · execute + suspend

ALTER TASK PIPELINE.TASK_MERGE_ORDERS RESUME;
EXECUTE TASK PIPELINE.TASK_MERGE_ORDERS;
SHOW TASKS LIKE 'TASK_MERGE%' IN SCHEMA PIPELINE;
SELECT * FROM PIPELINE.ORDERS_CURATED ORDER BY ORDER_ID;
ALTER TASK PIPELINE.TASK_MERGE_ORDERS SUSPEND;

Full MERGE edge cases live in sql/12_streams_tasks.sql — keep the HTML chunk readable.

✓ Curated has new/updated orders; task left SUSPENDED

Before Step 4

Morgan: “Why not Dynamic Tables for everything?”

MorganManager

If DT refreshes itself, delete the tasks. Less code, less Priya.

DevCoach

DT is great for declarative freshness. You still pay for refresh compute. Complex orchestration, tests, and cross-tool graphs often stay in dbt/Tasks. Exam: know TARGET_LAG and that DT is Snowflake-native.

4 Dynamic Table (or study the comment)

What this does: Declares a continuously refreshed aggregate. If your edition blocks CREATE DYNAMIC TABLE, leave it commented — still say the TARGET_LAG story out loud.

SQL · dynamic table

CREATE OR REPLACE DYNAMIC TABLE PIPELINE.DT_ORDERS_DAILY
  TARGET_LAG = '1 minute'
  WAREHOUSE = WH_CORE_XS
AS
SELECT DATE_TRUNC('day', ORDER_TS) AS ORDER_DAY,
       STATUS,
       COUNT(*) AS ORDER_CNT,
       SUM(AMOUNT_USD) AS AMOUNT_SUM
FROM LAB.ORDERS
GROUP BY 1, 2;

SHOW DYNAMIC TABLES LIKE 'DT_ORDERS%' IN SCHEMA PIPELINE;

✓ DT created — or you can explain TARGET_LAG without running it

Exam presses further — say these out loud

Q1. What is a stream “offset” in plain English?

Say: The bookmark of which changes you’ve already consumed. Reading the stream in a consuming DML advances it so the next run only sees newer changes.

Q2. What is a task graph?

Say: Tasks can depend on other tasks (AFTER …). Root task has a schedule; children run when parents succeed — Snowflake-native DAG without an external orchestrator.

Q3. What drives Dynamic Table cost?

Say: Refresh compute on the warehouse you assign, driven by how often lag requires refresh and how heavy the SELECT is. Tighter TARGET_LAG → more frequent work.

Q4. When prefer dbt vs Dynamic Tables?

Say: Prefer DT for simple Snowflake-native continuous transforms with a lag SLO. Prefer dbt when you need Git-centric models, tests, docs, and complex multi-project DAGs. Exam may highlight Snowflake-native DT.

Q5. Snowpipe vs stream — can a stream replace Snowpipe?

Say: No. Streams track DML on tables already in Snowflake. Snowpipe loads files from stages/storage. Different layer of the pipeline.

Priya’s wrap-up (say this out loud)
“I put a stream on LAB.ORDERS, merged changes into PIPELINE.ORDERS_CURATED with a task, and practiced EXECUTE TASK then SUSPEND. Dynamic Tables give TARGET_LAG freshness without hand-built MERGE tasks — I still know when dbt wins. Outcome: I can separate Snowpipe, streams/tasks, and DTs on any Domain 3 question.”

Next Lab 13 — Query Profile and the three caches. Domain 4 begins.