Course/Level 3 · Performance & Production Engineering

Module 11: Structured Streaming

Structured Streaming: triggers, checkpoints, state, watermarks and Kafka.

9 hands-on6 concept15 topics

Topics in this module

Batch vs Stream Processing
Spark Structured Streaming Architecture
Sources and Sinks
Micro-Batch Processing
Streaming Triggers
Checkpointing
Watermarking
Stateful vs Stateless Processing
Streaming Joins
Windowed Aggregations
Late and Out-of-Order Events
Kafka and Event Hubs Integration
Exactly-Once Processing Semantics
Streaming Failure Recovery
End-to-End Streaming Pipelines

Process data as it arrives, incrementally and with state. You practise on a small event feed using the one trigger Free Edition supports: availableNow.

Serverless rule: only Trigger.AvailableNow works. processingTime and continuous are not supported (Once is deprecated). Every run: process what's new, then stop
Real always-on streams need classic compute or continuous pipeline mode: those topics are marked concept
Setup: a generator for JSON event files (500 events per batch, 10 seconds apart)

Python

from pyspark.sql import functions as F

catalog = spark.sql("SELECT current_catalog()").first()[0]
spark.sql(f"USE SCHEMA {catalog}.dbx_course")

base = f"/Volumes/{catalog}/dbx_course/raw/stream"
src = f"{base}/events"
schema = "event_id LONG, user_id INT, event_type STRING, event_ts TIMESTAMP, amount DOUBLE"

def land_events(batch, start, n=500):
    """Write n events (ids start..start+n-1) as JSON into the landing folder."""
    (spark.range(start, start + n).select(
        F.col("id").alias("event_id"),
        (F.pmod(F.hash("id", F.lit(1)), F.lit(50)) + 1).alias("user_id"),
        F.element_at(F.array(F.lit("view"), F.lit("click"), F.lit("buy")),
                     (F.pmod(F.hash("id", F.lit(2)), F.lit(3)) + 1)).alias("event_type"),
        F.timestamp_seconds(F.lit(1740787200) + F.col("id") * 10).alias("event_ts"),   # from 2025-03-01 UTC
        F.round(F.pmod(F.hash("id", F.lit(3)), F.lit(10000)) / 100, 2).alias("amount"),
    ).coalesce(1).write.mode("overwrite").json(f"{src}/batch_{batch}"))

land_events(1, 0)

11.01Concept

Batch vs Stream Processing

By the end: contrast batch and stream processing and know when streaming is worth it.

Batch: read everything, write the result
Streaming: treat the source as a table that keeps growing and process new rows
Incremental batch (availableNow) uses the streaming engine but stops when caught up: cheap, exactly-once, scheduled
Choose true streaming only when latency of seconds is a requirement

Latency need	Pick
Hours, daily	Batch job
Minutes	Incremental batch on a schedule
Seconds	Continuous streaming (classic compute or a continuous pipeline)

Remember

Same code, different trigger: one stream definition can run as incremental batch or continuous.

11.02Concept

Spark Structured Streaming Architecture

By the end: describe how Structured Streaming turns a stream into repeated small batch queries.

The same DataFrame API you know, plus a source, a sink, a trigger and a checkpoint.

You write a normal DataFrame query on spark.readStream
Spark runs it as a series of micro-batches, each over the new rows
A checkpoint records progress and state
Four choices define a stream: source, transformations, output mode, trigger

Output mode	Writes
`append`	Only new final rows (default)
`update`	Rows that changed
`complete`	The entire result, every time

11.03Hands-on on Free Edition

Sources and Sinks

By the end: read from and write to sources and sinks.

Python

# source: files (the schema is explicit)
raw = (spark.readStream.schema(schema)
       .option("recursiveFileLookup", "true").option("pathGlobFilter", "*.json")
       .json(src))

# sink: a Delta table
q = (raw.writeStream
        .option("checkpointLocation", f"{base}/_cp/bronze")
        .trigger(availableNow=True)
        .toTable("events_bronze"))
q.awaitTermination()
print(spark.table("events_bronze").count())     # 500

Source	Use
Files (`json`, `csv`, `parquet`)	Simple landing folders
Auto Loader (`cloudFiles`)	Files at scale: the better choice (Module 7)
Delta table (`readStream.table`)	Chain layers
Kafka, Event Hubs, Kinesis	Message streams

Sink	Use
Delta table (`toTable`)	Default
Kafka	Feed other systems
`foreachBatch`	Custom logic per batch, such as `MERGE`
Memory, console	Testing only

Streaming file sources need an explicit schema (or Auto Loader)
A Delta table is both the best sink and a stream source: spark.readStream.table("events_bronze")

