Spark Structured Streaming: patterns, pitfalls and fixes

Introduction

Stream processing at scale demands more than just subscribing to a Kafka topic and transforming rows. Spark Structured Streaming: patterns, pitfalls and fixes covers the real patterns teams use in production, from reading continuous streams through to writing results with guarantees that actually hold. We've learned these lessons operating stateful pipelines across trading and ride-hailing platforms, where a missing checkpoint or incorrect trigger can silently lose data or cause backpressure that cascades across your entire infrastructure.

This post walks through the core mechanics of Structured Streaming, then zooms into the choices and mistakes that separate a prototype from a system you'd trust with revenue data.

Reading a stream: connecting Kafka, file systems and other sources

Structured Streaming reads from a source using the same DataFrame API you already know. The engine treats the source as an unbounded table that grows over time. Here's the practical starting point—a Kafka reader:

events = (spark.readStream
  .format("kafka")
  .option("kafka.bootstrap.servers", "broker:9092")
  .option("subscribe", "events")
  .load())

That gives you a DataFrame with columns key, value, topic, partition, offset, and timestamp. The value is raw bytes; you'll parse it as JSON or Avro downstream.

In production, you'll also want to set consumer-group isolation and bootstrap timeout:

events = (spark.readStream
  .format("kafka")
  .option("kafka.bootstrap.servers", "broker:9092")
  .option("subscribe", "events")
  .option("startingOffsets", "latest")
  .option("failOnDataLoss", "false")
  .load())

The startingOffsets option controls where the stream begins—"latest" starts from new messages, while "earliest" replays from the beginning. failOnDataLoss defaults to true, which crashes the stream if Kafka deletes old offsets before Spark has checkpointed them; set it to false for clusters where log retention is aggressive.

File sources work similarly. For a directory of JSON files that grow over time:

events = (spark.readStream
  .format("json")
  .schema(schema)
  .load("/data/events/incoming/"))

Spark watches the directory and picks up new files automatically. This is useful for testing or when your data producer writes to cloud storage rather than a message broker.

For socket connections, rate limits, and custom sources, check the Spark documentation—but Kafka is where the real work happens at scale.

Transform: leverage the familiar DataFrame API

The powerful part of Structured Streaming is that your transformations don't change syntax. You write the same select(), filter(), groupBy() and join() calls as in batch jobs. The engine optimizes them to run incrementally.

Parse JSON, add business logic, and aggregate:

from pyspark.sql.functions import col, from_json, window, count, sum as spark_sum

schema = "user_id STRING, amount DOUBLE, ts TIMESTAMP"

parsed = (events
  .select(from_json(col("value").cast("string"), schema).alias("data"))
  .select("data.*"))

revenue = (parsed
  .filter(col("amount") > 0)
  .groupBy(
    window(col("ts"), "1 minute", "30 seconds"),
    col("user_id")
  )
  .agg(
    count("*").alias("txn_count"),
    spark_sum("amount").alias("total")
  ))

This creates a tumbling 1-minute window with 30-second overlap, grouped by user. Each micro-batch computes only the new data and merges it with state.

Stateful operations like groupBy and join are where Structured Streaming shines—and where you need checkpoints. We'll cover that next.

One gotcha: rdd.foreach() and other RDD operations break streaming logic. Stick to the DataFrame API. If you must do custom logic, use foreachBatch() in the write phase, not transforms.

Write with checkpoints: making streams recoverable

A checkpoint is a directory where Spark writes its state and progress. If your driver or executor crashes, Structured Streaming resumes from the last checkpoint, replaying only the micro-batches that haven't been written yet.

Without a checkpoint, a stateful stream like groupBy() loses intermediate aggregations on restart, corrupting results. With one, you get exactly-once semantics (assuming your sink is idempotent).

Write to Kafka with a checkpoint:

