Exercise: Streaming

YellowLine NYC story · full hands-on lab

title: “Exercise: Streaming” subtitle: “YellowLine NYC story · full hands-on lab” —

Estimated time: 50–60 min (Databricks: 25 min · Snowflake relay + Dynamic Tables: 20 min · dbt: 10 min · Power BI: 5 min)

YellowLine NYC context (Module 8)

YellowLine NYC live dispatch in the story; Aiven user-activity in the lab.

NoteModule prerequisites

Modules 2–3 required · Module 4 recommended for the dbt dynamic_table track. Lab dataset: Aiven Kafka user-activity (YellowLine NYC taxi streaming in the story animation).

TipWorking Environment

Use GitHub Codespaces for a ready-to-use environment — core and optional lab packages are pre-installed from pyproject.toml (including [streaming], [ml], and [notebook] extras). Open your fork on GitHub → Code → Codespaces → Create codespace on master. Streaming notebooks run in Databricks (Maven libraries on the cluster); relay consumer, dbt, and Snowflake scripts run in the Codespace terminal.

The normal path is fork + Codespace (Prerequisites § Step 2). Use Lab source files only if your facilitator approved it — e.g. you cannot create or use GitHub before class.

NoteOptional Module

This exercise is part of Module 8 (Optional). Your facilitator starts the Aiven User Activity generator (events flowing in Aiven Kafka) before you begin.

Prerequisites

  • Modules 2 and 3 completed (Databricks and Snowflake batch pipelines working)
  • Facilitator has confirmed: Aiven Kafka topic user-activity has events flowing
  • Two Kafka libraries installed on your Databricks cluster (Runtime 16.4 LTS or 15.4 LTS — both ship Spark 3.5.x; match Maven versions below):
    • org.apache.spark:spark-sql-kafka-0-10_2.12:3.5.0
    • org.apache.spark:spark-avro_2.12:3.5.0
  • Databricks cluster running and attached for streaming notebooks (Exercise: Databricks § cluster)
  • DE_WORKSHOP_ROLE + DE_WORKSHOP_WH Started in Snowsight for Snowflake relay / Dynamic Table / Power BI (workshop role)
  • Module 8 streaming setup scripts (streaming/snowflake/01_setup_streaming.sql) run as ACCOUNTADMIN, then switch back to DE_WORKSHOP_ROLE
  • Aiven SSL credentials are in Databricks secrets (workshop-scope):
    • workshop-scope/aiven-bootstrap-servers, workshop-scope/aiven-ca-cert, workshop-scope/aiven-client-cert, workshop-scope/aiven-client-key, workshop-scope/aiven-topic
Note.env check (dbt stretch only)

Databricks streaming and Snowflake Dynamic Tables in Snowsight do not use .env.

For the dbt dynamic_table stretch (Codespace terminal), reuse the same .env as Module 4:

bash .devcontainer/setup-environment.sh

All variables ✅? Continue. Any ❌? Complete Exercise: dbt § Configure .env first (cp .env.template .env at repo root → fill values → run the script again).

Base Exercise — Databricks Structured Streaming

Step 1: Run the Bronze streaming notebook

  1. In your Databricks Git folder (sidebar (left) → Workspace), open streaming/databricks/01_streaming_bronze.py
  2. Set ATTENDEE_ID = "de_XX_yourname" at the top
  3. Run all cells in order
  4. When you reach the writeStream cell — the streaming query starts and keeps running
  5. Below the cell, look for the streaming progress bar showing query status and rows processed (indicates the query is actively streaming)

Unlike the batch pipelines in Modules 2–4 (which read a fixed dataset, transform it, and exit), a streaming query processes data as it arrives. The writeStream call starts a long-running micro-batch loop: every ~10 seconds it checks for new Kafka events, processes them, and appends the results to the Bronze table. The query only stops when you call q.stop() or the cluster terminates.

Reference: Structured Streaming Programming Guide

Verify (run in a new cell or separate notebook):

-- Run this a few times — count should increase
SELECT
    COUNT(*)                         AS total_events,
    COUNT(DISTINCT country)          AS distinct_countries,
    MAX(bronze_processing_timestamp) AS last_processed_at
