Structured Streaming in Production: Triggers, Checkpoints, Watermarks, and Exactly-Once Sinks

The Spark stream that has caused me the most grief did not fail. It ran for eleven months, then someone changed a groupBy key, and it would not restart. The checkpoint held state that no longer matched the query plan, the error message was about an incompatible state schema, and the only options were replaying five weeks of Kafka or accepting a gap.

That is what production streaming is actually about. The API is genuinely elegant — a streaming DataFrame is the same DataFrame, readStream and writeStream bookend the same transformations you already know. What the API hides is that a stream is a long-lived stateful process with a durable identity on disk, and every choice you make on day one is a constraint on day four hundred.

Four decisions dominate: how often the stream runs, where and how it checkpoints, how much lateness it tolerates, and what its sink guarantees. Get those right and the stream is boring. Get any of them wrong and you learn about it during an incident.

Here is how I set each one, with the numbers I use and the specific mistakes I have made.

Triggers: most streams should not be always-on

Structured Streaming is micro-batch by default. The trigger decides when the next batch starts, and it is the single biggest cost lever you have.

# 1. Fixed interval: a batch every 60 seconds, cluster always up
q1 = (df.writeStream
        .trigger(processingTime="60 seconds")
        .option("checkpointLocation", "s3://lake/_ckpt/orders_60s/")
        .toTable("prod.silver.orders"))

# 2. AvailableNow: process everything pending, then stop.
#    Run this from a scheduled job every 15 minutes.
q2 = (df.writeStream
        .trigger(availableNow=True)
        .option("checkpointLocation", "s3://lake/_ckpt/orders_batch/")
        .toTable("prod.silver.orders"))

# 3. Continuous: sub-second latency, very limited operator support
q3 = (df.writeStream
        .trigger(continuous="1 second")
        .format("kafka")
        .option("checkpointLocation", "s3://lake/_ckpt/orders_cont/")
        .start())

I use option 2 for the large majority of pipelines, and it is the recommendation I most often have to argue for. availableNow processes all available data in rate-limited micro-batches and then terminates, which means the cluster shuts down. A stream that runs for three minutes every fifteen costs about 20 percent of an always-on equivalent, and it keeps all the incremental semantics: checkpoints, offset tracking, exactly-once writes.

The honest question is what latency the business needs. I have audited perhaps thirty streaming pipelines and found maybe four where a sub-minute SLA was real. The rest were “streaming” because the previous engineer chose streaming, and a 15-minute scheduled availableNow job would have satisfied every consumer.

Fixed-interval triggers earn their cost when the freshness requirement is genuinely under five minutes, or when the stateful operators need continuously warm state. Continuous processing supports only map-like operations, no aggregations, and I have never shipped it.

Rate limiting matters at every trigger type. On Kafka use maxOffsetsPerTrigger; on file sources use maxFilesPerTrigger or maxBytesPerTrigger. Without them, the first batch after a backlog tries to read everything and OOMs, which is really a partition-sizing failure with a streaming label on it.

The checkpoint is the stream

The checkpoint directory holds four things, and I want you to know all four because each one causes a distinct incident:

• offsets/ — for each batch, which source positions it covers. Written before the batch runs.
• commits/ — which batches completed. Written after.
• state/ — the actual keyed state for aggregations, joins, and flatMapGroupsWithState, as RocksDB or HDFS-backed files.
• metadata — the stream’s ID and query configuration.

sequenceDiagram
  participant S as Source
  participant E as Spark engine
  participant C as Checkpoint
  participant K as Sink
  E->>S: request available offsets
  E->>C: write offsets for batch N
  E->>S: read data for batch N
  E->>E: apply transformations and state
  E->>K: write batch N with batch ID
  K-->>E: ack
  E->>C: write commit for batch N
  Note over E,C: crash before commit means batch N replays

Offsets before, commit after. That ordering is what makes replay safe and why sinks must be idempotent.

Rules I treat as non-negotiable:

One checkpoint per stream, in durable storage, never shared. Two queries pointing at one checkpoint location corrupt each other in ways that take hours to understand.

Checkpoint path is part of the deployment, not a temp dir. If the path lives under a workspace-local or ephemeral filesystem, a cluster replacement silently resets the stream to the beginning or to the latest offset, depending on your startingOffsets.

Query changes are constrained. You can safely change filters, add columns to a projection, and adjust most options. You cannot change the aggregation keys, the output schema of a stateful operator, or the type of a stateful operation and still resume. Spark will refuse, and refusing is the kind behavior — the alternative is wrong results.

When I do need a breaking change, the migration is: start a second stream with a new checkpoint writing to the same table with a distinct source range, or backfill with a batch job and start the new stream from a known offset. Never delete the checkpoint of a running production stream and hope.