11.04Hands-on on Free Edition

Micro-Batch Processing

By the end: see micro-batches and read their progress metrics.

Python

# land a second batch, then process only the new files
land_events(2, 500)

q = (spark.readStream.schema(schema)
        .option("recursiveFileLookup", "true").option("pathGlobFilter", "*.json").json(src)
        .writeStream.option("checkpointLocation", f"{base}/_cp/bronze")
        .trigger(availableNow=True).toTable("events_bronze"))
q.awaitTermination()

p = q.lastProgress
print(p["batchId"], p["numInputRows"])           # one micro-batch with the 500 new rows
print(spark.table("events_bronze").count())      # 1000

Field in `lastProgress`	Tells you
`batchId`	Which micro-batch (no new id when no data)
`numInputRows`	Rows read
`processedRowsPerSecond`	Speed
`durationMs`	Time split: planning, `addBatch`, `triggerExecution`
`stateOperators`	State size, rows dropped by the watermark
`sources`, `sink`	Offsets per source

maxFilesPerTrigger / maxBytesPerTrigger split a large backlog into several micro-batches
Name streams with .queryName("...") so they're easy to find
StreamingQueryListener sends metrics to external systems

Not shown

Exact values in lastProgress can differ slightly between runs. Look at numInputRows, not the whole object.

11.05Hands-on on Free Edition

Streaming Triggers

By the end: choose a trigger and know what serverless allows.

Trigger	Behaviour	Serverless
(none)	Continuous, ~3–5 s latency	No
`processingTime="10 seconds"`	Micro-batch every interval	No
`availableNow=True`	Process everything available, then stop	Yes
`realTime="5 minutes"`	Sub-second latency	No
`continuous`	Not supported anywhere	No
`once=True`	Deprecated	Works, but use `availableNow`

Python

(df.writeStream
   .trigger(availableNow=True)          # scheduled incremental run
   .option("checkpointLocation", cp)
   .toTable("target"))

availableNow suits a job on a schedule: run every 15 minutes, process the new data, stop
Without a trigger, a stream polls constantly and can generate many storage API calls and cost
You can change the trigger later and keep the checkpoint
For always-on on serverless: a continuous pipeline (Module 9)

Default to availableNow

Same results, same checkpoint, but you pay only while it runs.

11.06Hands-on on Free Edition

Checkpointing

By the end: prove that the checkpoint remembers progress.

Python

def run_bronze():
    q = (spark.readStream.schema(schema)
           .option("recursiveFileLookup", "true").option("pathGlobFilter", "*.json").json(src)
           .writeStream.option("checkpointLocation", f"{base}/_cp/bronze")
           .trigger(availableNow=True).toTable("events_bronze"))
    q.awaitTermination()
    return spark.table("events_bronze").count()

print(run_bronze())    # 1000
print(run_bronze())    # 1000: nothing new, nothing reprocessed

The checkpoint stores
Offsets	What each batch read
Commits	Which batches reached the sink
State	Data of stateful operators
Metadata	The query's id

One checkpoint per query, never shared
Put it in a Unity Catalog volume path, outside the table
Deleting the checkpoint = start from scratch (reprocess everything, duplicates if you append)

Safe to change with the same checkpoint: filters, rate limits, trigger.

Needs a new checkpoint: different sources, topics or paths; state schema or stateful operators; sink type.

Don't reuse a checkpoint

Pointing a different query at an old checkpoint gives errors or wrong results.

11.07Hands-on on Free Edition

Watermarking

By the end: bound late data with a watermark.

Python

events = spark.readStream.table("events_bronze")

windows = (
    events
    .withWatermark("event_ts", "10 minutes")
    .groupBy(F.window("event_ts", "5 minutes"), "event_type")
    .agg(F.count("*").alias("events"), F.round(F.sum("amount"), 2).alias("amount"))
)

Windows that end before the watermark are final. Their state is dropped.

Watermark = "how late can data be, in event time"
Needed for append-mode window results and to limit state size
Records within the threshold are always processed; older ones may be dropped
Choose the threshold from how late your data really is (Module 8.12)

Watermark delays output

In append mode a window is written only after the watermark passes it. The newest windows appear when later data arrives.

11.08Concept

Stateful vs Stateless Processing

By the end: tell stateful and stateless operations apart.

Stateful operators keep state that grows with keys and windows
Unbounded state = slow, large checkpoints, out-of-memory
Bound it with watermarks, fewer keys, shorter windows
Prefer the RocksDB state store for large state (the default on current Databricks runtimes)
stateOperators in progress shows numRowsTotal and numRowsDroppedByWatermark
Changing a stateful operator or its schema needs a new checkpoint

