Failure Recovery Procedures

Datanest Digital — https://datanest.dev

Overview

Streaming pipelines fail. The difference between a minor incident and a major outage is how quickly and correctly you recover. This guide provides step-by-step recovery procedures for the most common failure scenarios in Databricks Structured Streaming and Delta Live Tables.

Each section follows the same structure: symptoms, root cause, impact assessment, recovery steps, and prevention measures.

Scenario 1: Checkpoint Corruption

Symptoms

• Stream fails on restart with StreamingQueryException
• Error messages referencing OffsetOutOfRange, FileNotFoundException in the checkpoint path, or metadata file is corrupted
• Stream starts but immediately stops after batch 0

Root Cause

• Hard cluster termination during a micro-batch commit
• Storage system returning partial writes (rare but possible with object storage under load)
• Manual modification or deletion of checkpoint files

Impact Assessment

Recovery Steps

Step 1: Stop the stream and back up the checkpoint.

for q in spark.streams.active:
    if q.name == "your_stream_name":
        q.stop()
        q.awaitTermination(timeout=60)

from datetime import datetime
ts = datetime.now().strftime("%Y%m%d_%H%M%S")
dbutils.fs.cp(checkpoint_path, f"{checkpoint_path}_backup_{ts}", recurse=True)

Step 2: Diagnose the corruption.

def diagnose_checkpoint(path: str) -> dict:
    result = {}
    try:
        result["metadata"] = dbutils.fs.head(f"{path}/metadata", 1024)
    except:
        result["metadata"] = "MISSING_OR_CORRUPT"

    for subdir in ["offsets", "commits"]:
        try:
            files = dbutils.fs.ls(f"{path}/{subdir}")
            ids = sorted([int(f.name) for f in files])
            result[subdir] = {"count": len(ids), "range": f"{min(ids)}-{max(ids)}" if ids else "empty"}
        except:
            result[subdir] = "MISSING"

    return result

import json
print(json.dumps(diagnose_checkpoint(checkpoint_path), indent=2))

Step 3: Apply the appropriate repair.

• Missing commit file: Create an empty file at the missing batch ID (see config/checkpoint_management.md for details).
• Corrupt metadata: Regenerate the metadata JSON with the original stream UUID.
• Corrupt state: Delete the state/ directory; the stream will rebuild state from scratch.
• Irrecoverable: Archive the checkpoint and restart from startingOffsets.

Step 4: Restart the stream and verify.

# After repair, restart the stream
query = (
    stream_df.writeStream
    .option("checkpointLocation", checkpoint_path)
    .start()
)

# Verify the first few batches complete successfully
import time
time.sleep(30)
print(json.dumps(query.lastProgress, indent=2, default=str))

Prevention

• Store checkpoints on durable cloud object storage (ADLS Gen2, S3, GCS) — never ephemeral DBFS
• Enable spark.sql.streaming.minBatchesToRetain=100 for automatic checkpoint compaction
• Monitor checkpoint health using the dashboard queries in monitoring/streaming_dashboard.sql

Scenario 2: Schema Evolution

Symptoms

• AnalysisException: Cannot resolve column name after source schema changes
• mergeSchema errors on Delta writes
• Deserialization returns null for new fields; data silently lost

Root Cause

• Upstream producer adds, removes, or renames fields without coordination
• Schema registry version mismatch between producer and consumer
• Source table schema changed via DDL while stream is running

Impact Assessment

Recovery Steps

Step 1: Identify the schema change.

# Compare current source schema to the schema in the streaming query
source_df = spark.read.table("analytics.bronze.raw_events")
source_df.printSchema()

# Check the target table schema
target_df = spark.read.table("analytics.silver.processed_events")
target_df.printSchema()

Step 2: Handle the change based on type.

For new fields (additive change):

# Ensure mergeSchema is enabled on the sink
query = (
    stream_df.writeStream
    .option("mergeSchema", "true")
    .start()
)

For renamed or removed fields:

# Update the stream's transformation logic to map old name to new name
stream_df = stream_df.withColumnRenamed("old_field", "new_field")

# Or handle missing fields gracefully
from pyspark.sql.functions import lit
if "removed_field" not in stream_df.columns:
    stream_df = stream_df.withColumn("removed_field", lit(None).cast("string"))

For type changes:

# Cast the field to the expected type with error handling
from pyspark.sql.functions import col, when
stream_df = stream_df.withColumn(
    "amount",
    when(col("amount").cast("double").isNotNull(), col("amount").cast("double"))
    .otherwise(None)
)