For stateful streams, use RocksDB. It has been the better default for years because state lives off-heap and GC stops being the limiting factor:

spark.conf.set(
    "spark.sql.streaming.stateStore.providerClass",
    "org.apache.spark.sql.execution.streaming.state.RocksDBStateStoreProvider",
)
# Changelog checkpointing uploads deltas instead of full snapshots
spark.conf.set("spark.sql.streaming.stateStore.rocksdb.changelogCheckpointing.enabled", "true")

Watermarks: a promise Spark enforces by dropping data

Any stateful operator on event time needs to know when it can stop waiting for late records and release state. That is the watermark: maximum event time seen so far, minus your configured delay.

from pyspark.sql import functions as F

events = (spark.readStream
    .format("kafka")
    .option("kafka.bootstrap.servers", "broker:9092")
    .option("subscribe", "events")
    .option("maxOffsetsPerTrigger", 500000)   # rate limit, always
    .load()
    .select(F.from_json(F.col("value").cast("string"), SCHEMA).alias("e"))
    .select("e.*")
    .withColumn("event_ts", F.to_timestamp("event_ts"))
)

windowed = (events
    .withWatermark("event_ts", "30 minutes")     # measured, not guessed
    .groupBy(F.window("event_ts", "5 minutes"), "region")
    .agg(F.count("*").alias("events"),
         F.approx_count_distinct("user_id").alias("users"))
)

Two things follow that people learn the hard way. First, records arriving with event time older than the watermark are silently dropped in append mode. No error, no metric that says “I threw away your data” other than the state metrics. Second, the window result is not emitted until the watermark passes the window end, so a 5-minute window with a 30-minute watermark means results are available 35 minutes after the window opens.

Pick the delay from measured lag, not intuition. I run this against a week of historical data before choosing:

-- Observed lateness distribution: pick the watermark from p99, not p50
SELECT
  percentile_approx(unix_timestamp(ingest_ts) - unix_timestamp(event_ts), 0.50) AS p50_s,
  percentile_approx(unix_timestamp(ingest_ts) - unix_timestamp(event_ts), 0.99) AS p99_s,
  percentile_approx(unix_timestamp(ingest_ts) - unix_timestamp(event_ts), 0.999) AS p999_s,
  max(unix_timestamp(ingest_ts) - unix_timestamp(event_ts)) AS max_s
FROM prod.bronze.events_raw
WHERE ingest_date >= current_date() - INTERVAL 7 DAYS;

If p99 lateness is 4 minutes and the max is 6 hours because of a mobile client that batches uploads offline, do not set a 6-hour watermark to catch everything — that holds six hours of state and delays every result. Set 30 minutes, accept the drop, and handle the genuinely late tail with a daily batch reconciliation job. That split is far cheaper than making the stream tolerate the worst case.

For stream-stream joins, both sides need watermarks plus a time range condition, or state grows without bound. Getting this wrong is how a stream that ran fine for two weeks starts OOMing.

Exactly-once, precisely

“Exactly-once” in Structured Streaming means end-to-end, and it requires three cooperating parts:

1. A replayable source with tracked offsets: Kafka, Kinesis, file sources, Delta.
2. Checkpointed offsets written before the batch.
3. An idempotent sink that can recognize a replayed batch.

Delta and Parquet file sinks get this natively — the sink writes with a batch ID and the transaction log deduplicates a replay. That mechanism is the txn action described in the transaction log internals.

foreachBatch is where guarantees leak, because you are now responsible for idempotency:

def upsert_batch(batch_df, batch_id: int):
    """Called once per micro-batch. May be called twice for the same
    batch_id after a failure, so everything here must be idempotent."""
    from delta.tables import DeltaTable

    target = DeltaTable.forName(spark, "prod.silver.customers")

    # MERGE is naturally idempotent on a stable key
    (target.alias("t")
        .merge(batch_df.dropDuplicates(["customer_id"]).alias("s"),
               "t.customer_id = s.customer_id")
        .whenMatchedUpdateAll(condition="s.updated_at > t.updated_at")
        .whenNotMatchedInsertAll()
        .execute())

(events.writeStream
    .foreachBatch(upsert_batch)
    .option("checkpointLocation", "s3://lake/_ckpt/customers_merge/")
    .trigger(availableNow=True)
    .start())

A MERGE on a stable key is idempotent, so a replayed batch is harmless. An INSERT is not. An HTTP POST to a payments API is emphatically not, and if you must call an external service, you need your own dedupe table keyed on batch_id — the sink guarantee is at-least-once and no configuration changes that.