Python

# dedupe within the watermark: state is bounded
dedup = (events.withWatermark("event_ts", "10 minutes")
               .dropDuplicatesWithinWatermark(["event_id"]))

Know your state

Every stateful stream needs an answer to: what stops this state from growing forever?

11.09Hands-on on Free Edition

Streaming Joins

By the end: join a stream to static data and to another stream.

Stream with a static table (the common case):

Python

customers = spark.table("customers")            # static, read fresh each batch

enriched = (
    spark.readStream.table("events_bronze")
    .join(customers.select(F.col("customer_id").alias("user_id"), "country"),
          "user_id", "left")
)

Stream with stream needs time bounds and watermarks on both sides:

Python

views = (spark.readStream.table("events_bronze").filter("event_type = 'view'")
         .withWatermark("event_ts", "10 minutes").alias("v"))
buys = (spark.readStream.table("events_bronze").filter("event_type = 'buy'")
        .withWatermark("event_ts", "10 minutes").alias("b"))

view_then_buy = views.join(
    buys,
    F.expr("v.user_id = b.user_id AND b.event_ts BETWEEN v.event_ts AND v.event_ts + INTERVAL 5 MINUTES"),
    "inner",
).select(F.col("v.user_id"), F.col("v.event_id").alias("view_id"), F.col("b.event_id").alias("buy_id"))

Stream–static: simple, stateless. The static side is the latest version at each batch
Stream–stream: only append mode; keeps state for both sides
Outer joins require watermarks and emit unmatched rows (with nulls) late
No watermark = Spark keeps every key forever

11.10Hands-on on Free Edition

Windowed Aggregations

By the end: aggregate over tumbling, sliding and session windows.

Python

def write(df, table, cp):
    return (df.writeStream.outputMode("append")
              .option("checkpointLocation", f"{base}/_cp/{cp}")
              .trigger(availableNow=True).toTable(table))

q = write(windows, "events_5min", "win5")      # windows from topic 7
q.awaitTermination()
display(spark.table("events_5min").orderBy("window").limit(10))

Window	Code
Tumbling (no overlap)	`F.window("event_ts", "5 minutes")`
Sliding (overlap)	`F.window("event_ts", "10 minutes", "5 minutes")`
Session (gap-based)	`F.session_window("event_ts", "2 minutes")`

The slide can't be larger than the window length
The window column is a struct with start and end
Update mode writes partial results; use foreachBatch + MERGE for it (topic 13)

Last windows

With append mode, the newest windows stay open until more data moves the watermark. Land another batch and rerun to see them close.

11.11Hands-on on Free Edition

Late and Out-of-Order Events

By the end: handle late and out-of-order events.

Python

# an event that is hours late (event time 2025-03-01 00:01, arrives with batch 3)
(spark.createDataFrame([(99999, 7, "buy", "2025-03-01 00:01:00", 12.5)],
                       "event_id LONG, user_id INT, event_type STRING, event_ts STRING, amount DOUBLE")
      .withColumn("event_ts", F.to_timestamp("event_ts"))
      .coalesce(1).write.mode("overwrite").json(f"{src}/batch_3"))
land_events(4, 1000)                      # newer data moves the watermark forward
run_bronze()                              # bronze keeps all rows

Situation	What happens
Late but within the watermark	Included in its window
Later than the watermark	May be dropped from aggregates
Out of order inside a batch	Handled: event time, not arrival order

Bronze keeps everything, whatever its lateness (Module 8). The watermark only limits stateful aggregates
numRowsDroppedByWatermark counts what was discarded
To fix aggregates after the fact: reprocess the affected range in batch from bronze (Module 8.12)

Don't rely on streaming alone for correctness

Use the stream for fast results and a periodic batch recompute from bronze for exact ones.

11.12Concept

Kafka and Event Hubs Integration

By the end: read Kafka and Event Hubs topics (needs external systems).

Python

raw = (spark.readStream.format("kafka")
         .option("kafka.bootstrap.servers", "<host:port>")
         .option("subscribe", "events")
         .option("startingOffsets", "latest")
         .load())

parsed = (raw.select(F.col("value").cast("string").alias("json"), "topic", "partition", "offset", "timestamp")
             .select(F.from_json("json", schema).alias("e"), "*").select("e.*", "offset"))

Column from Kafka	Type
`key`, `value`	binary: cast and parse yourself
`topic`, `partition`, `offset`	where the record came from
`timestamp`	Kafka timestamp