Step 3: Reset the checkpoint if the schema change is incompatible with the existing state.

Stateful operations (aggregations, dedup) store data in the old schema. If the schema change affects columns used in state, you must reset:

# Archive old checkpoint
dbutils.fs.mv(checkpoint_path, f"{checkpoint_path}_pre_schema_change", recurse=True)

# Restart stream — will reprocess from startingOffsets

Step 4: Backfill if needed.

If the stream was down during the schema transition, use availableNow to catch up:

query = (
    stream_df.writeStream
    .trigger(availableNow=True)
    .option("checkpointLocation", checkpoint_path)
    .start()
)
query.awaitTermination()

Prevention

• Use a schema registry (Confluent, Azure Schema Registry) with compatibility enforcement
• Set mergeSchema=true on all streaming sinks as a default
• Test schema changes in staging before applying to production sources
• Add schema validation expectations in your DLT pipelines (see dlt/expectations_library.py)

Scenario 3: Out of Memory (OOM)

Symptoms

• Executor lost: ExecutorLostFailure (executor X exited caused by one of the running tasks) Reason: Container killed by YARN for exceeding memory limits
• java.lang.OutOfMemoryError: Java heap space or GC overhead limit exceeded
• Streaming query stalls with increasing batch durations before eventual failure

Root Cause

• State store growth exceeding executor memory (most common)
• Single micro-batch too large (spike in source data volume)
• Skewed partitions concentrating data on a single executor
• Insufficient memory overhead for RocksDB native memory

Impact Assessment

OOM failures cause the micro-batch to fail and retry. With checkpointing, no data is lost, but the pipeline stalls until the OOM condition is resolved. Repeated OOM on the driver can crash the entire cluster.

Recovery Steps

Step 1: Identify the memory consumer.

# Check state store size
progress = query.lastProgress
for op in progress.get("stateOperators", []):
    print(f"State rows: {op.get('numRowsTotal', 0)}")
    print(f"Memory used: {op.get('memoryUsedBytes', 0) / 1024 / 1024:.1f} MB")

# Check micro-batch size
print(f"Input rows: {progress.get('numInputRows', 0)}")
print(f"Batch size bytes: {progress.get('sources', [{}])[0].get('numInputRows', 0)}")

Step 2: Apply immediate mitigation.

For state store OOM — tighten the watermark:

# Reduce watermark from 24 hours to 1 hour (evicts old state faster)
stream_df = stream_df.withWatermark("event_timestamp", "1 hour")

For batch size OOM — limit data per trigger:

# Kafka: reduce max offsets per trigger
.option("maxOffsetsPerTrigger", "10000")  # was 100000

# Auto Loader: reduce max files per trigger
.option("cloudFiles.maxFilesPerTrigger", "100")  # was 1000

For executor OOM — increase memory:

spark.executor.memory=32g
spark.executor.memoryOverhead=8g
spark.sql.streaming.stateStore.rocksdb.blockCacheSizeMB=512

Step 3: Switch to RocksDB state store if using the default provider.

spark.sql.streaming.stateStore.providerClass=com.databricks.sql.streaming.state.RocksDBStateStoreProvider

RocksDB stores state on local SSD instead of the JVM heap, dramatically reducing OOM risk for stateful operations.

Step 4: Restart the stream.

After applying config changes, restart the cluster and the streaming query. The query will resume from the last committed checkpoint.

Prevention

• Always use RocksDB state store for stateful streaming operations
• Set maxOffsetsPerTrigger or equivalent to cap micro-batch size
• Monitor state store growth and set alerts (see monitoring/streaming_alerts.sql)
• Use memory-optimized instance types for stateful workloads (see config/autoscaling_config.md)
• Configure watermarks as tight as your late-arrival tolerance allows

Scenario 4: Data Source Unavailability

Symptoms

• KafkaException: Failed to construct kafka consumer or TimeoutException
• EventHubException: The connection was inactive for more than the allowed idle timeout
• Auto Loader stream stops progressing; no new files detected
• query.status shows isTriggerActive: false with no errors (silent stall)

Root Cause

• Source system outage (Kafka broker down, Event Hub throttled, storage account unreachable)
• Network partition between Databricks cluster and source
• Authentication credential expiration (SASL token, SAS token, service principal secret)
• Source topic/container deleted or renamed

