Real-Time Pipeline with Snowflake Streams

Build an automated ELT pipeline using Snowflake Streams and Tasks.

Introduction

30 Second Summary

Think about how an online store handles thousands of new orders every hour. Someone needs to figure out which orders are new, clean up messy data, and produce reports for the business team. Doing that by hand every time new data arrives is a recipe for burnout.

In this project, you will build an automated data pipeline in Snowflake that detects new order events the instant they arrive, cleans them, and produces a daily sales summary. All without you lifting a finger after the initial setup.

What You'll Build

You will insert a batch of new orders, then sit back and watch your pipeline automatically detect the new rows, clean them, and refresh a revenue report in your browser.

By the end of this project, you'll have:

A Stream that instantly identifies which rows in your orders table are new, so downstream processing never rescans the full table.
A scheduled Task that fires every minute, reads only the new rows, cleans the data, and writes results to a target table automatically.
A chained task graph where a second Task runs after the first completes, producing a daily revenue summary you can query on demand.
Secret Mission: Add a third task that builds a top-products leaderboard, triggered after the reporting task completes.

Are there any prerequisites?

You need basic SQL knowledge (SELECT, INSERT, JOIN, GROUP BY) and an existing Snowflake account. If you do not have a Snowflake account yet, sign up for a free trial at signup.snowflake.com. The trial gives you $400 in credits, which is more than enough for this project.

Before We Start

Before we dive in, let's lock in what you're about to build and why it matters.

This project gives you hands-on experience with the exact Snowflake features that come up in data engineering interviews: Streams, Tasks, incremental processing, and task chaining.

Set Up Your Pipeline Infrastructure

Every pipeline needs a home. In Snowflake, a database and schema give your tables, streams, and tasks their own isolated namespace.

In this step, you will create a dedicated database, schema, and virtual warehouse for the project. Then you will build a raw events table and fill it with sample e-commerce orders so you have real data to work with.

In this step, get ready to:

Create a database, schema, and XS warehouse for the project.
Create the raw events table with columns for order data.
Insert a batch of sample order events and verify the data loaded correctly.

Create a database, schema, and warehouse

All SQL in this project runs inside a Snowsight worksheet. You need to open one before writing any code.

In the Snowsight navigation menu, select Projects, then Worksheets.
Click the + button and select SQL Worksheet.
A blank worksheet opens. Confirm the role in the top-left context selector is set to ACCOUNTADMIN.

Why ACCOUNTADMIN?

On a trial account, ACCOUNTADMIN has all the privileges needed to create databases, warehouses, streams, and tasks. In a production environment you would use a more restricted role, but for learning purposes this keeps things simple.

Create the project infrastructure by pasting the following SQL into your worksheet and running it with Cmd+Shift+Return (macOS) or Ctrl+Alt+Enter (Windows) to run all statements:

-- Create project infrastructure
CREATE OR REPLACE DATABASE pipeline_db;
CREATE OR REPLACE SCHEMA pipeline_db.pipeline_schema;
USE DATABASE pipeline_db;
USE SCHEMA pipeline_schema;

CREATE OR REPLACE WAREHOUSE pipeline_wh
  WAREHOUSE_SIZE = 'XSMALL'
  AUTO_SUSPEND = 60
  AUTO_RESUME = TRUE;

USE WAREHOUSE pipeline_wh;

What does this code do?

CREATE OR REPLACE DATABASE pipeline_db creates a brand-new database to hold all project objects.
CREATE OR REPLACE SCHEMA pipeline_db.pipeline_schema adds a schema inside that database to keep tables and streams organized.
The USE statements set the session context so every subsequent command targets this database and schema.
The warehouse is sized XSMALL (the smallest available) with AUTO_SUSPEND = 60 so it shuts down after 60 seconds of inactivity, saving credits.
AUTO_RESUME = TRUE means it restarts automatically when the next query arrives.

You should see a success status for each statement in the results pane at the bottom of the worksheet.

Seeing a permission error?

Make sure the role selector in your worksheet is set to ACCOUNTADMIN. Other roles may lack CREATE DATABASE or CREATE WAREHOUSE privileges.
If you see a message about an existing object, the OR REPLACE clause handles that safely. Re-run the statements.

Help me troubleshoot a permission error when creating my Snowflake database and warehouse.