Pick one: subscribe, subscribePattern, or assign
Event Hubs and Amazon MSK are reached through their Kafka-compatible endpoints, with authentication options for each
Keep passwords and keys in secrets or Unity Catalog service credentials, never in code
Land the raw value in bronze first, then parse (replayable)

Free Edition

Outbound access is limited to trusted domains, so connecting to your own Kafka is not possible. This topic is read-only reference.

11.13Concept

Exactly-Once Processing Semantics

By the end: explain how exactly-once results are achieved and where they stop.

Exactly-once = replayable source + checkpoint + idempotent sink
Delta sinks commit each batch atomically
foreachBatch is at-least-once: a batch may run again after a failure. Make the work idempotent

Python

def upsert(batch_df, batch_id):
    if batch_df.isEmpty():
        return
    batch_df.createOrReplaceTempView("batch")
    batch_df.sparkSession.sql("""
        MERGE INTO events_latest t USING batch s ON t.event_id = s.event_id
        WHEN MATCHED THEN UPDATE SET *
        WHEN NOT MATCHED THEN INSERT *
    """)

spark.sql("CREATE TABLE IF NOT EXISTS events_latest AS SELECT * FROM events_bronze WHERE 1 = 0")
(spark.readStream.table("events_bronze").writeStream
      .foreachBatch(upsert)
      .option("checkpointLocation", f"{base}/_cp/latest")
      .trigger(availableNow=True).start().awaitTermination())

MERGE on the key makes a replayed batch harmless
Delta merges from streams must go through foreachBatch
Handle empty batches. With stateful operators, consume the batch fully
Dedupe by key at the source if the producer can send duplicates

Spark Connect

foreachBatch on serverless runs under Spark Connect, where print and dbutils inside the function behave differently. Keep the function to Spark code.

11.14Concept

Streaming Failure Recovery

By the end: recover from failures and restart safely.

A failed run restarts from the checkpoint: committed batches are not repeated
The batch planned before the failure is retried first. If it failed from memory, give the restart more compute
Run streaming as a job with retries (Module 10.9), not by hand
Continuous streams on classic compute: use the job's continuous trigger (new run after failure, exponential backoff) and no autoscaling
Don't awaitTermination() on a continuous query in a job. Do wait for availableNow queries: they end

Failure	Typical cause	Fix
Out of memory on restart	Huge first batch or big state	More compute, `maxBytesPerTrigger`, watermark
Checkpoint incompatibility	Changed sources or stateful logic	New checkpoint, reprocess from bronze
Schema change in source	New columns	Auto Loader modes (Module 7.7)
Source data gone	Retention expired before you read it	Restore the source or restart fresh

A new checkpoint reprocesses

Starting over means reading from the start (or startingOffsets). Know your sink can take it: deduplicate or overwrite.

11.15Hands-on on Free Edition

End-to-End Streaming Pipelines

By the end: assemble a bronze, silver and gold streaming job that runs as incremental batch.

One job, three availableNow queries, each with its own checkpoint.

Python

def run_layer(df, table, cp, mode="append"):
    q = (df.writeStream.outputMode(mode)
           .option("checkpointLocation", f"{base}/_cp/{cp}")
           .trigger(availableNow=True).toTable(table))
    q.awaitTermination()

# bronze: files -> table
run_layer(spark.readStream.schema(schema).option("recursiveFileLookup", "true")
          .option("pathGlobFilter", "*.json").json(src), "events_bronze", "bronze")

# silver: valid, deduplicated
silver = (spark.readStream.table("events_bronze")
          .filter("amount IS NOT NULL AND amount >= 0")
          .withWatermark("event_ts", "10 minutes")
          .dropDuplicatesWithinWatermark(["event_id"]))
run_layer(silver, "events_silver", "silver")

# gold: windowed counts (append mode closes windows after the watermark)
gold = (spark.readStream.table("events_silver")
        .withWatermark("event_ts", "10 minutes")
        .groupBy(F.window("event_ts", "30 minutes"), "event_type")
        .agg(F.count("*").alias("events"), F.round(F.sum("amount"), 2).alias("amount")))
run_layer(gold, "events_by_type", "gold")

Put the notebook in a job and schedule it (Module 10). Each run handles whatever landed since the last one
Land more batches with land_events(n, start) and rerun: only the new rows flow through
Compare: Module 9 does the same with less code and managed checkpoints

Each layer, its own checkpoint

Name checkpoints after the table. Never share one between layers.

Recap

What you can do now

Batch, streaming and incremental batch, and the availableNow rule on serverless
Sources, sinks, checkpoints, progress metrics
Watermarks, state, windows, joins, late data
Exactly-once, foreachBatch upserts and recovery