Today's Answer: Process Events as They Arrive

Batch jobs are excellent for historical analysis — but some decisions can't wait 30 minutes.

Structured Streaming is the same DataFrame API you already know — extended to an unbounded, continuously-arriving dataset.

Week 12 Recap

Assignment 3 (NoSQL Design + Query Federation) — was due at the end of Week 12. Check Canvas if not yet submitted.

The Batch Problem

What happens when the answer arrives too late?

The fundamental mismatch: Batch processing treats data as a bounded snapshot. The real world produces events continuously.

Streaming vs. Batch — The Core Difference

The practical boundary is blurry: Micro-batch at 5-second intervals covers most "near real-time" needs without true continuous processing complexity.

Micro-Batch vs. Continuous Processing

Rule of thumb: Use micro-batch for 99% of production workloads. The latency is low enough and the semantics are much stronger.

When to Choose Streaming

The question to ask: "What is the business cost of a 5-minute delay?" If the answer is "nothing significant" — batch is simpler.

The Unified API

The key insight: A streaming DataFrame is just a table that keeps growing.

df = spark.read.csv("input/")
result = df.groupBy("user_id").count()
result.write.parquet("output/")

df = spark.readStream.format("kafka")...
result = df.groupBy("user_id").count()
result.writeStream.format("console").start()

Same groupBy, same count, same Catalyst optimizer. The only differences: readStream instead of read, writeStream instead of write, and .start() to begin continuous execution.

Input Sources

Kafka is the correct answer for production. Socket should never leave your laptop.

raw = (spark.readStream
    .format("kafka")
    .option("kafka.bootstrap.servers", "kafka-broker:9092")
    .option("subscribe", "clicks")
    .option("startingOffsets", "latest")
    .load())

Output Sinks

query = (result.writeStream
    .format("console")
    .option("truncate", False)
    .trigger(processingTime="5 seconds")
    .start())

The Streaming Query Lifecycle

1. Driver defines the plan — readStream → transformations → writeStream
2. query.start() — query runs in a background thread
3. Each trigger interval:
   • Read new data from source (Kafka offsets, new files)
   • Run an incremental Spark job on the micro-batch
   • Write output to sink
   • Commit offsets and checkpoint state
4. On failure: restart from last checkpoint, replay Kafka offsets from last committed position — no data lost, no duplicates

query.lastProgress   # stats for the most recent micro-batch
query.status         # "is the query running / waiting / error?"
query.stop()         # graceful shutdown

Demo: Setup

# Verify Kafka and Spark are running
docker-compose ps

# From docker/ directory: reset Week 14 topics and produce one smoke-test event
make week14-reset

# Terminal 2: run automated click producer inside Docker
docker exec -it jupyter env KAFKA_BOOTSTRAP=kafka-broker:9092 \
    python /home/jovyan/week14_clicks/produce_clicks.py --rate 2

# Optional: send exactly 100 events then stop
docker exec -it jupyter env KAFKA_BOOTSTRAP=kafka-broker:9092 \
    python /home/jovyan/week14_clicks/produce_clicks.py --rate 5 --count 100

Keep the producer terminal open — Spark will continuously ingest events while it runs.

Demo: Session Setup

# streaming_demo.py — run in PySpark shell
from pyspark.sql import SparkSession
from pyspark.sql.functions import col, from_json, to_timestamp
from pyspark.sql.types import StructType, StructField, StringType

spark = SparkSession.builder \
    .appName("ClickStreamDemo") \
    .config("spark.jars.packages",
            "org.apache.spark:spark-sql-kafka-0-10_2.12:3.5.7") \
    .getOrCreate()

Run this either in the PySpark shell (pyspark --packages ...) or in a Jupyter Notebook cell at http://localhost:8888 (token: bigdata).

Demo: Define Schema

schema = StructType([
    StructField("user_id",    StringType()),
    StructField("page",       StringType()),
    StructField("event_time", StringType()),
])

raw = (spark.readStream
    .format("kafka")
    .option("kafka.bootstrap.servers", "kafka-broker:9092")
    .option("subscribe", "clicks")
    .option("startingOffsets", "latest")
    .load())

Demo: Parse JSON and Start Query

clicks = (raw
    .select(from_json(col("value").cast("string"), schema).alias("d"))
    .select("d.*")
    .withColumn("event_time", to_timestamp("event_time")))

query = (clicks.writeStream
    .format("console")
    .option("truncate", False)
    .trigger(processingTime="5 seconds")
    .start())

query.awaitTermination()

Type these events into the producer terminal:

{"user_id": "u1", "page": "/home",    "event_time": "2026-03-17 09:00:01"}
{"user_id": "u2", "page": "/product", "event_time": "2026-03-17 09:00:03"}
{"user_id": "u1", "page": "/cart",    "event_time": "2026-03-17 09:00:07"}