Create the raw events table

The pipeline needs a landing table where raw order events arrive. Each row represents one event in an order lifecycle (created, paid, shipped).

Add the table definition to your worksheet below the previous statements, then run it:

-- Create raw events table
CREATE OR REPLACE TABLE raw_orders (
  event_id STRING,
  event_time TIMESTAMP_NTZ,
  user_id STRING,
  order_id STRING,
  event_type STRING,
  product_id STRING,
  amount NUMBER(10,2)
);

What does this table hold?

event_id uniquely identifies each event.
event_time uses TIMESTAMP_NTZ (no timezone) to record when the event happened.
event_type tracks the lifecycle stage: created, paid, or shipped.
amount stores the order value as a decimal with two decimal places.

You should see a success status confirming the table was created.

Table not created?

Make sure the USE DATABASE pipeline_db and USE SCHEMA pipeline_schema statements from the previous chunk ran successfully. Without them, the table lands in the wrong location.
Check for missing commas between column definitions.

Help me debug why my CREATE TABLE statement failed in Snowflake.

Insert sample order events and verify the data

With the table in place, you can load a first batch of simulated e-commerce events. These represent order activity from multiple users and products.

Insert 10 sample rows by adding and running this statement:

-- Insert initial batch of sample data
INSERT INTO raw_orders VALUES
  ('e001', '2026-08-10 09:01:00', 'u101', 'ord001', 'created', 'prod_a', 29.99),
  ('e002', '2026-08-10 09:05:00', 'u102', 'ord002', 'created', 'prod_b', 49.99),
  ('e003', '2026-08-10 09:10:00', 'u101', 'ord001', 'paid', 'prod_a', 29.99),
  ('e004', '2026-08-10 09:15:00', 'u103', 'ord003', 'created', 'prod_c', 19.99),
  ('e005', '2026-08-10 09:20:00', 'u102', 'ord002', 'paid', 'prod_b', 49.99),
  ('e006', '2026-08-10 10:01:00', 'u104', 'ord004', 'created', 'prod_a', 29.99),
  ('e007', '2026-08-10 10:05:00', 'u103', 'ord003', 'paid', 'prod_c', 19.99),
  ('e008', '2026-08-10 10:10:00', 'u104', 'ord004', 'paid', 'prod_a', 29.99),
  ('e009', '2026-08-10 10:15:00', 'u101', 'ord001', 'shipped', 'prod_a', 29.99),
  ('e010', '2026-08-10 10:20:00', 'u105', 'ord005', 'created', 'prod_b', 49.99);

What does this data represent?

These 10 events simulate a stream of e-commerce activity. Five different users place orders for three products (prod_a, prod_b, prod_c). Each order moves through lifecycle stages: created, paid, and sometimes shipped.

Later steps will add more batches to simulate real-time data arriving over time.

Before you verify, how many rows do you expect the count query to return?

Verify the data loaded by running this query:

SELECT COUNT(*) FROM raw_orders;

You should see the result 10 in the results pane.

Preview the actual rows by running:

SELECT * FROM raw_orders LIMIT 5;

You should see five rows with columns for event_id, event_time, user_id, order_id, event_type, product_id, and amount.

Seeing 0 rows?

Confirm the INSERT statement ran without errors. Look for a row count of 10 in its result.
Check that your session context still points to pipeline_db.pipeline_schema. If you opened a new worksheet tab, the context resets.

Help me figure out why my table is empty after running INSERT.

✔️ Awesome, I've got everything!

Great. Make sure you have kept the worksheet open. You will use it for the rest of the project.

ⓧ I'd like to double check the full code

Scroll up in your worksheet and confirm these three sections appear in order:

Infrastructure block: CREATE OR REPLACE DATABASE, CREATE OR REPLACE SCHEMA, the two USE statements, CREATE OR REPLACE WAREHOUSE, and USE WAREHOUSE.
Table creation: CREATE OR REPLACE TABLE raw_orders with seven columns.
Data insert: INSERT INTO raw_orders VALUES with 10 rows (e001 through e010).

If any section is missing, copy it from the chunks above and re-run.

Your pipeline workspace is ready. You have a database, warehouse, and table loaded with real order data. Next up, you will write a manual summary query and discover why manual processing breaks the moment new data arrives.