FROM {attendee_id}_streaming.user_activity_bronze

Expected: rows increasing every ~10 seconds as new events arrive.

Step 2: Run the Silver streaming notebook

  1. In your Git folder, open streaming/databricks/02_streaming_silver.py
  2. Set your ATTENDEE_ID
  3. Run all cells — a second streaming query starts (Bronze → Silver)

Late-arriving events are inevitable in real networks — a user activity event might arrive 5 minutes after it happened. The watermark tells Spark: “I expect events within 10 minutes of their timestamp; anything older can be dropped.” Without it, Spark would keep state for every possible window forever, and memory would grow unbounded. Snowflake Dynamic Tables handle this internally — no watermark concept exists.

Reference: Watermarking in Structured Streaming

Verify:

SELECT
    action_category,
    COUNT(*) AS count
FROM {attendee_id}_streaming.user_activity_silver
GROUP BY action_category
ORDER BY count DESC

You should see conversion, engagement, browse, and other categories appearing.

Step 3: Run the Gold streaming notebook (facilitator demo / advanced)

Often demonstrated live by the facilitator. Advanced attendees can run it themselves.

  1. In your Git folder, open streaming/databricks/03_streaming_gold.py
  2. Run to start the windowed aggregation query

Observe the live KPI (refresh every 30 seconds):

SELECT
    country,
    SUM(event_count)   AS total_events,
    SUM(purchases)     AS total_purchases,
    MAX(unique_users)  AS peak_unique_users
FROM {attendee_id}_streaming.activity_by_country_5min
WHERE window_start >= CURRENT_TIMESTAMP() - INTERVAL 1 HOUR
GROUP BY country
ORDER BY total_events DESC
LIMIT 10

Discussion question: Which country generates the most purchase events? Is there a strong correlation between view count and purchase count?

Base Exercise — Snowflake Dynamic Tables

Step 1: Create the Bronze table

  1. Projects → Workspaces → + → SQL File — see Exercise: Snowflake § run SQL
  2. Open streaming/snowflake/01_setup_streaming.sql
  3. Run all as ACCOUNTADMIN (Ctrl+Shift+Enter) — the script loads attendee_id and sas_token from _workshop_config (set in Modules 2–3 setup). Stops with ✖ ERROR if config is missing.
  4. Switch back to DE_WORKSHOP_ROLE for verification queries below.

Verify (use your schema — replace with your attendee_id from _workshop_config, uppercased):

SHOW TABLES LIKE 'STREAMING_BRONZE%'
IN SCHEMA DE_MASTERCLASS.DE_01_ALICE_STREAMING;  -- example: your {ATTENDEE_ID}_STREAMING

The table exists but is empty — the relay consumer (facilitator-managed) will fill it via Snowpipe.

Step 2: Wait for Bronze rows to arrive

-- Poll until you see rows (may take ~1–2 minutes after relay consumer starts)
SELECT COUNT(*), MAX(ingest_ts) AS last_insert
FROM DE_MASTERCLASS.{ATTENDEE_ID}_STREAMING.STREAMING_BRONZE_USER_ACTIVITY;

Step 3: Create Silver + Gold Dynamic Tables

  1. Open streaming/snowflake/02_dynamic_tables.sql
  2. Run all as ACCOUNTADMIN — attendee_id auto-loads from _workshop_config
  3. Wait 1 minute for the first automatic refresh
  4. (Optional) The script also creates STREAMING_GOLD_TOP_PAGES (30-min page views) — facilitator demo; not required for the lab comparison table.

Dynamic Tables are pull-based — Snowflake periodically re-runs the underlying query to catch up with new data. TARGET_LAG tells Snowflake how stale the table is allowed to be. A 1-minute lag means the table refreshes approximately every minute. Shorter lag = more compute cost; longer lag = cheaper but staler. This is the streaming equivalent of a batch cron schedule, but managed entirely by Snowflake.

Reference: Dynamic Tables

Check Dynamic Table status:

SHOW DYNAMIC TABLES LIKE 'STREAMING_%'
IN SCHEMA DE_MASTERCLASS.{ATTENDEE_ID}_STREAMING;

