Structured Streaming production checklist for Azure Databricks - checkpoints, rate limits, state, idempotent sinks and monitoring

#How to use this checklist

Run it before a streaming job goes to production, and again after any change to the query shape. Most streaming incidents trace back to something decided on day one: a shared checkpoint, an unbounded first batch, state that grows forever, or a sink that isn't idempotent. Each item has a check you can run.

#1. Checkpoints

• One checkpoint per query, never shared. Two queries writing to the same checkpoint corrupt each other's offsets.
• Checkpoint stored on governed storage. Use a Unity Catalog volume or an external location path. Don't use DBFS root or a path under the table directory.
• Checkpoint path derived from a stable name, such as /Volumes/ops/checkpoints/stream/<job>/<query>, not from a timestamp or run ID.
• queryName set, so metrics and the Spark UI identify the stream.
• Checkpoint storage excluded from lifecycle rules. An ADLS lifecycle policy that deletes "old" files breaks recovery.
• Change procedure written down. Know which code changes are checkpoint-compatible (filters, projections) and which are not (new aggregation keys, changed stateful operators, a different source). An incompatible change needs a new checkpoint and an explicit starting position.

q = (df.writeStream
       .queryName("orders_silver")
       .option("checkpointLocation", "/Volumes/ops/checkpoints/stream/orders/orders_silver")
       .trigger(processingTime="1 minute")
       .toTable("main.sales.orders_silver"))

#2. Source rate limits and the first batch

The first batch after a fresh start, or after a long outage, is the one that runs out of memory.

• A rate limit is set for the source:

• Starting position is explicit. Kafka: startingOffsets. Delta: startingVersion or startingTimestamp when you need a point in time.
• Kafka data-loss behaviour decided. failOnDataLoss defaults to true. Set it to false only if you accept skipping expired offsets, and alert when it happens.
• Retention is longer than your worst outage. Event Hubs retention and Delta source table log retention must outlast the longest time the stream could be down.

Warning

With availableNow, rate-limit options split the backlog into several batches but still process all of it. That's the safe way to catch up after an outage. Removing the rate limit to "go faster" is how a backlog becomes a driver OOM.

#3. Triggers and cost

• Trigger chosen against the latency requirement. Use processingTime for minutes-level latency, availableNow on a schedule for hourly or daily freshness, and the default (as fast as possible) only when you need it.
• Low-volume streams aren't on dedicated always-on clusters. Combine them on one cluster or run them as scheduled availableNow jobs.
• Sink tables have optimized writes and auto compaction, or predictive optimization. Every trigger is a commit.

#4. State

Skip this section for stateless streams: maps, filters and appends.

• Watermark on every stateful operation (aggregations, stream-stream joins, dropDuplicates). Without one, state grows forever.
• Use dropDuplicatesWithinWatermark rather than dropDuplicates when duplicates arrive within a bounded window (Spark 3.5 / DBR 13.3 LTS and above).
• RocksDB state store for anything beyond small state. It's the default in Databricks Runtime 17.3 and above. On older runtimes, set it explicitly:

spark.conf.set("spark.sql.streaming.stateStore.providerClass",
               "com.databricks.sql.streaming.state.RocksDBStateStoreProvider")
spark.conf.set("spark.sql.streaming.stateStore.rocksdb.changelogCheckpointing.enabled", "true")

• Changelog checkpointing enabled (default in 17.3 and above). It cuts batch latency for large state. Once a query runs with it on, that checkpoint needs Databricks Runtime 13.3 LTS or above.
• Shuffle partitions fixed before first start. For stateful queries, the number of state partitions is stored in the checkpoint and can't be changed later without a new checkpoint. Set spark.sql.shuffle.partitions to match the cluster size you expect at peak.
• State size tracked. stateOperators[].numRowsTotal and memoryUsedBytes in the progress event should level off, not climb forever.

#5. Sinks and exactly-once

• Delta sink via toTable. Delta sinks are exactly-once using the checkpoint.
• foreachBatch writes are idempotent. Use MERGE on a business key, or Delta idempotent writes with txnAppId and txnVersion:

def upsert(batch_df, batch_id):
    (batch_df.write.format("delta")
        .option("txnAppId", "orders_silver_upsert")   # stable per query
        .option("txnVersion", batch_id)
        .mode("append")
        .saveAsTable("main.sales.orders_audit"))

df.writeStream.foreachBatch(upsert).option("checkpointLocation", ckpt).start()

• No external side effects without dedup, such as REST calls, emails or JDBC inserts. A failed batch is retried, so plan for at-least-once.
• MERGE source deduplicated per batch. Otherwise you get multiple source rows matched.

#6. Restarts and job configuration

• Runs as a job, not on an all-purpose cluster. Use continuous mode, or a schedule with availableNow.
• Retries configured. Use unlimited or high retry counts for always-on streams. Most transient failures (spot evictions, storage throttling) recover on restart.
• max_concurrent_runs: 1. Two runs of one stream on one checkpoint fail or corrupt state.
• Schema evolution failures handled. Auto Loader in addNewColumns mode stops the stream on purpose when it sees new columns. The job retry restarts it with the new schema, so make sure retries are on.
• Runtime upgrades tested against a copy of the checkpoint in a non-production environment.

#7. Monitoring and alerting

• A progress listener or metric export sends batch duration, input rate and backlog to Log Analytics or system tables:

from pyspark.sql.streaming import StreamingQueryListener

class LagListener(StreamingQueryListener):
    def onQueryStarted(self, e): pass
    def onQueryIdle(self, e): pass
    def onQueryTerminated(self, e): print(f"terminated {e.id} {e.exception}")
    def onQueryProgress(self, e):
        p = e.progress
        for s in p.sources:
            print(p.name, p.batchId, p.durationMs.get("triggerExecution"), s.metrics)

spark.streams.addListener(LagListener())

onQueryIdle exists in recent Spark versions only. Drop it on older runtimes.

• Backlog metric chosen per source. Kafka: avgOffsetsBehindLatest and maxOffsetsBehindLatest. Delta and Auto Loader: numFilesOutstanding and numBytesOutstanding. Names vary by source and runtime, so print lastProgress once and confirm them.
• Job-level streaming backlog alerts configured in job notifications where your workspace supports them.
• Alert when batch duration exceeds the trigger interval for several batches in a row. This is the earliest sign that the stream is falling behind.

#8. Recovery drill

• Kill the job mid-batch in test and confirm the restart produces no duplicates in the sink.
• Stop the stream for longer than a normal outage, restart it, and confirm the rate limits keep catch-up batches stable.
• Document how to start from a new checkpoint at a known position, and which downstream tables need deduplication if you do.

Caution

Deleting a checkpoint to "fix" a failing stream reprocesses from the configured starting position. That can mean duplicates in non-idempotent sinks, or a full re-read of the source. Copy the checkpoint aside first, and decide the new starting position on purpose.