Load Data and See It Break Without Automation

Your infrastructure is set up and your raw_orders table has 10 rows of order events. Now it's time to do something useful with that data.

Manually querying raw data to produce reports works fine for a small batch. But the moment new data arrives, you have to re-run everything from scratch with no way to know which rows are new and which have already been processed. This step makes you feel that pain firsthand.

In this step, get ready to:

Write a manual daily revenue summary query.
Insert a second batch of order events to simulate new data arriving.
Re-run the summary and discover the limitation of this manual approach.

Write a daily revenue summary query

A common reporting task is summarising revenue by event type and date. This tells a business how much money flowed through each stage of the order lifecycle on a given day.

In your Snowsight worksheet, run this query to produce a daily revenue summary from your existing 10 rows:

-- Manual daily revenue summary
SELECT
  DATE(event_time) AS event_date,
  event_type,
  COUNT(*) AS event_count,
  SUM(amount) AS total_amount
FROM raw_orders
GROUP BY event_date, event_type
ORDER BY event_date, event_type;

What does this query do?

DATE(event_time) extracts just the date from each timestamp, so all events on the same day group together.

GROUP BY event_date, event_type splits results into buckets like "created on Aug 10" and "paid on Aug 10". SUM(amount) then totals the revenue within each bucket.

You should see a results table with three rows, one for each event type (created, paid, shipped) all on the date 2026-08-10.

Seeing zero rows or an error?

Make sure your session context is set. Run USE DATABASE pipeline_db; and USE SCHEMA pipeline_schema; if you opened a new worksheet.
Confirm data exists by running SELECT COUNT(*) FROM raw_orders; which should return 10.

Help me troubleshoot my Snowflake summary query

Insert a new batch of order events

In a real pipeline, new data arrives constantly. Let's simulate that by inserting 5 more order events, as if a second batch just landed from your e-commerce platform.

Insert 5 new order events by running this statement:

-- Insert second batch (simulating new data arriving)
INSERT INTO raw_orders VALUES
  ('e011', '2026-08-10 11:01:00', 'u106', 'ord006', 'created', 'prod_d', 79.99),
  ('e012', '2026-08-10 11:05:00', 'u105', 'ord005', 'paid', 'prod_b', 49.99),
  ('e013', '2026-08-10 11:10:00', 'u106', 'ord006', 'paid', 'prod_d', 79.99),
  ('e014', '2026-08-10 11:15:00', 'u107', 'ord007', 'created', 'prod_a', 29.99),
  ('e015', '2026-08-10 11:20:00', 'u102', 'ord002', 'shipped', 'prod_b', 49.99);

What does this simulate?

These 5 rows represent a fresh batch of activity: two new orders created, two payments processed, and one order shipped. In production, this data would arrive from an application database or event stream throughout the day.

Confirm the insert worked by running:

SELECT COUNT(*) FROM raw_orders;

You should see 15 as the result. Your table now holds both the original 10 rows and the 5 new ones.

Still seeing 10 rows?

Check that you ran the INSERT statement, not just the SELECT COUNT. Look for the success message above your results panel.
Make sure you selected and ran the full INSERT (all 5 value rows). If you highlighted only part of the statement, Snowsight runs just the selected text.

Help me figure out why my INSERT didn't work

Re-run the summary and notice the gap

Your table now has 15 rows. The report you ran earlier is stale. Time to update it.

Before you re-run the summary query, think about this: once you see the updated results, will there be any way to tell which 5 rows are new versus the 10 you already processed?

Run the same summary query again:

-- Manual daily revenue summary
SELECT
  DATE(event_time) AS event_date,
  event_type,
  COUNT(*) AS event_count,
  SUM(amount) AS total_amount
FROM raw_orders
GROUP BY event_date, event_type
ORDER BY event_date, event_type;

The totals are higher now. The created event type shows more events and more revenue, paid has grown too, and shipped has an extra row. But look at the results carefully.

Where is the problem?

Nothing in this output tells you which rows are new. The query rescans all 15 rows every time. If you needed to process only the new batch into a cleaned table, you would have no way to isolate those 5 rows without tracking them yourself.

This is the core problem of manual reporting: every run is a full rescan. As the table grows to millions of rows, this becomes slow, wasteful, and error-prone.