Look at the target_lag, scheduling_state, and last_suspended_on columns.

Verify Gold is updating:

SELECT
    country,
    SUM(event_count)   AS total_events,
    SUM(purchases)     AS total_purchases
FROM DE_MASTERCLASS.{ATTENDEE_ID}_STREAMING.STREAMING_GOLD_ACTIVITY_BY_COUNTRY
GROUP BY country
ORDER BY total_events DESC
LIMIT 10;

Run this query multiple times 1 minute apart — the numbers should increase.

Base Exercise — Power BI Live Dashboard

Note

Requires Power BI Desktop installed and the Snowflake Gold Dynamic Table running (Snowflake steps above completed).

Step 1: Connect Power BI to Snowflake in DirectQuery mode

  1. Open Power BI Desktop
  2. Home → Get Data → Snowflake
  3. Fill in:
    • Server: account host from View account details (e.g. el30551.west-europe.azure.snowflakecomputing.com) — same locator as in the Snowsight URL …/west-europe.azure/el30551/…
    • Warehouse: DE_WORKSHOP_WH
    • Database: DE_MASTERCLASS
    • Data Connectivity mode: DirectQuery ← important!
  4. Navigate to schema {ATTENDEE_ID}_STREAMING
  5. Select STREAMING_GOLD_ACTIVITY_BY_COUNTRY → Load

Step 2: Build 3 visuals

Visual 1 — Top countries bar chart

  • Chart type: Clustered bar chart
  • Axis: country
  • Values: SUM(event_count)
  • Filters: Top N = 10 (by event_count)

Visual 2 — Total purchases card

  • Chart type: Card
  • Field: SUM(purchases)
  • Label: “Total purchases (last hour)”

Visual 3 — Event velocity over time

  • Chart type: Line chart
  • X-axis: window_minute
  • Y-axis: SUM(event_count)
  • Legend: country (filter to top 5)

Step 3: Enable automatic page refresh

  1. Click on an empty area of the canvas
  2. Format pane → Page refresh → On
  3. Set refresh interval: 1 minute

Watch the bar chart update as new windows close in Snowflake.

Verify the dashboard is live: - Note the current SUM(event_count) for your top country (e.g. US) - Wait 2 minutes - Hit the manual refresh button (or wait for auto-refresh) - The count should have increased

Stretch Goal — dbt Dynamic Table Materialization

Stretch: Deploy the streaming pipeline via dbt

Prerequisite: Configure .env from Exercise: dbt (dbt debug --target snowflake passing).

  1. Open dbt_project/models/streaming/streaming_silver_user_activity.sql — observe the materialized: dynamic_table config

  2. Open dbt_project/models/streaming/_streaming.yml — review source and model definitions

  3. Run from the main dbt project (Snowflake target only):

    cd dbt_project/
dbt run --target snowflake \
    --select streaming_silver_user_activity streaming_gold_activity_by_country

  4. Verify the Dynamic Tables were created:

    SHOW DYNAMIC TABLES LIKE 'STREAMING_%'
IN SCHEMA DE_MASTERCLASS.{ATTENDEE_ID}_DBT_STREAMING;

    Note: dbt materializes these into {ATTENDEE_ID}_DBT_STREAMING, isolated from the native _STREAMING schema built by 02_dynamic_tables.sql. Both pipelines can run side-by-side without overwriting each other — compare the two directly.

  5. Generate and view dbt docs:

    dbt docs generate --target snowflake
dbt docs serve

    Open the lineage graph — observe the source → silver → gold dependency chain.

Discussion question: What does dbt add compared to running the raw SQL from 02_dynamic_tables.sql directly? (Version control, lineage graph, automated tests, reusable macros, self-documenting schema)

Stretch Goal — Compare Streaming vs Batch

Stretch: Change the aggregation window

In streaming/databricks/03_streaming_gold.py, find this line:

.groupBy(
    window(col("event_timestamp"), "5 minutes", "1 minute"),
    col("country"),
)

Change the window to 1 minute tumbling (no slide):

.groupBy(
    window(col("event_timestamp"), "1 minute"),   # tumbling window
    col("country"),
)

