Structured Streaming Basics

Analogy categoryAnalogies

Streams as Unbounded Tables

Structured Streaming models an incoming data stream as an unbounded table that new rows are continuously appended to, letting you write the exact same DataFrame transformations — filter(), groupBy(), join() — you would use for a batch job, and Spark incrementally applies them to just the new data as it arrives. You start a stream by reading from a source with spark.readStream instead of spark.read, apply transformations as normal, and terminate the pipeline with .writeStream instead of .write, specifying an output mode and a trigger interval.

🏏

Cricket analogy: It is like a live scorecard that treats the entire match as one continuously growing ball-by-ball table — the same run-rate formula you'd apply to a completed match's data just recalculates incrementally as each new delivery is bowled.

Output Modes and Triggers

Output mode controls what gets written to the sink on each trigger: append writes only new rows and is required for non-aggregated streams, complete rewrites the entire result table every trigger and is needed for aggregations without watermarking, and update writes only the rows that changed since the last trigger. Triggers control how often micro-batches fire — Trigger.ProcessingTime("10 seconds") for a fixed cadence, Trigger.Once() (or availableNow=True) to process everything available and stop, or Trigger.Continuous() for Spark's lower-latency continuous processing mode, which supports a narrower set of operations.

🏏

Cricket analogy: It is like a scoreboard operator choosing between updating only the latest ball's runs (append), redisplaying the entire scorecard fresh every over (complete), or only refreshing the numbers that actually changed since the last ball (update).

python

from pyspark.sql import functions as F

# Read a stream of JSON events landing in a directory
events_stream = (
    spark.readStream
         .format("json")
         .schema(events_schema)
         .option("maxFilesPerTrigger", 10)
         .load("s3://data/events/")
)

windowed_counts = (
    events_stream
        .withWatermark("event_time", "10 minutes")
        .groupBy(
            F.window("event_time", "5 minutes"),
            "event_type"
        )
        .count()
)

query = (
    windowed_counts.writeStream
        .format("console")
        .outputMode("update")
        .trigger(processingTime="30 seconds")
        .option("checkpointLocation", "s3://data/checkpoints/event-counts")
        .start()
)

query.awaitTermination()

Watermarks and Late Data

Because a stream is unbounded, Spark cannot wait forever for late-arriving data before finalizing a windowed aggregation — withWatermark("event_time", "10 minutes") tells Spark to consider event time up to 10 minutes behind the maximum event time seen so far as the cutoff, after which state for older windows is dropped and further late data for those windows is discarded. Watermarking is essential for bounding the memory Spark uses to keep aggregation state, and it only applies when using event-time windows, not the wall-clock processing time at which Spark happened to receive the record.

🏏

Cricket analogy: It is like a scorer who accepts a corrected delivery count up to 10 minutes after an over ends, but refuses to reopen the books for an over from an hour ago just because a late correction request comes in.

checkpointLocation is not optional for production streaming jobs — it stores the stream's progress (offsets processed) and aggregation state, allowing the query to resume exactly where it left off after a driver restart or deployment, rather than reprocessing from the beginning or skipping data.

Using outputMode("complete") on an unbounded, unwindowed aggregation will cause Spark to keep growing the entire result set in memory forever, since there is no watermark to expire old state — this is a common cause of streaming jobs that run fine for hours and then suddenly run out of memory.

Structured Streaming treats a live stream as an unbounded table processed with the same DataFrame API as batch jobs.
spark.readStream and .writeStream replace spark.read and .write for streaming pipelines.
Output modes append, complete, and update control what gets written to the sink on each trigger.
Triggers (ProcessingTime, Once/availableNow, Continuous) control how often micro-batches execute.
withWatermark() bounds how long Spark waits for late event-time data before dropping aggregation state.
checkpointLocation persists offsets and state so a restarted query resumes without reprocessing or data loss.
Unbounded complete-mode aggregations without watermarking can grow memory usage indefinitely.