You just felt the pain of manual reporting. Every time new data lands, you re-run the full query with no idea what changed. In the next step, you will create a Stream that solves this problem by automatically tracking which rows are new.

Create a Stream to Track Changes Automatically

In the last step, you had to manually re-run your summary query every time new data arrived. You also had no way to tell which rows were new versus already processed.

A Stream solves this problem. It watches your raw table and maintains an offset so downstream processing reads only unprocessed changes. No more rescanning the full table.

In this step, get ready to:

Create a standard Stream on the raw_orders table.
Query the Stream to understand its metadata columns.
Insert new rows and verify the Stream captures only those changes.

Create a Stream on your raw orders table

A Stream in Snowflake is a change data capture object. It sits on top of a table and automatically tracks every INSERT, UPDATE, and DELETE that happens after the Stream is created.

Create the Stream by running this SQL in your Snowsight worksheet:

-- Create a Stream on the raw_orders table
CREATE OR REPLACE STREAM raw_orders_stream ON TABLE raw_orders;

What does this do?

This creates a standard Stream called raw_orders_stream that watches the raw_orders table.

The Stream sets its offset to the current version of the table. From this moment on, it tracks every new change (INSERT, UPDATE, DELETE) that happens to raw_orders.

Confirm the Stream was created by running:

-- Check the Stream exists
SELECT * FROM raw_orders_stream;

You should see an empty result set. This is expected because the Stream only tracks changes that happen after creation.

Why is the Stream empty?

The Stream's offset starts at the current table version. The 15 rows already in raw_orders existed before the Stream was created, so they don't appear as changes.

Think of it like starting a stopwatch now. You only measure what happens after you press start, not what happened before.

Seeing an error instead of empty results?

Make sure you're using the pipeline_wh warehouse and the pipeline_schema schema. Run USE SCHEMA pipeline_db.pipeline_schema; and try again.

help me fix my Stream creation error in Snowflake

Insert new rows and watch the Stream capture them

Now you'll simulate a third batch of order events arriving. The Stream should detect these new rows automatically.

Insert 3 new order events by running:

-- Insert a third batch to see the Stream capture new rows
INSERT INTO raw_orders VALUES
  ('e016', '2026-08-10 12:01:00', 'u108', 'ord008', 'created', 'prod_c', 19.99),
  ('e017', '2026-08-10 12:05:00', 'u107', 'ord007', 'paid', 'prod_a', 29.99),
  ('e018', '2026-08-10 12:10:00', 'u108', 'ord008', 'paid', 'prod_c', 19.99);

What does this batch represent?

These 3 rows simulate new order events arriving in your system. User u108 creates and pays for an order, and user u107 pays for a previously created order.

Before you query the Stream, predict what you'll see. Will it show all 18 rows in the table, or just the 3 you inserted after creating the Stream?

Query the Stream to see what it captured:

-- View stream contents
SELECT METADATA$ACTION, METADATA$ISUPDATE, * FROM raw_orders_stream;

You should see exactly 3 rows, each with METADATA$ACTION showing INSERT and METADATA$ISUPDATE showing FALSE.

What are these metadata columns?

METADATA$ACTION tells you what happened: INSERT for new rows, DELETE for removed rows.
METADATA$ISUPDATE tells you if the action is part of an UPDATE operation (shown as a DELETE + INSERT pair).
Since you only inserted new rows, every row shows INSERT and FALSE.

Why only 3 rows and not 18?

The Stream only tracks changes that happened after it was created. The 15 rows that existed before the Stream was created are invisible to it.

This is the key advantage. Downstream tasks can read the Stream and process only the new data without rescanning the entire table.

Verify the Stream tracks only new changes

Before you move on, confirm the Stream count and that querying alone does not advance the offset.

Run this count query to verify:

SELECT COUNT(*) FROM raw_orders_stream;

You should see 3 returned. These are all the unconsumed rows the Stream is tracking.

Run the metadata query one more time:

SELECT METADATA$ACTION, * FROM raw_orders_stream LIMIT 3;

You still see the same 3 rows. Querying a Stream does not advance its offset. The data stays available until a DML transaction consumes it.

What does 'consuming' the Stream mean?

A Stream's offset only advances when you use the Stream inside a DML statement (like INSERT INTO ... SELECT FROM stream). Simply reading the Stream with a SELECT leaves the offset unchanged.

