Spark Structured Streaming and Exactly-Once

Micro-batches, checkpointing, watermarks, and late data — the concepts behind cutting a pipeline from hours to minutes, and how to explain that work in an interview.

Structured Streaming's core idea is simple: treat a stream as a table that keeps growing. You write almost the same DataFrame code as batch, and Spark runs it repeatedly over new data.

A typical real-time network analytics pipeline — the shape behind most 'reduced latency from hours to minutes' stories.

A minimal but production-shaped job

from pyspark.sql import functions as F
from pyspark.sql.types import StructType, StringType, TimestampType, DoubleType

schema = (StructType()
    .add("subscriber_id", StringType())
    .add("cell_site_id", StringType())
    .add("event_ts", TimestampType())
    .add("volume_gb", DoubleType()))

raw = (spark.readStream
    .format("kafka")
    .option("kafka.bootstrap.servers", "broker:9092")
    .option("subscribe", "network-events")
    .option("startingOffsets", "latest")
    .option("maxOffsetsPerTrigger", 500000)     # backpressure control
    .load())

events = (raw
    .select(F.from_json(F.col("value").cast("string"), schema).alias("d"))
    .select("d.*")
    .withWatermark("event_ts", "10 minutes"))    # how late data may arrive

agg = (events
    .groupBy(F.window("event_ts", "5 minutes"), "cell_site_id")
    .agg(F.sum("volume_gb").alias("total_gb"),
         F.approx_count_distinct("subscriber_id").alias("subs")))

query = (agg.writeStream
    .outputMode("append")
    .format("bigquery")
    .option("checkpointLocation", "gs://bucket/checkpoints/network-kpi")  # REQUIRED
    .trigger(processingTime="1 minute")
    .start())

Watch out — The checkpoint is not optional

The checkpoint directory stores offsets and aggregation state. Without it there is no fault tolerance and no exactly-once. Never delete or share a checkpoint between jobs — and know that changing the query in incompatible ways invalidates it.

Event time, processing time, and watermarks

Concept	Meaning	Why it matters
Event time	When the event actually happened (in the record)	The only correct basis for business aggregates
Processing time	When Spark saw the record	Easy but wrong — a network delay would shift your metrics
Watermark	'I will wait up to X for late data, then finalise the window'	Bounds state size so the job does not grow forever

A watermark is a trade-off you choose: a longer watermark catches more late records but holds state longer and delays results. A shorter one is faster and cheaper but drops stragglers. Say that trade-off out loud — interviewers are listening for it.

Output modes

Mode	Writes	Use with
Append	Only new, finalised rows	Windowed aggregates with a watermark; immutable sinks
Update	Only rows that changed this batch	Live dashboards backed by an upsertable store
Complete	The entire result table every batch	Small aggregates only — state grows unbounded

How exactly-once actually works

Spark records the exact Kafka offset range for each micro-batch in the checkpoint before processing
If the job crashes mid-batch, it restarts and replays that exact same range — this gives replayable input
The sink must then be idempotent or transactional, so re-writing the same batch does not duplicate rows
Together: replayable source + deterministic processing + idempotent sink = effectively exactly-once

Tip — Making a BigQuery sink idempotent

Use foreachBatch and MERGE on a deterministic key (for example a hash of subscriber_id + window_start), or write to a staging table partitioned by batch id and overwrite that partition. Both make a replayed batch a no-op instead of a duplicate.

Choosing streaming at all

Streaming is an operational cost, not a badge of modernity. Justify it with a latency requirement.

Practice questions

QYou cut end-to-end latency from hours to minutes. Walk me through what was slow and what you changed.show

Structure it as: where the time went → what you replaced → how you proved it. This is the single most likely question from a streaming-heavy résumé.

Before — batch ingestion on a schedule: files landed hourly, a Hadoop job ran, results appeared hours later. Latency was dominated by *waiting for the next scheduled run*, not by compute
Change — event-driven ingestion into Kafka, Spark Structured Streaming with a short trigger interval, writing to partitioned warehouse tables consumed directly by dashboards
Correctness — event-time windows with watermarks so late records still land in the right window; idempotent sink so retries are safe
Proof — measured consumer lag and an end-to-end timestamp (event time → queryable time), tracked on a Grafana dashboard before and after
Trade-off named honestly — higher operational complexity and always-on cost, justified by an alerting use case that needed minutes, not hours

Likely follow-up: What broke first when you went live, and how did you find it?

QHow do you handle late-arriving data?show

Decide a lateness budget with the business, encode it as a watermark, and have an explicit plan for anything later than that.

Within the watermark → included automatically in the correct event-time window
Beyond the watermark → dropped by the streaming job; capture these to a side table and count them
For material corrections, run a periodic batch reconciliation that recomputes recent days from the raw log and overwrites those partitions
This 'streaming for speed, batch for truth' pattern is worth naming explicitly

QWhat is stateful streaming and what goes wrong with it?show

Operations like windowed aggregation, deduplication and stream-stream joins must remember past data — that memory is state, stored in the state store and checkpointed.

Unbounded state growth is the classic failure — no watermark means state never expires and the job eventually dies
Large state slows every micro-batch because it is written to the checkpoint each time
Stream-stream joins need watermarks on both sides plus a time constraint in the join condition
Mitigations: always watermark, keep keys coarse, consider RocksDB state store for large state

QMicro-batch vs continuous processing — which do you use and why?show

Structured Streaming's default micro-batch model processes small batches every trigger interval, giving latency in the hundreds-of-milliseconds to seconds range with full exactly-once support.

Micro-batch — mature, supports all operations and aggregations, exactly-once; latency bounded by trigger interval
Continuous processing — ~1 ms latency but experimental, at-least-once only, and supports only simple map-like operations
In practice micro-batch with a short trigger is the right answer for nearly every analytics pipeline; sub-second hard real time is usually a Flink or in-app decision instead

← PreviousKafka Deep Dive for Interviews Next →BigQuery Deep Dive: Performance and Cost