Also note the MERGE predicate. Scoping the merge narrowly matters for concurrency: an unpartitioned full-table merge fighting a 60-second trigger loses optimistic concurrency races, which is one of the failure modes I described in the transaction log post.

Files, state, and the operational surface

Two things grow whether you watch them or not.

File count. Every micro-batch commits at least one file per partition it touches. A 60-second trigger writing to 24 hourly partitions produces up to 34,560 files a day. Turn on optimized writes and auto compaction from the first deploy:

ALTER TABLE prod.silver.orders SET TBLPROPERTIES (
  'delta.autoOptimize.optimizeWrite' = 'true',
  'delta.autoOptimize.autoCompact'   = 'true',
  'delta.targetFileSize'             = '128mb'
);

State size. Watch stateOperators in the progress payload. If numRowsTotal grows monotonically across days, either your watermark is too generous or a stream-stream join is missing its time bound. I alert on state row count growth rate, not absolute size, because the trend catches it a week before the OOM.

The progress listener is where monitoring actually lives:

q = writer.start()

# Poll from a monitoring job, or register a StreamingQueryListener
p = q.lastProgress
print(p["batchId"], p["numInputRows"], p["durationMs"]["triggerExecution"])
print(p["sources"][0].get("numInputRows"))
# Kafka lag: endOffset minus latestOffset per partition
print(p["sources"][0].get("metrics", {}))
for op in p.get("stateOperators", []):
    print(op["numRowsTotal"], op["memoryUsedBytes"])

The three metrics I alert on: batch duration exceeding the trigger interval (falling behind), input rows at zero for longer than expected (upstream broken, and the stream will not tell you), and state rows trending up. Kafka consumer lag from the broker side is the fourth, and it belongs on the same dashboard as your Kafka fundamentals monitoring.

Pitfalls

Sharing or moving a checkpoint location. Two streams on one checkpoint corrupt state. A checkpoint on ephemeral storage resets the stream on cluster replacement. Both are silent.

Setting the watermark from the maximum observed lateness. You pay for the worst case in state size and result latency, forever. Use p99 plus a batch reconciliation for the tail.

Non-idempotent work in foreachBatch. Plain inserts, external API calls, and counter increments all double on replay. Assume the function runs twice for the same batch_id and design for it.

No rate limit on the source. The first batch after a backlog or a fresh startingOffsets=earliest tries to read everything. Set maxOffsetsPerTrigger or maxFilesPerTrigger before you ever hit the situation.

A one-minute trigger into many partitions with no compaction. The file count math is multiplicative and it degrades read performance for every downstream consumer within weeks.

Assuming a stateful query can be edited freely. Changing aggregation keys, output modes, or state schema breaks resumption. Plan the migration path before shipping the first version.

FAQ

Should I use availableNow or a continuous 60-second trigger?

Default to availableNow on a schedule unless the freshness requirement is under five minutes. You get the same incremental correctness at roughly a fifth of the compute cost, and a failed run is a normal job failure rather than a stalled long-lived process.

What happens to data later than the watermark?

In append mode it is dropped without an error. In update mode it can still update an existing window if state has not been evicted. Either way, if the late tail is business-relevant, reconcile it with a periodic batch job rather than widening the watermark.

How large can streaming state get before it is a problem?

With RocksDB and changelog checkpointing I have run tens of millions of keys per partition comfortably. The failure signal is not absolute size but growth rate: bounded state plateaus, unbounded state climbs forever, and only the second one pages you.

Can I run multiple streams on one cluster?

Yes, and it is often the right cost decision, but each needs its own checkpoint and they compete for cores. Use spark.scheduler.mode=FAIR with pools so one backlogged stream cannot starve the others.

Is Auto Loader different from a file source stream?

It is a file source with a scalable file-discovery mechanism — notification-based rather than directory listing — so it stays fast on directories with millions of files. The trigger, checkpoint, and watermark reasoning in this post applies unchanged.

What this means for your pipelines

Start every streaming design with the latency question, answered by a consumer rather than an engineer. In my experience four out of five “streaming” requirements are satisfied by a 15-minute availableNow job, and that choice removes an always-on cluster, simplifies failure handling into ordinary job retries, and still gives you exactly-once incremental processing.

Then treat the checkpoint as production state with the same care as a database. It goes in durable storage at a path recorded in your deployment config, one per stream, and any change to a stateful query gets a written migration plan before it merges. Half the streaming incidents I have worked were checkpoint incidents wearing a different hat.

Finally, make the invisible things visible. Watermark drops, state row growth, batch duration against trigger interval, and file counts per table are all trends that turn into incidents weeks later. Put them on a dashboard on day one, keep table maintenance running alongside the stream, and streaming becomes the boring part of the platform — which is exactly what you want from something that never stops running.