In the next step, you'll create a Task that consumes the Stream automatically, advancing the offset so the same rows are never processed twice.

Stream showing 0 rows after the insert?

Make sure you ran the INSERT statement in the same worksheet session where you created the Stream. If you opened a new worksheet, run USE DATABASE pipeline_db; and USE SCHEMA pipeline_schema; first to set the correct context.

help me figure out why my Snowflake Stream is empty

Your Stream is now watching raw_orders and knows exactly which rows are new. Next up, you'll build a Task that automatically reads those changes, cleans the data, and advances the offset so nothing is ever processed twice.

Build a Cleaning Task That Consumes the Stream

Your Stream is now tracking every new row that lands in raw_orders. The next question is: who reads those changes and does something useful with them?

In this step, you'll create a Task that fires on a schedule, reads only the unprocessed rows from the Stream, cleans them, and writes them to a target table. Once the Task consumes the Stream in a DML transaction, the offset advances automatically so the same rows are never processed twice.

In this step, get ready to:

Create a cleaned_orders target table for processed data.
Build a scheduled Task that reads from the Stream and cleans the data.
Resume the Task and verify it processes all 18 pending rows.

Create the cleaned_orders target table

Before the Task can write anywhere, it needs a destination. The cleaned_orders table mirrors the raw table's columns but adds a processed_at timestamp so you can track when each row was processed.

Create the target table by running this SQL in your Snowsight worksheet:

-- Create cleaned orders target table
CREATE OR REPLACE TABLE cleaned_orders (
  event_id STRING,
  event_time TIMESTAMP_NTZ,
  user_id STRING,
  order_id STRING,
  event_type STRING,
  product_id STRING,
  amount NUMBER(10,2),
  processed_at TIMESTAMP_NTZ DEFAULT CURRENT_TIMESTAMP()
);

What does this table do?

The cleaned_orders table stores processed rows after the Task cleans them. Its columns match raw_orders except for the extra processed_at column, which defaults to CURRENT_TIMESTAMP() so every row records the moment it was cleaned.

Confirm the table was created by running:

SELECT COUNT(*) FROM cleaned_orders;

You should see 0 rows. The table is empty and ready to receive cleaned data.

Seeing an error about the table?

Make sure your worksheet context is set to pipeline_db and pipeline_schema. Check the database and schema dropdowns at the top of your Snowsight worksheet.

Help me troubleshoot my cleaned_orders table creation

Create a Task with a WHEN condition

A Snowflake Task is a scheduled SQL statement that runs automatically. The key feature here is the WHEN clause, which checks whether the Stream has new data before the Task actually fires. If there is nothing new, the Task skips execution and consumes zero credits.

Create the cleaning task by running this SQL:

-- Create the cleaning task
CREATE OR REPLACE TASK clean_task
  WAREHOUSE = pipeline_wh
  SCHEDULE = '1 MINUTES'
  WHEN SYSTEM$STREAM_HAS_DATA('raw_orders_stream')
AS
  INSERT INTO cleaned_orders (event_id, event_time, user_id, order_id, event_type, product_id, amount)
  SELECT
    event_id,
    event_time,
    user_id,
    order_id,
    LOWER(TRIM(event_type)),
    product_id,
    amount
  FROM raw_orders_stream
  WHERE METADATA$ACTION = 'INSERT';

What does this task do?

WAREHOUSE = pipeline_wh tells Snowflake which compute to use when the Task runs.
SCHEDULE = '1 MINUTES' sets the Task to check every minute. In production you would use a longer interval.
WHEN SYSTEM$STREAM_HAS_DATA('raw_orders_stream') prevents execution when the Stream is empty, saving credits.
LOWER(TRIM(event_type)) normalizes the event type string. This is the "cleaning" step of the pipeline.
WHERE METADATA$ACTION = 'INSERT' filters to only new inserts, ignoring updates or deletes.

Verify the task exists by running:

SHOW TASKS;

You should see clean_task listed with a state of suspended. Tasks always start suspended when first created.

Task not appearing in SHOW TASKS?

Confirm your worksheet is using the pipeline_wh warehouse and your role has CREATE TASK privileges. On a trial account with the ACCOUNTADMIN role, you have all necessary privileges.

Help me debug why my task is not showing up