Output appears ~5 seconds after producing. Each micro-batch shows a batch number.

Event Time vs. Processing Time

Rule: Always window on event time when the timestamp is available. Processing time is only for synthetic/benchmarking streams.

Tumbling Windows

Fixed-size, non-overlapping intervals. Each event belongs to exactly one window.

|---5min---|---5min---|---5min---|
[9:00–9:05][9:05–9:10][9:10–9:15]

from pyspark.sql.functions import window

windowed_counts = clicks.groupBy(
    window(col("event_time"), "5 minutes"),
    col("page")
).count()

Use case: periodic, non-overlapping summaries (billing intervals, per-minute snapshots).

Run Tumbling

tumbling_query = (windowed_counts.writeStream
    .outputMode("update")
    .format("console")
    .option("truncate", False)
    .trigger(processingTime="5 seconds")
    .start())

Output appears every trigger interval with updated window counts.

Sliding Windows

Overlapping intervals. Each event may belong to multiple windows.

|---10min--|
    |---10min--|
        |---10min--|
← slide: 5 min →

sliding_counts = clicks.groupBy(
    window(col("event_time"), "10 minutes", "5 minutes"),
    col("page")
).count()

sliding_query = (sliding_counts.writeStream
    .outputMode("update")
    .format("console")
    .option("truncate", False)
    .trigger(processingTime="5 seconds")
    .start())
# window("column", window_duration, slide_duration)

Use case: "Rolling 10-minute click rate, updated every 5 minutes" — smoothed metrics, trend detection, moving averages. Expect more output rows than tumbling (each event counted in multiple windows).

Session Windows

Gap-based intervals — window closes after a period of inactivity.

from pyspark.sql.functions import session_window

session_counts = clicks.groupBy(
    session_window(col("event_time"), "10 minutes"),
    col("user_id")
).count()

session_query = (session_counts.writeStream
    .outputMode("update")
    .format("console")
    .option("truncate", False)
    .trigger(processingTime="5 seconds")
    .start())
# window closes if no event arrives within 10 minutes

Before starting the next example, stop the previous one (tumbling_query.stop(), sliding_query.stop(), or session_query.stop()) to avoid mixed console output.

user1: [9:00]---[9:02]---[9:04]     [gap > 10min]     [9:25]---[9:27]
       |_______ session 1 _________|                   |__ session 2 __|

Use case: User session analytics, variable-length activity windows. Each user gets their own dynamically-sized session.

Choosing the Right Window

Quick decision rule:

• "Every N minutes, give me a fresh count" → Tumbling
• "Every 5 minutes, give me the last 15 minutes" → Sliding
• "How long was each user active?" → Session

Activity: Window Design Challenge

Time: 8 minutes | Individual

For each scenario below, choose the best window type and explain why:

1. A payment processor wants to count transactions per merchant per hour to detect unusual volume spikes
2. A streaming platform wants a 30-second rolling average of concurrent viewers, updated every 10 seconds
3. An e-commerce site wants to track each customer's shopping session — from first page view until 20 minutes of inactivity
4. A data center monitors server CPU every 10 seconds and wants peak CPU per 5-minute interval

Write your answers — we'll discuss as a class.

Key Takeaways

• Structured Streaming = batch DataFrame API + readStream/writeStream — the same transformations, same Catalyst, same SQL
• Micro-batch collects events over a trigger interval, runs a Spark job, emits results — ~100ms minimum latency, exactly-once semantics
• Event time is when the event occurred (in the payload); processing time is when Spark saw it — always prefer event time
• Tumbling: one window per interval, non-overlapping
• Sliding: overlapping windows — each event counted in multiple windows
• Session: gap-based, dynamic per-entity — closes after inactivity

Session 2: Watermarks, output modes, stream joins, checkpointing — and 60 minutes of hands-on lab

The Gaps Structured Streaming Leaves Open

• Sub-millisecond latency — micro-batch minimum is ~100ms; financial trading and telemetry alerting need true per-record processing (covered in Week 13)
• Complex batch transformations — SQL-based data modeling, lineage tracking, and test-driven transformation pipelines are not a streaming concern → dbt
• Pipeline orchestration — who schedules the streaming job, retries it on failure, coordinates it with batch ETL, and alerts on SLA violations? → Airflow
• Cross-source federated queries — joining a live stream with historical S3 data at query time → Trino / Athena

What Comes Next

Spark Structured Streaming is the right tool for micro-batch real-time processing — but a production data platform wraps it with orchestration (Airflow), batch transformation (dbt), and observability (Grafana). That full stack is Week 15.