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).
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.
Practice what you learned
1. How does Structured Streaming conceptually model an incoming data stream?
2. Which output mode rewrites the entire result table to the sink on every trigger?
3. What is the purpose of withWatermark() in a streaming aggregation?
4. What does the checkpointLocation option provide for a streaming query?
5. What risk arises from using outputMode("complete") on an aggregation with no watermark?
Was this page helpful?
You May Also Like
Window Functions in Spark SQL
Understand how to compute running totals, rankings, and moving averages across partitions of rows using Spark's window function API.
Reading and Writing Data Sources
Learn how Spark's DataFrameReader and DataFrameWriter APIs handle formats like Parquet, CSV, and JDBC, plus schema handling and partitioned writes.
Running SQL Queries on DataFrames
Learn how to register DataFrames as temporary views and query them with standard SQL syntax, and how SQL and the DataFrame API interoperate under the hood.
Related Reading
Related Study Notes in Programming
Browse all study notesApache Flink Study Notes
Programming · 30 topics
ProgrammingHadoop Study Notes
Programming · 30 topics
ProgrammingSnowflake Study Notes
Programming · 30 topics
ProgrammingApache Airflow Study Notes
Programming · 30 topics
Programmingdbt (Data Build Tool) Study Notes
Programming · 30 topics
ProgrammingPowerShell Study Notes
Programming · 30 topics