Resume the Task and verify the pipeline fired

The Task exists but is suspended. You need to explicitly resume it before Snowflake will start scheduling runs.

Resume the Task by running:

-- Resume the task
ALTER TASK clean_task RESUME;

What happens next?

Snowflake will now check every minute whether raw_orders_stream has data. Since it currently holds 18 unconsumed rows, the Task will fire on its next scheduled run and insert all 18 cleaned rows into cleaned_orders.

Wait approximately 60 seconds for the Task to fire on its next scheduled interval.

Before you check the results, what do you think you'll find in cleaned_orders? And what will the Stream look like after the Task consumes it?

Check how many rows the Task processed by running:

SELECT COUNT(*) FROM cleaned_orders;

You should see 18 rows. That matches all the unconsumed rows that were waiting in the Stream.

Now confirm the Stream offset advanced by running:

SELECT * FROM raw_orders_stream;

The result is empty. The Task consumed all 18 rows in a DML transaction, which advanced the Stream's offset past them. Those rows will never be reprocessed.

Still seeing 0 rows in cleaned_orders?

The Task may not have fired yet. Wait another 60 seconds and try the SELECT COUNT(*) query again. You can also check the task's last run by running SHOW TASKS and looking at the state column to confirm it shows started.

If the state still shows suspended, run ALTER TASK clean_task RESUME; again.

Help me figure out why my task has not fired

✔️ Awesome, I've got everything!

Great. Your cleaning Task is live and has already processed its first batch. Confirm that cleaned_orders has 18 rows and raw_orders_stream is empty.

ⓧ I'd like to double check the full code

Here are the three SQL blocks you ran in this step, in order:

The CREATE OR REPLACE TABLE cleaned_orders block with 8 columns including processed_at.
The CREATE OR REPLACE TASK clean_task block with SCHEDULE = '1 MINUTES', the WHEN condition, and the INSERT...SELECT body.
ALTER TASK clean_task RESUME; to start the schedule.

Scroll up and confirm each block is present in your worksheet and that you ran all three.

Your cleaning Task is humming along, automatically picking up new rows and writing clean results. Next up, you'll chain a second Task that aggregates your cleaned data into a daily sales summary, building the reporting layer of your pipeline.

Chain a Reporting Task for Daily Sales Analytics

Your cleaning task is automatically processing every new row that lands in raw_orders. The cleaned data is there, but answering "how much revenue per event type per day?" still requires writing and re-running a manual query every time.

In this step you will create a second Task that is chained to clean_task using the AFTER keyword. Every time the cleaning task finishes, Snowflake automatically fires the reporting task to aggregate data into a daily_sales_summary table.

In this step, get ready to:

Create a daily_sales_summary table to hold aggregated results.
Create a child task that runs AFTER the cleaning task and uses MERGE to upsert aggregated data.
Resume the task, insert a new batch of raw events, and watch the full pipeline fire end to end.

Create the daily sales summary table

Before the reporting task can write anywhere, you need a destination table to hold the aggregated revenue data.

Create the daily_sales_summary table by running this query in your Snowsight worksheet:

-- Create daily sales summary table
CREATE OR REPLACE TABLE daily_sales_summary (
  event_date DATE,
  event_type STRING,
  event_count NUMBER,
  total_revenue NUMBER(12,2),
  last_updated TIMESTAMP_NTZ DEFAULT CURRENT_TIMESTAMP()
);

What does this table store?

event_date and event_type form the composite key. Each unique date-plus-type combination gets one row.
event_count tracks how many events occurred for that combination.
total_revenue sums the dollar amounts.
last_updated timestamps when the row was last refreshed, so you always know how fresh the data is.

Confirm the table was created by running:

SHOW TABLES LIKE 'daily_sales_summary';

You should see one row listing daily_sales_summary in the results.

Don't see the table listed?

Make sure you are still using the correct context. Run USE DATABASE pipeline_db; and USE SCHEMA pipeline_schema; if needed.
Check for typos in the table name. The name must be exactly daily_sales_summary.

Help me debug why my table did not appear in SHOW TABLES.

Create the reporting task chained after the cleaning task

The AFTER clause tells Snowflake to run this task only after its predecessor completes successfully. This creates a task graph where clean_task is the root and your new task is its child.

