Spark Structured Streaming and Stateful Processing

Published by Programming Line in Data Engineering in Depth. Last updated .

Purpose and outcome

Spark Structured Streaming lets a data engineering pipeline treat an unbounded event stream as a table that is continuously appended to, updated, or completed. Instead of writing a separate streaming program with a different mental model from batch Spark, you express the computation with DataFrame or SQL operations, and Spark repeatedly plans and executes small incremental jobs. The outcome of this lesson is practical: you should be able to explain what happens inside a streaming query, choose output modes and triggers deliberately, use watermarks and checkpoints correctly, and diagnose the common ways stateful pipelines fail.

Stateful processing is the part that makes streaming more than row-by-row transport. Aggregations by window, duplicate suppression, stream-stream joins, and custom session logic all require Spark to remember facts from previous micro-batches. That memory is held in a state store, keyed by grouping columns or join keys, and protected by checkpoint metadata so the query can recover after interruption.

How Structured Streaming works internally

A Structured Streaming query has a logical plan, a streaming source, a trigger, a sink, and a checkpoint location. Spark first analyzes the DataFrame plan and determines which parts are streaming. At runtime the driver asks each source for the next range of unread data. With the Kafka source this range is usually offsets; with file sources it is newly discovered files; with the rate source it is generated rows. Spark records progress information in the checkpoint, builds a physical plan for the next micro-batch, runs Spark tasks across executors, commits state changes, then commits sink output.

The default engine is micro-batch execution. Each trigger interval creates a bounded batch from the stream, so familiar Spark operators can run on each increment. This design gives high throughput and integrates with Spark SQL optimization, but latency is normally measured in hundreds of milliseconds or seconds, not single-digit milliseconds. Continuous processing exists for lower latency workloads, but many stateful operations and sinks are designed around micro-batch semantics, so most production Structured Streaming data engineering jobs use micro-batches.

Stateful operators write and read key-value state through a state store provider. For a windowed count, the key contains the group columns and the time window; the value contains the aggregate buffer. Spark versions commonly use an HDFS-compatible state store by default, and can also use RocksDB-backed state for larger local state footprints when configured. The checkpoint stores offsets, commits, operator metadata, and state snapshots or deltas under the query’s checkpoint directory. Deleting or sharing this directory changes correctness, because it is the query’s memory of what it already processed.

Watermarks bound how long Spark keeps event-time state. A watermark is Spark’s statement that rows older than a threshold behind the maximum observed event time are late enough to drop for a particular stateful operator. For example, with withWatermark("event_time", "10 minutes"), once Spark has seen event times up to 12:30, the watermark may allow it to evict state for windows ending before about 12:20. This does not mean every row arrives in order; it means the pipeline accepts a controlled risk that very late rows will not revise already-finalized state.

API anatomy

A streaming read starts with spark.readStream, a source format, source options, and usually a schema. A transformation section looks like ordinary DataFrame code: select, filter, parse JSON, join static dimensions, group by windows, or deduplicate. A streaming write starts with writeStream and defines the sink, output mode, trigger, checkpoint location, and query name. The important distinction is that the DataFrame is not executed until start() creates a running query.

Output mode controls what rows the sink receives after each trigger. append writes only rows that will not change again, which is typical for event logs and watermark-finalized windows. update writes changed result rows after each micro-batch, useful for mutable aggregates in sinks that can handle repeated keys. complete writes the entire result table each time, which is convenient for console debugging but expensive for large aggregations.

Example 1: windowed counts from the rate source

The rate source is deterministic enough for learning because it produces a timestamp and increasing value. This example buckets rows into ten-second event-time windows and counts even and odd values. The expected console output is a table containing window structs, parity values, and counts. Counts grow while the window is open and become stable after no more rows for that window are generated.

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, window

spark = SparkSession.builder.appName("structured-streaming-rate-counts").getOrCreate()
spark.sparkContext.setLogLevel("WARN")

events = (
    spark.readStream.format("rate")
    .option("rowsPerSecond", 5)
    .load()
    .select(col("timestamp").alias("event_time"), (col("value") % 2).alias("parity"))
)

