Today's Answer: Watermarks Bound the Trade-Off

You declare a maximum lateness tolerance. Spark uses it to:

Everything runs in Docker against the Kafka + Spark stack you used last week.

Lab Goals + Environment Check

By the end of this session you will have built:

1. A windowed sensor aggregation pipeline writing Parquet to HDFS
2. A stream-static join enriching live events with a dimension table
3. A stream-stream join matching ad impressions to clicks
4. A recovered streaming query that resumes from a checkpoint

Check your environment:

docker-compose ps   # kafka, spark-master, spark-worker should be Running

# Optional (from docker/): reset Week 14 topics + one smoke-test click event
make week14-reset

# Verify topics
docker exec -it kafka-broker kafka-topics --list --bootstrap-server localhost:9092

The Late Data Problem

Without watermarks, Spark must keep all window state forever.

Event at 9:02 AM → arrives at 9:14 AM (8 min late)

Tumbling windows:
  [9:00–9:05]  ← should this window update?
  [9:05–9:10]  ← or this one?
  [9:10–9:15]  ← the event's processing time falls here

• Without watermark: Spark retains all windows indefinitely — unbounded memory growth
• With watermark: Spark tracks the maximum event time seen, and discards events/state older than max_event_time − watermark_delay

The trade-off: larger delay → more late events accepted, more memory; smaller delay → less memory, more events dropped

How Watermarks Work

# "Accept events up to 10 minutes late. Drop anything older."
watermarked = clicks.withWatermark("event_time", "10 minutes")

counts = watermarked.groupBy(
    window(col("event_time"), "5 minutes"),
    col("page")
).count()

Watermark value = max(event_time seen so far) − watermark_delay

After 9:16, any event for 9:00–9:05 is silently dropped.

Output Modes

# Append: each window result written once after watermark passes
.writeStream.outputMode("append")

# Update: updated rows re-emitted each trigger (good for live dashboards)
.writeStream.outputMode("update")

# Complete: full table every trigger (only for small result sets!)
.writeStream.outputMode("complete")

Output Mode Selection Rules

The most common production choice: aggregation + watermark + append mode + Parquet sink

This writes each finalized result exactly once — making the output idempotent and easy to consume downstream.

Lab Setup — Start the Sensor Producer

# Terminal 1: start continuous sensor event producer
docker exec -it spark-master \
  python /datasets/sensors/produce_sensors.py

Lab Setup — Run Spark

Open a second terminal for your PySpark job:

docker exec -it spark-master pyspark \
  --packages org.apache.spark:spark-sql-kafka-0-10_2.12:3.5.7

Lab Setup — Jupyter Option

Or open a Jupyter Notebook at http://localhost:8888 (token: bigdata) and use the PySpark kernel.

We'll build sensor_pipeline.py step by step.

Activity 1

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, from_json, window, avg, max as spark_max
from pyspark.sql.types import StructType, StructField, StringType, DoubleType, TimestampType

spark = SparkSession.builder.appName("SensorPipeline").getOrCreate()
spark.sparkContext.setLogLevel("WARN")

schema = StructType([
    StructField("sensor_id",   StringType()),
    StructField("temperature", DoubleType()),
    StructField("ts",          TimestampType()),
])

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

readings = (raw
    .select(from_json(col("value").cast("string"), schema).alias("d"))
    .select("d.*")
    .withWatermark("ts", "2 minutes"))

Activity 1

# Tumbling 5-minute window: avg + max temperature per sensor
tumbling = (readings
    .groupBy(window("ts", "5 minutes"), "sensor_id")
    .agg(avg("temperature").alias("avg_temp"),
         spark_max("temperature").alias("max_temp")))

# Sliding 10-minute window, 5-minute slide
sliding = (readings
    .groupBy(window("ts", "10 minutes", "5 minutes"), "sensor_id")
    .agg(avg("temperature").alias("avg_temp")))

# Query A: show tumbling output in console (easy to verify in class)
q1_view = (tumbling.writeStream
  .outputMode("update")
  .format("console")
  .option("truncate", False)
  .trigger(processingTime="10 seconds")
  .queryName("tumbling_view")
  .start())