Create summary_task by running this query:

-- Create the reporting task chained after clean_task
CREATE OR REPLACE TASK summary_task
  WAREHOUSE = pipeline_wh
  AFTER clean_task
AS
  MERGE INTO daily_sales_summary tgt
  USING (
    SELECT
      DATE(event_time) AS event_date,
      event_type,
      COUNT(*) AS event_count,
      SUM(amount) AS total_revenue
    FROM cleaned_orders
    GROUP BY event_date, event_type
  ) src
  ON tgt.event_date = src.event_date AND tgt.event_type = src.event_type
  WHEN MATCHED THEN UPDATE SET
    tgt.event_count = src.event_count,
    tgt.total_revenue = src.total_revenue,
    tgt.last_updated = CURRENT_TIMESTAMP()
  WHEN NOT MATCHED THEN INSERT (event_date, event_type, event_count, total_revenue)
    VALUES (src.event_date, src.event_type, src.event_count, src.total_revenue);

What does this task do?

This task uses a MERGE statement instead of a plain INSERT. MERGE compares incoming rows against existing rows in daily_sales_summary using the composite key of event_date and event_type.

If a matching row already exists (WHEN MATCHED), the task updates the count and revenue to the latest totals.
If no matching row exists (WHEN NOT MATCHED), the task inserts a brand-new row.
The AFTER clean_task clause means this task has no schedule of its own. It fires automatically whenever clean_task completes.

Verify the task was created by running:

SHOW TASKS;

You should see two tasks listed: clean_task with state started and summary_task with state suspended. The new task starts suspended because all tasks are created in a suspended state by default.

Seeing an error on CREATE TASK?

If you see an error about the predecessor task, confirm clean_task exists by checking the SHOW TASKS output from above.
Make sure the warehouse name matches exactly: pipeline_wh.

Help me fix my CREATE TASK summary_task statement.

Resume the task and trigger the full pipeline

With both tasks in place, you need to resume summary_task so it is ready to fire when clean_task finishes. Then you will insert new raw data so the Stream picks it up, triggering the entire chain.

Resume the child task by running:

-- Resume the child task
ALTER TASK summary_task RESUME;

Confirm both tasks are now active by running:

SHOW TASKS;

Both clean_task and summary_task should show state started in the output.

Insert a new batch of raw order events to trigger the pipeline:

-- Insert new data to trigger the pipeline
INSERT INTO raw_orders VALUES
  ('e019', '2026-08-11 08:01:00', 'u109', 'ord009', 'created', 'prod_b', 49.99),
  ('e020', '2026-08-11 08:05:00', 'u110', 'ord010', 'created', 'prod_a', 29.99),
  ('e021', '2026-08-11 08:10:00', 'u109', 'ord009', 'paid', 'prod_b', 49.99);

What happens next?

These three rows land in raw_orders and immediately appear in raw_orders_stream. Within the next minute, clean_task checks the Stream, finds new data, cleans it into cleaned_orders, and finishes. That completion triggers summary_task, which MERGEs aggregated results into daily_sales_summary.

Wait approximately 60-90 seconds for the task chain to fire.
Before you query the results, think about this: the new batch uses the date 2026-08-11 while all previous data uses 2026-08-10. How many distinct dates do you expect to see in the summary table?

Check the results by running:

SELECT * FROM daily_sales_summary ORDER BY event_date DESC;

You should see rows for both 2026-08-11 and 2026-08-10, with aggregated revenue per event type per day reflecting all batches loaded so far.

What should I see?

The results show each unique combination of date and event type as one row. The 2026-08-11 rows come from the batch you just inserted. The 2026-08-10 rows reflect all 18 earlier events. Every row has a last_updated timestamp showing when the task last refreshed it.

Table is empty after waiting?

Confirm both tasks are active by running SHOW TASKS; and checking that both show state started.
Check whether the Stream has data by running SELECT COUNT(*) FROM raw_orders_stream;. If it returns rows, the cleaning task has not fired yet. Wait another minute and re-check.
If the Stream is empty but daily_sales_summary is also empty, check cleaned_orders has rows. The summary task aggregates from cleaned_orders, not from the Stream.

Help me figure out why daily_sales_summary is empty.

✔️ Awesome, I've got everything!

Great. Confirm you have saved all your work in the worksheet.