counts = events.groupBy(window(col("event_time"), "10 seconds"), col("parity")).count()

query = (
    counts.writeStream.format("console")
    .outputMode("complete")
    .option("truncate", "false")
    .start()
)

query.processAllAvailable()
query.stop()

This is stateful because Spark maintains an aggregate buffer for each window and parity key. Complete mode is acceptable here because the result table is tiny. In a real sink with many keys, complete mode would repeatedly write a large table and create unnecessary load.

Example 2: deduplicating orders with a watermark

Duplicate events are common when producers retry sends or ingestion jobs replay files. Structured Streaming can deduplicate by key, but an unbounded stream would otherwise require infinite memory. The watermark tells Spark how long to remember seen order identifiers.

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, to_timestamp
from pyspark.sql.types import StringType, StructField, StructType

spark = SparkSession.builder.appName("structured-streaming-deduplicate-orders").getOrCreate()

schema = StructType(
    [
        StructField("order_id", StringType(), False),
        StructField("customer_id", StringType(), False),
        StructField("event_time", StringType(), False),
    ]
)

orders = (
    spark.readStream.format("json")
    .schema(schema)
    .load("/tmp/orders_in")
    .select(
        col("order_id"),
        col("customer_id"),
        to_timestamp("event_time").alias("event_time"),
    )
)

deduped = orders.withWatermark("event_time", "10 minutes").dropDuplicates(["order_id"])

query = (
    deduped.writeStream.format("json")
    .outputMode("append")
    .option("path", "/tmp/orders_out")
    .option("checkpointLocation", "/tmp/orders_checkpoint")
    .start()
)

If two records with the same order_id arrive within the retained event-time horizon, only the first is written. If a duplicate arrives after Spark has evicted the key from state, it may be written again. That trade-off is intentional: bounded state in exchange for a defined lateness policy.

Example 3: stream-static enrichment before a stateful aggregate

Many data engineering streams need reference data such as customer tier, device metadata, or product category. Joining a stream to a static DataFrame is not stateful by itself, because the static side is bounded, but the downstream aggregation is stateful.

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, window

spark = SparkSession.builder.appName("structured-streaming-enriched-revenue").getOrCreate()

payments = (
    spark.readStream.format("rate")
    .option("rowsPerSecond", 3)
    .load()
    .select(
        col("timestamp").alias("event_time"),
        (col("value") % 4).cast("string").alias("customer_id"),
        ((col("value") % 20) + 1).cast("double").alias("amount"),
    )
)

customers = spark.createDataFrame(
    [("0", "standard"), ("1", "premium"), ("2", "standard"), ("3", "enterprise")],
    ["customer_id", "tier"],
)

revenue = (
    payments.join(customers, "customer_id")
    .withWatermark("event_time", "30 seconds")
    .groupBy(window(col("event_time"), "1 minute"), col("tier"))
    .sum("amount")
)

query = (
    revenue.writeStream.format("console")
    .outputMode("update")
    .option("truncate", "false")
    .option("checkpointLocation", "/tmp/revenue_checkpoint")
    .start()
)

The console will show changing revenue totals for each tier and one-minute window. Update mode writes only rows whose sums changed in the latest trigger. If the customer table changes, the running query does not automatically reread it for every row unless the code or deployment pattern refreshes the static DataFrame by restarting or using a managed dimension strategy.

Design choices and trade-offs

Choose event time when the business question is about when something happened, and processing time when the question is about when Spark observed it. Event time enables correct daily revenue, sessionization, and late-arrival handling, but requires producers to provide trustworthy timestamps. Processing time is simpler and often adequate for operational metrics such as ingestion rate.

Trigger intervals balance latency and cost. A very short trigger checks sources frequently and can create scheduling overhead if each batch contains little data. A longer trigger increases end-to-end latency but gives Spark more data per batch, often improving throughput. Available-now triggers are useful when the input is finite or when you want streaming checkpoint semantics for recurring backfills.