query = (revenue
  .writeStream
  .format("kafka")
  .option("kafka.bootstrap.servers", "broker:9092")
  .option("topic", "revenue_aggregated")
  .option("checkpointLocation", "/checkpoints/revenue_agg")
  .start())

query.awaitTermination()

The checkpoint directory holds:

• offsets – Kafka offset ranges processed in each batch
• state – Aggregation keys and values (for stateful ops)
• metadata – Batch IDs, timestamps, schema

Never point two queries at the same checkpoint. If you restart a query, reuse the same checkpoint path. If you want a fresh start, delete the checkpoint and accept that you'll reprocess data or miss some.

In a SaaS environment, store checkpoints on a durable filesystem—S3, HDFS, or a network mount. Local disk checkpoints are fine for dev but will fail in a cluster when the driver restarts on a different node.

Checkpoint failure is silent if permissions are wrong. Test checkpoint writes early. A common mistake is using a glob pattern like /checkpoints/*—Spark doesn't support that; use a specific directory per query.

Triggers: controlling batch intervals and latency

A trigger defines how often the stream processes new data. Spark supports four modes:

• Default (micro-batch) – As soon as the previous batch completes, start a new one
• Fixed interval – Process every N seconds, even if no data arrived
• Continuous – Near real-time, single-pass latency (experimental)
• availableNow – Process all available data once, then stop

For most use cases, the default micro-batch is right. It balances latency and throughput:

query = (revenue
  .writeStream
  .format("kafka")
  .option("checkpointLocation", "/checkpoints/revenue_agg")
  .start())

If you want explicit control, set a 10-second interval:

from pyspark.sql.streaming import Trigger

query = (revenue
  .writeStream
  .format("kafka")
  .option("checkpointLocation", "/checkpoints/revenue_agg")
  .trigger(Trigger.ProcessingTime("10 seconds"))
  .start())

Fixed intervals reduce checkpoint overhead but increase latency if data arrives sporadically. For event-driven pipelines (e.g., fraud detection), stick with the default. For batch-like loads every hour, use ProcessingTime("1 hour").

availableNow is handy for backfilling or one-off jobs. It drains the source entirely in a single run:

query = (revenue
  .writeStream
  .format("parquet")
  .option("path", "/data/revenue_backfill")
  .option("checkpointLocation", "/checkpoints/backfill")
  .trigger(Trigger.AvailableNow())
  .start())

query.awaitTermination()

Continuous triggers are available but rarely used—they break some stateful operations and add operational complexity.

Common mistakes

No checkpoint, then wondering where data went. Stateful operations without checkpoints silently lose state on restart. Always set checkpointLocation for any query with groupBy, join, or window.

Reusing checkpoints across different query definitions. If you change your SQL logic, the old state is incompatible. Delete the checkpoint or use a new path. Spark won't warn you; results will be corrupted.

Sinks that aren't idempotent. Exactly-once guarantees only work if your sink can handle duplicate writes (e.g., upsert into a database, not append to a file). If your sink is append-only, you'll get duplicates on restart.

Ignoring backpressure.** If your sink is slow, Spark queues micro-batches in memory. This can exhaust heap and crash the driver. Monitor query progress using query.lastProgress and scale your sinks appropriately.

Parsing failures blocking the entire stream. If a single malformed JSON message arrives, the whole stream crashes by default. Use from_json(..., mode="PERMISSIVE") and filter nulls, or catch exceptions in a custom sink.

Key takeaways

• Read unbounded sources (Kafka, files) using readStream; the DataFrame API is identical to batch mode.
• Checkpoint every query with stateful operations; store checkpoints on durable shared storage in production.
• Use default micro-batch triggers for event-driven pipelines; fixed intervals for periodic loads.
• Exactly-once semantics require idempotent sinks and proper checkpointing—verify both in testing.
• Monitor query progress and backpressure; streaming queries can silently degrade under load.
• Avoid RDD operations and custom foreach logic in transforms; use foreachBatch() only in the write phase.