ⓧ I'd like to double check the full code

Scroll up through this step and confirm you ran each of these code blocks in order:

CREATE OR REPLACE TABLE daily_sales_summary (the target table).
CREATE OR REPLACE TASK summary_task with the MERGE body (the chained reporting task).
ALTER TASK summary_task RESUME; (activating the child task).
INSERT INTO raw_orders VALUES with events e019 through e021 (the trigger batch).
SELECT * FROM daily_sales_summary ORDER BY event_date DESC; (verification query).

Your pipeline now runs fully automatically: raw data lands, the Stream captures it, clean_task cleans it, and summary_task rolls it up into a business-ready report. Next up, you will extend this pipeline even further in a bonus challenge.

Secret mission

Build a Top Products Leaderboard

Your pipeline cleans data and builds daily summaries. But what if stakeholders want a live leaderboard of top-selling products? Add a third task to the chain and prove you can build deeper task graphs.

Clean Up Your Resources

Decide whether to keep your resources running, pause them to come back later, or delete them entirely. The pipeline_wh warehouse auto-suspends after 60 seconds of inactivity, so cost is minimal. Tasks consume trial credits only when they fire.

Cost warning

Your Snowflake trial comes with $400 in free credits, and this project uses very little of that balance. However, resumed tasks continue to fire every minute (consuming a small amount of credits each time) for as long as new data appears in the stream. If you are done with the project, suspend or delete your tasks to preserve your trial balance for other experiments.

Resources you used:

pipeline_db database (contains all tables, streams, and tasks).
pipeline_wh virtual warehouse (XSMALL, auto-suspends after 60 seconds).
3 scheduled tasks: clean_task, summary_task, and leaderboard_task (Secret Mission only). These live inside pipeline_db and consume credits each time they run.

Keep everything running

No action needed. Choose this if you are still actively experimenting with the pipeline or plan to add more data batches soon.

Your tasks will continue to fire every minute whenever new data lands in raw_orders.
The warehouse auto-suspends after 60 seconds of inactivity, so you only pay for the seconds each task run actually uses.
Monitor your credit usage in Snowsight under Admin > Cost Management.

Pause - I'll come back to this later

Suspend your tasks to stop scheduled runs while keeping all objects intact. You can resume them later with a single command.

Suspend the root task to stop the entire chain from firing. Run this in a Snowsight worksheet:

-- Suspending the root task stops the entire chain
ALTER TASK clean_task SUSPEND;
ALTER TASK summary_task SUSPEND;
ALTER TASK leaderboard_task SUSPEND;

What if I did not complete the Secret Mission?

If you skipped the Secret Mission, you do not have a leaderboard_task. Just run the first two ALTER TASK ... SUSPEND lines and ignore the third.

All tables, streams, and data remain in place. When you are ready to resume, run ALTER TASK clean_task RESUME; (and resume child tasks before the root).

Delete - I don't want to use this again

Remove all project resources and stop all costs. Dropping the database removes every object inside it (tables, streams, tasks, and schemas). Dropping the warehouse removes the compute resource.

Run the following commands in a Snowsight worksheet to delete everything:

-- Remove all objects (tables, streams, tasks, schemas)
DROP DATABASE pipeline_db;

-- Remove the warehouse
DROP WAREHOUSE pipeline_wh;

After running these two commands, all tables, streams, tasks, and the warehouse are gone. No further credits will be consumed by this project.
Your Snowflake account itself remains active for the rest of your trial period. No account deletion is needed.

Nice Work!

Nice work! You just built a fully automated ELT pipeline in Snowflake that incrementally processes order events from raw ingestion through cleaning to analytics, all without manual intervention.

Here's what you accomplished:

Built a Stream that automatically tracks new rows landing in your raw table, eliminating full-table rescans and giving downstream processes a clean offset of only unprocessed changes.
Created a scheduled Task that consumes Stream data on a 1-minute cadence, cleans it, and advances the Stream offset so the same rows are never processed twice.
Chained a second Task using AFTER to build a daily sales summary from cleaned data, proving you can orchestrate multi-step pipelines natively inside Snowflake.
(Secret Mission) Added a third leaderboard Task chained after the reporting task, producing a top-products ranking and demonstrating a 3-level task graph with MERGE logic.

Ready to quiz yourself?