# Query B: persist tumbling to HDFS Parquet for checkpoint recovery later
q1 = (tumbling.writeStream
  .outputMode("append")
  .format("parquet")
  .option("checkpointLocation", "hdfs:///checkpoints/tumbling/")
  .option("path", "hdfs:///output/sensor_tumbling/")
  .trigger(processingTime="10 seconds")
  .queryName("tumbling_5min")
  .start())

# Query C: sliding to console (update mode — re-emits changed windows)
q2 = (sliding.writeStream
    .outputMode("update")
    .format("console")
    .option("truncate", False)
    .trigger(processingTime="10 seconds")
    .queryName("sliding_10min")
    .start())

spark.streams.awaitAnyTermination()

Demo: Inject a Late Event

After the pipeline has been running for ~3 minutes:

# Late by 90 seconds — WITHIN the 2-minute watermark → accepted
echo '{"sensor_id":"s1","temperature":99.9,"ts":"2026-03-17T08:58:30"}' | \
  docker exec -i kafka-broker kafka-console-producer \
    --topic sensors --bootstrap-server localhost:9092

Inject Dropped Event

# Late by 5 minutes — PAST the watermark → silently dropped
echo '{"sensor_id":"s1","temperature":99.9,"ts":"2026-03-17T08:55:00"}' | \
  docker exec -i kafka-broker kafka-console-producer \
    --topic sensors --bootstrap-server localhost:9092

Observe Results

• First event may update an open window.
• Second event is dropped silently.
• Check q1.lastProgress["numInputRows"] after each batch.

Checkpoint: Lab Part 1

Before moving to Part 2, verify:

# HDFS output should have Parquet files
docker exec -it spark-master \
  hdfs dfs -ls /output/sensor_tumbling/

# Checkpoint directory should have offsets/ and commits/
docker exec -it spark-master \
  hdfs dfs -ls /checkpoints/tumbling/

Verify Output

• /output/sensor_tumbling/ should contain .parquet files.
• /checkpoints/tumbling/offsets/ should contain numbered files.
• If empty, inspect q1.lastProgress before continuing.

Activity 2: Static Join

Enrich the live sensor stream with a static metadata table

# sensor_metadata.csv: sensor_id, location, building, floor
sensor_meta = spark.read.csv(
    "hdfs:///datasets/sensors/sensor_metadata.csv",
    header=True, inferSchema=True)

enriched = readings.join(sensor_meta, on="sensor_id", how="left")

Run Static Join

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

Key point: The static DataFrame is loaded once and broadcast to executors. It does not need a watermark. New rows added to the CSV after the query starts will not be picked up automatically.

Activity 3: Stream-Stream Join

Match ad impressions to click events within a 5-minute window

from pyspark.sql.functions import expr

impression_schema = StructType([
    StructField("ad_id",    StringType()),
    StructField("user_id",  StringType()),
    StructField("shown_at", TimestampType()),
])
click_schema = StructType([
    StructField("ad_id",      StringType()),
    StructField("user_id",    StringType()),
    StructField("clicked_at", TimestampType()),
])

Activity 3: Read Streams

def read_topic(topic, schema):
    return (spark.readStream.format("kafka")
        .option("kafka.bootstrap.servers", "kafka-broker:9092")
        .option("subscribe", topic).option("startingOffsets", "latest")
        .load()
        .select(from_json(col("value").cast("string"), schema).alias("d"))
        .select("d.*"))

impressions = read_topic("impressions", impression_schema) \
    .withWatermark("shown_at",   "10 minutes")
clicks      = read_topic("clicks_ads",  click_schema) \
    .withWatermark("clicked_at", "10 minutes")

Activity 3: Join Run

# Join: same ad, same user, click within 5 minutes of impression
matched = impressions.join(clicks,
    expr("""
        impressions.ad_id   = clicks.ad_id   AND
        impressions.user_id = clicks.user_id AND
        clicked_at BETWEEN shown_at AND shown_at + INTERVAL 5 MINUTES
    """),
    joinType="leftOuter")

query = (matched.writeStream
    .format("console")
    .option("truncate", False)
    .outputMode("append")
    .option("checkpointLocation", "hdfs:///checkpoints/ad_join/")
    .trigger(processingTime="10 seconds")
    .start())
query.awaitTermination()

Why watermarks are required here: Both streams buffer rows waiting for a match. Without watermarks, Spark retains the full history of both streams — unbounded memory. Watermarks define when a row can no longer find a match and should be evicted.

Activity 4: Stop and Check