Checkpoint locations should be durable, unique per query, and treated as part of the application’s state. Reusing one checkpoint for incompatible query code can fail analysis or, worse, mix old state with new assumptions. Moving from one sink path to another while keeping the same checkpoint can also surprise operators because Spark believes some source offsets are already committed.

Failure modes and troubleshooting

A common symptom is a query that restarts but skips input files or Kafka offsets. The likely cause is that the checkpoint already recorded those offsets as processed. Inspect the checkpoint path, query name, sink output, and lastProgress. Correct it by restoring the intended checkpoint with the intended sink, or by starting a new checkpoint only when replaying from the source is acceptable.

Another symptom is growing state and slower batches. Causes include missing watermarks, watermarks applied after the stateful operator instead of before it, high-cardinality keys, or event times that never advance. Diagnose with the streaming progress JSON: check input rows per second, processed rows per second, state operator row counts, memory used, and watermark value. Correct by adding an event-time watermark before aggregation or deduplication, reducing key cardinality, filtering bad timestamps, or choosing a state store provider suited to larger state.

A third symptom is late data missing from aggregates. The cause is usually an allowed-lateness threshold that is too short for the real arrival distribution. Verify by sampling event_time compared with ingestion time and measuring lateness percentiles. Correct by increasing the watermark delay, improving upstream delivery, or accepting a compensating batch correction job for exceptionally late records.

Security, performance, and reliability implications

Streaming jobs often touch credentials for Kafka, object storage, and warehouses. Keep secrets outside code, use the platform secret manager, and ensure checkpoint and sink paths have restricted access because checkpoint metadata can reveal source names, offsets, and schema details. For reliability, make sink writes idempotent or transactional where possible. File sinks and Delta-style table sinks can coordinate commits; external APIs usually need explicit deduplication keys.

Performance depends on source partitioning, shuffle partitions, state size, and sink throughput. A stream with many Kafka partitions can provide parallelism, but a downstream aggregation with a small number of hot keys can still bottleneck. Monitor batch duration against trigger interval. If batch duration consistently exceeds the trigger interval, the query is falling behind even if it has not failed.

Hands-on lab

Prerequisites: a local Spark installation with PySpark available, write access to /tmp, and a shell where you can run pyspark or spark-submit. The lab builds a file-backed stream so you can control the input.

  1. Create empty directories: /tmp/lesson_orders_in, /tmp/lesson_orders_out, and /tmp/lesson_orders_checkpoint.
  2. Start the deduplication query from Example 2, changing the three paths to the lesson directories.
  3. Write a JSON file into the input directory containing two records with the same order_id and event times one minute apart.
  4. Wait for one trigger and inspect /tmp/lesson_orders_out. You should see one output order, not two.
  5. Add another file with a new order_id. You should see a second output order.
  6. Stop the query and start it again with the same checkpoint. Previously written input should not be emitted again.

Verification is based on both output files and query progress. The number of unique order identifiers in the output should match the number of non-late unique identifiers in the input. Cleanup is to stop the query and remove the three /tmp/lesson_orders_* directories. Do not remove a production checkpoint unless you intentionally want Spark to forget processed offsets and state.

Assessment exercises

  • You have a windowed aggregation whose state grows forever. Identify three possible causes and explain how you would distinguish them using query progress metrics.
  • A team wants one-second triggers for a stream that receives data every five minutes. Explain the trade-off and propose a trigger strategy.
  • Design a deduplication policy for payment events where duplicates may arrive up to six hours late but the business also receives rare corrections days later.
  • Explain why a stream-static join can still produce stale enrichment and describe one operational pattern to refresh the static side safely.
  • Given an append-mode windowed aggregation, explain why Spark needs a watermark before it can know when a result row is final.

Summary

Structured Streaming brings Spark’s table-oriented programming model to unbounded data. The important internal idea is that each trigger processes a bounded slice while checkpoints and state stores preserve continuity across slices. Stateful operators make windows, deduplication, joins, and session logic possible, but they require explicit choices about event time, watermarks, output modes, and recovery. In a data engineering system, those choices determine not only correctness, but also replay behavior, storage cost, latency, and how confidently downstream tables can be trusted.