Impact Assessment

Structured Streaming will retry indefinitely on transient source failures. The stream pauses but does not lose its checkpoint position. Data produced during the outage accumulates at the source and will be processed when connectivity resumes (assuming source retention is sufficient).

Recovery Steps

Step 1: Verify source availability.

# Kafka — test broker connectivity
import socket
broker_host, broker_port = kafka_brokers.split(",")[0].split(":")
try:
    sock = socket.create_connection((broker_host, int(broker_port)), timeout=10)
    sock.close()
    print("Kafka broker reachable")
except Exception as e:
    print(f"Kafka broker unreachable: {e}")

# Auto Loader — verify storage path
try:
    files = dbutils.fs.ls(source_path)
    print(f"Storage accessible: {len(files)} files found")
except Exception as e:
    print(f"Storage unreachable: {e}")

Step 2: Check for credential expiration.

# Test secret access
try:
    secret = dbutils.secrets.get(scope="kafka", key="sasl_password")
    print("Secret accessible (value redacted)")
except Exception as e:
    print(f"Secret access failed: {e}")

If credentials have expired, rotate them in the Databricks secret scope and restart the stream.

Step 3: Check the streaming query status.

for q in spark.streams.active:
    print(f"Query: {q.name}")
    print(f"  Active: {q.isActive}")
    print(f"  Status: {q.status}")
    if q.exception():
        print(f"  Exception: {q.exception()}")

Step 4: Handle extended outages.

If the source was down longer than its retention period, data may be lost. For Kafka:

# Check if offsets are still available
# If you get OffsetOutOfRange on restart, you must decide:
# Option A: Reset to earliest available offset (may cause duplicates)
.option("startingOffsets", "earliest")
# Note: this only applies to a NEW checkpoint. For existing checkpoints,
# set failOnDataLoss=false to skip past unavailable offsets:
.option("failOnDataLoss", "false")

Step 5: Restart and monitor catch-up.

After source recovery, monitor the stream's catch-up progress:

import time
for _ in range(10):
    time.sleep(30)
    progress = query.lastProgress
    if progress:
        input_rate = progress.get("inputRowsPerSecond", 0)
        process_rate = progress.get("processedRowsPerSecond", 0)
        print(f"Input: {input_rate:.0f} rows/sec | Processed: {process_rate:.0f} rows/sec")

When processedRowsPerSecond stabilizes near inputRowsPerSecond, the backlog is cleared.

Prevention

• Set failOnDataLoss=false for Kafka sources in production to survive offset expiration gracefully
• Configure source retention to exceed your maximum expected outage window (e.g., 7 days for Kafka, 7 days for Event Hubs)
• Use multiple Kafka brokers and Event Hub partitions for high availability
• Monitor source connectivity from the streaming dashboard
• Set up credential rotation alerts before expiration

Testing Recovery Procedures

Do not wait for a production failure to discover your recovery process is broken. Test these procedures regularly in a staging environment.

Test Matrix

Running a Recovery Drill

# 1. Record current stream position
pre_test_progress = query.lastProgress
pre_test_batch = pre_test_progress.get("batchId", 0)
print(f"Pre-test batch: {pre_test_batch}")

# 2. Stop the stream
query.stop()
query.awaitTermination(timeout=60)

# 3. Introduce the failure (e.g., delete a commit file)
dbutils.fs.rm(f"{checkpoint_path}/commits/{pre_test_batch}")

# 4. Restart the stream
query = stream_df.writeStream.option("checkpointLocation", checkpoint_path).start()

# 5. Verify recovery
import time
time.sleep(60)
post_test_progress = query.lastProgress
post_test_batch = post_test_progress.get("batchId", 0)
print(f"Post-test batch: {post_test_batch}")
assert post_test_batch >= pre_test_batch, "Stream did not recover"
print("Recovery test PASSED")

Recommended Schedule

• Monthly: Checkpoint corruption and full reset tests
• Quarterly: OOM simulation and schema evolution tests
• After infrastructure changes: Source connectivity and credential rotation tests
• Before go-live: Full test matrix execution

Quick Reference: Recovery Decision Tree

