Spark Structured Streaming

Spark Structured Streaming lets you write a stream-processing job as if it were a batch query over a table that never stops growing. This page explains the model, how micro-batches and checkpoints give fault tolerance, triggers and output modes, and a complete PySpark example.

1. What It Is

Spark Structured Streaming is Apache Spark's stream-processing engine, introduced in Spark 2.0 and marked production-ready in 2.2. Its central idea is that a stream is just a table that keeps growing. You write an ordinary DataFrame or SQL query against it, and Spark figures out how to run that query incrementally as new data arrives, maintaining whatever state is needed between runs.

It replaced the older DStream API (Spark Streaming), which exposed raw RDD micro-batches and left event-time handling and exactly-once delivery largely to the developer. Structured Streaming runs on the same Catalyst optimiser and Tungsten execution engine as batch Spark, so the same code can often run in batch or streaming mode by switching read to readStream and write to writeStream.

2. Why It Matters

Many teams already use Spark for batch ETL. Structured Streaming lets them add low-latency pipelines (Kafka to Delta Lake, CDC ingestion, near-real-time aggregates) without learning a second engine and without maintaining two code paths. It provides end-to-end exactly-once guarantees for replayable sources and idempotent or transactional sinks, event-time windows with watermarks, stateful operations, and stream-static and stream-stream joins. It is also the engine behind Databricks Auto Loader and Delta Live Tables. Its latency floor (typically hundreds of milliseconds to seconds in micro-batch mode) is the main reason teams choose Apache Flink instead for very low-latency work.

3. How It Works

Micro-batch execution

By default the engine runs a series of small batch jobs. For each trigger it asks every source for its latest available offset (Kafka partition offsets, new files in a directory), writes the planned offset range to a write-ahead log in the checkpoint location, runs an incremental job over exactly that range, writes the output to the sink, and records a commit. Stateful operators (aggregations, deduplication, joins, flatMapGroupsWithState) keep per-key state in a versioned state store, by default backed by memory plus checkpoint files, optionally RocksDB.

Because offsets are logged before processing and sources can replay, a crash at any point leads to the same batch being recomputed with the same data. If the sink is idempotent or transactional (Delta Lake, or foreachBatch code that upserts by key), the result is exactly-once. See Exactly-Once Semantics.

Triggers

The trigger controls when batches run: the default (as fast as possible), processingTime="30 seconds" (fixed interval), availableNow=True (process everything currently available in several batches, then stop; good for scheduled incremental jobs; it replaces the older once=True), and continuous="1 second" (experimental low-latency mode with a limited set of supported operations).

Output modes

Append writes only new result rows that will never change; with aggregations this requires a watermark so Spark knows when a window is final. Update writes only rows that changed in this trigger. Complete rewrites the entire result table each time, which is only practical for small aggregates.

Sources and sinks

Built-in sources include Kafka, files (Parquet, JSON, CSV, ORC and others in a directory), Delta tables, and a rate source for testing. Sinks include Kafka, files, Delta, console and memory for debugging, and foreach/foreachBatch for arbitrary output such as JDBC upserts.

4. Worked Example: Kafka to Parquet

This PySpark job reads JSON click events from Kafka, counts clicks per page in 5-minute event-time windows with a 10-minute watermark, and writes finalised windows to Parquet. Start it with the Kafka connector package matching your Spark version, for example spark-submit --packages org.apache.spark:spark-sql-kafka-0-10_2.12:3.5.1 clicks.py.

from pyspark.sql import SparkSession, functions as F, types as T

spark = SparkSession.builder.appName("clicks").getOrCreate()

schema = T.StructType([
    T.StructField("user_id", T.StringType()),
    T.StructField("page",    T.StringType()),
    T.StructField("ts",      T.TimestampType()),
])

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

events = (raw.select(F.from_json(F.col("value").cast("string"), schema).alias("e"))
             .select("e.*"))

counts = (events
          .withWatermark("ts", "10 minutes")                 # tolerate 10 min lateness
          .groupBy(F.window("ts", "5 minutes"), "page")
          .count())

query = (counts.writeStream
         .outputMode("append")                                # emit each window once, when final
         .format("parquet")
         .option("path", "/data/clicks_per_page")
         .option("checkpointLocation", "/chk/clicks_per_page")
         .trigger(processingTime="1 minute")
         .start())

query.awaitTermination()

Walk through what happens to one window. Suppose the latest event time Spark has seen is 12:17. The watermark is that maximum minus the allowed lateness, 12:07. The window 12:00 to 12:05 ends before the watermark, so no event that Spark is still willing to accept can change it: in append mode it is written to Parquet at the next trigger and its state is dropped. The window 12:05 to 12:10 is still open; a late event stamped 12:08 arriving now is counted. An event stamped 12:03 arriving now is older than the watermark and may be dropped. The watermark therefore trades completeness (how late data can be and still count) against latency (how long you wait before a window is emitted) and state size (how many open windows Spark keeps in memory). Choose the lateness from measured delays in your sources, not by guessing.

To test without Kafka, replace the source with spark.readStream.format("rate").option("rowsPerSecond", 100).load(), which emits timestamp and value columns, and the sink with format("console"). query.lastProgress returns a JSON dictionary with input rows per second, batch duration and state size, which is what you should monitor in production.

5. Comparison With Other Engines

6. Pitfalls

• Changing the query with an old checkpoint. Many changes (new aggregation keys, different stateful operators) are incompatible with existing state. Plan a new checkpoint and backfill.
• Sharing a checkpoint between queries. Each query needs its own checkpoint location.
• Missing watermark. Without one, aggregation state grows forever and append mode on aggregations is not allowed.
• Small files. Short triggers into file sinks create many tiny files; use longer triggers, Delta with compaction, or availableNow on a schedule.
• Non-idempotent foreachBatch. A batch can be replayed after failure; use batchId or key-based upserts so replays do no harm.
• Unbounded stream-stream joins. Add watermarks and time-range join conditions so join state can be purged.

7. Related Topics

• Stream Processing — section overview
• Batch vs Streaming — when to stream at all
• Event Time, Windows and Watermarks — the time model used above
• Exactly-Once Semantics — how replay plus idempotent sinks work
• Apache Flink — the main alternative engine
• PySpark Data Streaming — more PySpark streaming examples
• Kafka Producer and Consumer — the usual source

8. Further Reading

• Armbrust et al. (2018), Structured Streaming: A Declarative API for Real-Time Applications in Apache Spark, SIGMOD.
• Apache Spark documentation, Structured Streaming Programming Guide.
• Apache Spark documentation, Structured Streaming + Kafka Integration Guide.
• Akidau, Chernyak and Lax (2018), Streaming Systems (O'Reilly).