Stop the Gold query, clear the checkpoint, and restart. Observe the difference: - How does the number of rows in the Gold table change? - How does update frequency change compared to the sliding window? - Which is more useful for a real-time dashboard?

Compare Your Results

After running exercises on both platforms, record what you observed. This anchors the theory from Module 8 to real numbers.

What you observed Databricks Structured Streaming Snowflake Dynamic Tables dbt dynamic_table
Code you ran 3 Python notebooks 2 SQL files dbt run (1 command)
How the pipeline starts Manual — run writeStream cell Automatic after CREATE DYNAMIC TABLE Automatic after dbt run
First rows appeared after ~10 seconds ~1 minute (TARGET_LAG) ~1 minute (TARGET_LAG)
Your Bronze row count after 10 min ______ ______ N/A (same Bronze table)
How you stopped it q.stop() in notebook Nothing — runs until dropped Nothing — runs until dropped
Where you tracked progress Spark UI streaming tab SHOW DYNAMIC TABLES SHOW DYNAMIC TABLES
Checkpoint / state storage ADLS2 checkpoint files Snowflake-managed (invisible) Snowflake-managed (invisible)
Did late events need handling? Yes — .withWatermark() in Silver No — Dynamic Tables handle it No
TipThings you should have noticed

1. Databricks starts immediately, Snowflake has a lag — Structured Streaming delivers new rows within seconds (micro-batch every 10 seconds). Dynamic Tables have a TARGET_LAG = '1 minute' — you had to wait before Silver and Gold populated. This is intentional: Snowflake trades latency for simplicity.

2. Databricks needed you to manage state; Snowflake didn’t — The Structured Streaming Silver notebook required .withWatermark("event_timestamp", "10 minutes") to handle late events and free state. Dynamic Tables have no equivalent concept — Snowflake manages incremental refresh internally.

3. Stopping was asymmetric — Databricks streams ran until you explicitly called q.stop(). Snowflake Dynamic Tables and dbt dynamic_table models kept refreshing until you dropped them. Neither is better — they reflect different execution models (push vs pull).

4. dbt gave you lineage for free — After dbt run, dbt docs serve showed the full lineage graph from Bronze source to Gold. Databricks and Snowflake have no equivalent without additional tooling.

5. Both Gold tables showed the same top countries — US, DE, GB, FR tend to dominate regardless of platform. The processing platform doesn’t change the data — only the latency and operational model.

Discussion questions

  1. You couldn’t stop the Snowflake Dynamic Tables by closing a browser tab — unlike Databricks notebooks. What does this mean for cost management in production?
  2. Databricks needed two Maven libraries (spark-sql-kafka-0-10_2.12:3.5.0 and spark-avro_2.12:3.5.0) installed on the cluster before it could read Kafka Avro messages. Snowflake needed only the relay consumer script. Which approach is simpler for a new team member to set up?
  3. The watermark in Databricks is 10 minutes. What would happen if a user activity event took 11 minutes to arrive (extreme network delay)? Would Snowflake Dynamic Tables handle it differently?
  4. For a dashboard refreshing every minute, which streaming approach gives the most appropriate latency?

Expected Results

After running all three Databricks notebooks for ~10 minutes:

Table Expected rows Notes
user_activity_bronze 500–2000+ Depends on Aiven generator rate and time elapsed
user_activity_silver 400–1800+ Fewer — null-country and null-user events filtered
activity_by_country_5min 50–200 rows One row per country per 5-min window; grows slowly

Top countries typically: US, DE, GB, FR, JP, BR, IN, CA, AU

Clean up

Databricks — stop all streaming queries first, then drop the schema:

for q in spark.streams.active:
    q.stop()
DROP SCHEMA IF EXISTS mhpdeworkshop_databricks_2026.{attendee_id}_streaming CASCADE;

Snowflake: The _STREAMING schema (pipe, stage, Bronze table, dynamic tables) is removed by the central cleanup script — no separate step needed.

  • Workspaces SQL file: run snowflake/sql/setup/99_cleanup.sql
  • Workspaces notebook: open 00_setup/99_cleanup and click Run all

Return to module