Stream failed?
├── Check query.exception()
│   ├── Checkpoint error? → See Scenario 1
│   ├── Schema/Analysis error? → See Scenario 2
│   ├── OOM / memory error? → See Scenario 3
│   └── Connection / timeout error? → See Scenario 4
│
├── Stream running but not progressing?
│   ├── query.status shows waiting for data → Source may be empty (normal)
│   ├── query.status shows processing → Batch is slow (see performance tuning guide)
│   └── No lastProgress updates → Stream is stuck; restart the query
│
└── Stream running but producing wrong results?
    ├── Duplicates appearing → Check dedup watermark and checkpoint health
    ├── Missing data → Check source retention and failOnDataLoss setting
    └── Wrong aggregations → State store may be corrupt; consider state reset

*Real-Time Streaming Toolkit — Datanest Digital — https://datanest.dev*

Streaming Performance Tuning Guide

Datanest Digital — https://datanest.dev

Overview

Streaming performance tuning is an iterative process. Small configuration changes can have outsized impact on throughput, latency, and cost. This guide covers the primary levers available in Databricks Structured Streaming and Delta Live Tables, with concrete recommendations for each.

The goal is a pipeline that processes data within your latency SLA while using the fewest resources possible.

Trigger Intervals

The trigger interval controls how frequently Spark initiates a new micro-batch.

Configuration

# Fixed interval — most common for production
query = stream_df.writeStream.trigger(processingTime="10 seconds").start()

# Available once — process all available data, then stop (batch-like)
query = stream_df.writeStream.trigger(availableNow=True).start()

# Continuous processing (experimental) — sub-millisecond latency
query = stream_df.writeStream.trigger(continuous="1 second").start()

Choosing the Right Interval

Guidelines

• Do not set the trigger shorter than your average micro-batch processing time. If a batch takes 8 seconds, a 5-second trigger means batches queue up with no benefit.
• Start at 10 seconds and adjust based on observed batch duration and latency requirements.
• Use availableNow for backfill operations instead of lowering the trigger interval — it processes all available data in a single run without the overhead of repeated scheduling.

Partition Sizing

Partitioning determines parallelism and directly affects throughput.

Source Partitions

For Kafka and Event Hubs, the number of source partitions sets the upper bound on read parallelism.

# Override minimum read partitions (useful when source has few partitions)
stream_df = (
    spark.readStream
    .format("kafka")
    .option("minPartitions", "16")  # Read parallelism independent of topic partitions
    .load()
)

Sink Partitions

Control the number of output files per micro-batch:

# Repartition before writing to control file count
stream_df = stream_df.repartition(8)

# Or use Delta's optimizeWrite to handle it automatically
spark.conf.set("spark.databricks.delta.optimizeWrite.enabled", "true")

Recommendation: Enable optimizeWrite and let Delta handle output file sizing. Manual repartitioning is only needed for specific partitioning schemes (e.g., date-based).

Shuffle Partitions

The spark.sql.shuffle.partitions setting controls parallelism for operations that require data shuffling (joins, aggregations, deduplication).

Configuration

# Default is 200 — often too high for streaming, too low for large batch
spark.conf.set("spark.sql.shuffle.partitions", "auto")  # DBR 14.3+

# Or set explicitly
spark.conf.set("spark.sql.shuffle.partitions", "64")

Sizing Guidelines

Auto Shuffle Partitions

On Databricks Runtime 14.3+, use adaptive shuffle:

spark.sql.shuffle.partitions=auto
spark.sql.adaptive.enabled=true
spark.sql.adaptive.coalescePartitions.enabled=true

This automatically adjusts partition count based on data volume per micro-batch, avoiding the overhead of too many empty partitions during low-traffic periods.

State Store Management

Stateful operations (windowed aggregations, deduplication, stream-stream joins) maintain state across micro-batches. State store size is the most common cause of streaming OOM failures.

Use RocksDB State Store

The default in-memory state store does not scale. Switch to RocksDB:

spark.sql.streaming.stateStore.providerClass=com.databricks.sql.streaming.state.RocksDBStateStoreProvider

RocksDB Tuning

# Block cache — increase for large state (default 64MB)
spark.sql.streaming.stateStore.rocksdb.blockCacheSizeMB=256

# Write buffer — increase for high write throughput
spark.sql.streaming.stateStore.rocksdb.writeBufferSizeMB=64

# Compaction style
spark.sql.streaming.stateStore.rocksdb.compactOnCommit=false

Watermark Configuration

Watermarks control when state is evicted. A tighter watermark means less state but more late-arriving events are dropped.

# 1 hour watermark — state for each key lives at most 1 hour past max event time
stream_df = stream_df.withWatermark("event_timestamp", "1 hour")