Stop the tumbling pipeline from Activity 1 and inspect offsets

# Step 1: stop the running query
q1.stop()

# Step 2: check the last committed batch number
import os
# Or from HDFS:
# hdfs dfs -ls /checkpoints/tumbling/offsets/

Activity 4: Restart

# Step 3: restart with the SAME checkpoint location
q1_restart = (tumbling.writeStream
    .outputMode("append")
    .format("parquet")
    .option("checkpointLocation", "hdfs:///checkpoints/tumbling/")  # same!
    .option("path", "hdfs:///output/sensor_tumbling/")
    .trigger(processingTime="10 seconds")
    .start())

# Spark reads the checkpoint, finds the last committed Kafka offset,
# and resumes from the next unconsumed message.
print(q1_restart.lastProgress)

Inspect Checkpoint

# Inspect the checkpoint directory structure
docker exec -it spark-master \
  hdfs dfs -ls /checkpoints/tumbling/

# offsets/5 — the Kafka offset record for batch 5
docker exec -it spark-master \
  hdfs dfs -cat /checkpoints/tumbling/offsets/5

Look for numbered files in offsets/ and commits/ — the highest commit number is where Spark will resume.

Checkpoint Anatomy

/checkpoints/tumbling/
  metadata          ← query ID, schema, configuration
  offsets/          ← Kafka partition offsets per batch (what was READ)
    0, 1, 2, 3 ...
  commits/          ← which batches completed successfully (what was WRITTEN)
    0, 1, 2, 3 ...
  state/            ← stateful operator state (window aggregates)

Exactly-once guarantee: On restart, Spark reads the highest commits/N, finds the corresponding offsets/N, and starts reading Kafka from the next offset. Any partial writes from the failed batch are discarded.

Discussion Questions

Take 5 minutes — discuss with a neighbor:

1. Output mode: Your pipeline aggregates clicks per user per hour and writes to a live dashboard. Users are refreshed every 30 seconds. Should you use append, update, or complete mode? Why?

2. Watermark sizing: Your mobile app can buffer events for up to 30 minutes when offline. What watermark delay do you set? What's the memory cost of that choice?

3. Stream-stream vs. stream-static: You're joining live transactions to a fraud blocklist. The blocklist is updated every 10 minutes. Which join type do you use and why?

4. Checkpointing: You upgrade your Spark version. Can you restart the streaming job from the existing checkpoint? What risks exist?

Key Takeaways

• Watermarks bound stateful memory — declare the maximum lateness you'll tolerate; Spark evicts older state
• Output modes: Append (once, final, requires watermark) → Update (re-emits changed rows) → Complete (entire table, small cardinality only)
• Stream-static join: dimension enrichment, broadcast once, no watermark needed
• Stream-stream join: both sides buffer state; watermarks on both sides are required to prevent unbounded memory
• Checkpointing = fault tolerance: restart resumes from last committed Kafka offset with no duplicates and no data loss

The Gaps Structured Streaming Leaves Open

• Pipeline orchestration — who schedules the streaming job, retries it on failure, coordinates it with nightly batch ETL, and pages you at 3 AM when it stalls? → Apache Airflow
• SQL-based batch transformations with lineage — transforming historical data with documented, tested, versioned SQL models is a batch concern → dbt
• Production monitoring and SLA enforcement — query lag, state size, throughput, and watermark drift need dashboards and alerts → Prometheus + Grafana (Week 15)
• True sub-millisecond latency — micro-batch floor is ~100ms; for <10ms requirements, Apache Flink (covered in Week 13) is the answer

What Comes Next

Spark Structured Streaming is the right tool for micro-batch real-time processing — but production requires orchestration (Airflow to schedule it), transformation (dbt to model the historical side), and observability (Grafana to know when it breaks). That full stack is Week 15.

Homework Reminder

Streaming Pipeline homework — due Sunday 11:59 PM

Four tasks:

1. Sensor tumbling pipeline — 1-minute windows to Parquet, 30-second watermark, append mode
2. Sliding window comparison — 5-minute/1-minute sliding to console; written explanation of why windows repeat
3. Late data experiment — inject events inside and past the watermark; explain what happened and why
4. Checkpoint recovery — stop, restart, screenshot lastProgress before and after; explain how Spark knows where to resume

Submit: 4 Python scripts + 1 PDF with screenshots and written explanations