Monitoring State Size

Track state store metrics via the streaming query progress:

progress = query.lastProgress
state_info = progress.get("stateOperators", [])
for op in state_info:
    print(f"  Num rows total: {op.get('numRowsTotal', 0)}")
    print(f"  Memory used:    {op.get('memoryUsedBytes', 0) / 1024 / 1024:.1f} MB")
    print(f"  Rows updated:   {op.get('numRowsUpdated', 0)}")
    print(f"  Rows removed:   {op.get('numRowsRemovedByWatermark', 0)}")

If numRowsTotal grows continuously without corresponding removals, your watermark is too wide or state is not being evicted.

Batch Size Tuning

Control how much data each micro-batch processes.

Kafka

# Maximum offsets (messages) per trigger across all partitions
.option("maxOffsetsPerTrigger", "100000")

# Fetch size per partition
.option("kafka.max.partition.fetch.bytes", "1048576")  # 1 MB

Auto Loader

# Maximum files per trigger
.option("cloudFiles.maxFilesPerTrigger", "1000")

# Maximum bytes per trigger
.option("cloudFiles.maxBytesPerTrigger", "10g")

Event Hubs

# Maximum events per trigger
.option("maxEventsPerTrigger", "50000")

Sizing Strategy

1. Start conservative: Set maxOffsetsPerTrigger to a value that completes within 50% of your trigger interval.

2. Measure batch duration: Check query.lastProgress["durationMs"] for the processing time breakdown.

3. Increase gradually: Double the batch size and verify that batch duration stays within the trigger interval.

4. Watch for OOM: If executor memory usage spikes, reduce batch size or increase executor memory.

Micro-Batch Optimization

Processing Time Breakdown

Each micro-batch has measurable phases:

progress = query.lastProgress
durations = progress.get("durationMs", {})
# Typical output:
# {
#   "addBatch": 4500,        <-- actual data processing
#   "commitOffsets": 50,
#   "getBatch": 200,
#   "latestOffset": 100,
#   "queryPlanning": 150,
#   "triggerExecution": 5100,
#   "walCommit": 100
# }

Delta Write Optimization

# Combine small files automatically
spark.databricks.delta.optimizeWrite.enabled=true

# Auto-compact after writes
spark.databricks.delta.autoCompact.enabled=true
spark.databricks.delta.autoCompact.minNumFiles=50

# Target file size (default 128MB — reduce for streaming)
spark.databricks.delta.targetFileSize=64mb

Predicate Pushdown

Ensure filters are pushed to the source to reduce data volume:

# Good — filter is pushed to Kafka (topic-level)
stream_df = spark.readStream.format("kafka").option("subscribe", "events.clicks").load()

# Bad — reading all topics and filtering in Spark
stream_df = spark.readStream.format("kafka").option("subscribePattern", "events.*").load()
stream_df = stream_df.filter(col("topic") == "events.clicks")

Monitoring Key Metrics

Essential Metrics to Track

Streaming Metrics via Spark Listener

Enable metrics reporting for external monitoring systems:

spark.sql.streaming.metricsEnabled=true

This exposes metrics to the Spark metrics system, which can forward to Prometheus, Datadog, or other collectors configured on the cluster.

Dashboard Query for Batch Duration Trend

SELECT
    batch_id,
    trigger_execution_ms,
    add_batch_ms,
    num_input_rows,
    ROUND(num_input_rows / (add_batch_ms / 1000.0), 0) AS rows_per_second
FROM streaming_metrics_log
WHERE stream_name = '${stream_name}'
ORDER BY batch_id DESC
LIMIT 100;

Tuning Checklist

Use this checklist when deploying a new streaming pipeline or investigating performance issues:

• [ ] Trigger interval >= average batch duration
• [ ] Shuffle partitions set to 2-4x total cluster cores
• [ ] RocksDB state store enabled for stateful operations
• [ ] Watermark configured with appropriate delay for your use case
• [ ] maxOffsetsPerTrigger or equivalent batch size limit set
• [ ] optimizeWrite and autoCompact enabled for Delta sinks
• [ ] Streaming metrics enabled (metricsEnabled=true)
• [ ] Checkpoint on durable cloud storage (not local DBFS)
• [ ] Monitoring dashboard deployed with alerting on key metrics
• [ ] Tested at 2x expected peak throughput

*Real-Time Streaming Toolkit — Datanest Digital — https://datanest.dev*