← All posts

Databricks Spark Structured Streaming: Building Real-Time Pipelines That Actually Work in Production

6 min read4.7k
#Databricks#Spark#DataEngineering#Streaming#DeltaLake#Kafka#Python

Batch pipelines are predictable. They run, they finish, you check the results. Streaming pipelines are alive — they never stop, failures compound, and a small misconfiguration at 2am can quietly corrupt hours of data before anyone notices.

Apache Spark Structured Streaming on Databricks makes real-time pipelines significantly more manageable than raw Kafka consumer loops or custom Flink jobs. In this post I'll walk through building a production-grade streaming pipeline on Databricks — from Kafka ingest through stateful aggregations to Delta Lake sink — including the parts most tutorials skip: watermarking, late data handling, and checkpoint recovery.

Architecture Overview

Streaming Pipeline Flow

Key Concepts Before You Write Code

ConceptWhat it doesWhy it matters
TriggerControls micro-batch frequencyLatency vs cost tradeoff
WatermarkDefines how late data can arrivePrevents state from growing unbounded
CheckpointPersists stream state and offsetsExactly-once recovery after failure
Output modeAppend / Update / CompleteControls what gets written each batch
Stateful aggregationGroup-by with window functionsEnables time-windowed metrics
foreachBatchCustom sink logic per micro-batchEnables Delta MERGE in streaming

Step 1 — Read from Kafka / Azure Event Hub

Azure Event Hub exposes a Kafka-compatible endpoint so the same Spark code works for both.

python
from pyspark.sql import SparkSession
from pyspark.sql.functions import col, from_json, to_timestamp
from pyspark.sql.types import StructType, StructField, StringType, DoubleType, LongType

spark = SparkSession.builder.getOrCreate()

# Event schema — define this tightly, don't infer from JSON
EVENT_SCHEMA = StructType([
    StructField("event_id",    StringType(),  False),
    StructField("user_id",     StringType(),  False),
    StructField("event_type",  StringType(),  True),
    StructField("product_id",  StringType(),  True),
    StructField("amount",      DoubleType(),  True),
    StructField("event_ts",    StringType(),  True),  # parse as string first
    StructField("country",     StringType(),  True),
])

# Read from Kafka / Event Hub
raw_stream = (
    spark.readStream
        .format("kafka")
        .option("kafka.bootstrap.servers",    "your-eventhub.servicebus.windows.net:9093")
        .option("kafka.security.protocol",    "SASL_SSL")
        .option("kafka.sasl.mechanism",       "PLAIN")
        .option("kafka.sasl.jaas.config",     dbutils.secrets.get("kv-scope", "eh-sasl-config"))
        .option("subscribe",                  "user-events")
        .option("startingOffsets",            "latest")
        .option("maxOffsetsPerTrigger",       "50000")   # back-pressure control
        .load()
)

# Parse the value column (bytes → JSON → typed struct)
parsed_stream = (
    raw_stream
        .select(from_json(col("value").cast("string"), EVENT_SCHEMA).alias("data"))
        .select("data.*")
        .withColumn("event_ts", to_timestamp(col("event_ts")))
        .filter(col("event_id").isNotNull())
        .filter(col("user_id").isNotNull())
)

Step 2 — Write Raw Events to Bronze Delta

The Bronze sink is append-only. Use foreachBatch so you can add custom logic (deduplication, metadata) without losing streaming semantics.

python
import os

BRONZE_TABLE  = "orders.bronze.events"
CHECKPOINT_BRONZE = "abfss://checkpoints@yourstorage.dfs.core.windows.net/bronze/events"

def write_bronze(batch_df, batch_id):
    if batch_df.isEmpty():
        return

    (
        batch_df
            .withColumn("_batch_id",     lit(batch_id))
            .withColumn("_ingested_at",  current_timestamp())
            .write
            .format("delta")
            .mode("append")
            .option("mergeSchema", "true")
            .saveAsTable(BRONZE_TABLE)
    )
    print(f"Bronze batch {batch_id}: {batch_df.count()} rows written")

bronze_query = (
    parsed_stream
        .writeStream
        .foreachBatch(write_bronze)
        .option("checkpointLocation", CHECKPOINT_BRONZE)
        .trigger(processingTime="30 seconds")   # micro-batch every 30s
        .start()
)

Step 3 — Stateful Windowed Aggregations with Watermarking

This is where most streaming tutorials give up. Watermarking tells Spark how long to wait for late-arriving data before finalizing a window result. Without it, Spark holds state forever and eventually OOMs.

python
from pyspark.sql.functions import window, col, sum as _sum, count, avg

GOLD_TABLE       = "orders.gold.event_windows"
CHECKPOINT_GOLD  = "abfss://checkpoints@yourstorage.dfs.core.windows.net/gold/event_windows"

# Watermark: wait up to 10 minutes for late data
# Window: aggregate in 5-minute tumbling windows
windowed_agg = (
    parsed_stream
        .withWatermark("event_ts", "10 minutes")       # late data tolerance
        .groupBy(
            window(col("event_ts"), "5 minutes"),       # tumbling window
            col("country"),
            col("event_type"),
        )
        .agg(
            count("event_id")    .alias("event_count"),
            _sum("amount")       .alias("total_amount"),
            avg("amount")        .alias("avg_amount"),
        )
        .select(
            col("window.start").alias("window_start"),
            col("window.end")  .alias("window_end"),
            "country", "event_type",
            "event_count", "total_amount", "avg_amount",
        )
)

def write_gold(batch_df, batch_id):
    if batch_df.isEmpty():
        return

    from delta.tables import DeltaTable

    if DeltaTable.isDeltaTable(spark, f"spark_catalog.{GOLD_TABLE}"):
        gold_table = DeltaTable.forName(spark, GOLD_TABLE)
        # MERGE so re-processed windows update rather than duplicate
        (
            gold_table.alias("tgt")
                .merge(
                    batch_df.alias("src"),
                    """tgt.window_start = src.window_start
                       AND tgt.country    = src.country
                       AND tgt.event_type = src.event_type"""
                )
                .whenMatchedUpdateAll()
                .whenNotMatchedInsertAll()
                .execute()
        )
    else:
        batch_df.write.format("delta").saveAsTable(GOLD_TABLE)

gold_query = (
    windowed_agg
        .writeStream
        .foreachBatch(write_gold)
        .option("checkpointLocation", CHECKPOINT_GOLD)
        .outputMode("update")
        .trigger(processingTime="30 seconds")
        .start()
)

# Wait for both streams (in production, use separate jobs)
spark.streams.awaitAnyTermination()

Step 4 — Handling Stream Failures and Checkpoint Recovery

Checkpoints are what make Spark Structured Streaming exactly-once. Never delete them unless you want to reprocess from the beginning.

python
# Monitor active streams
for stream in spark.streams.active:
    print(f"Stream: {stream.name}")
    print(f"  Status:  {stream.status}")
    print(f"  Recent progress: {stream.recentProgress[-1] if stream.recentProgress else 'none'}")

# Graceful shutdown — let current batch finish before stopping
def graceful_shutdown():
    for stream in spark.streams.active:
        print(f"Stopping stream: {stream.name}")
        stream.stop()
    print("All streams stopped cleanly.")

# Check if checkpoint is healthy before restarting
def validate_checkpoint(checkpoint_path: str) -> bool:
    try:
        files = dbutils.fs.ls(checkpoint_path)
        has_offsets  = any(f.name == "offsets/"   for f in files)
        has_commits  = any(f.name == "commits/"   for f in files)
        has_metadata = any(f.name == "metadata"   for f in files)
        return has_offsets and has_commits and has_metadata
    except Exception:
        return False

# Run before starting a stream after an outage
if validate_checkpoint(CHECKPOINT_BRONZE):
    print("Checkpoint healthy — resuming from last committed offset")
else:
    print("WARNING: Checkpoint missing or corrupt — will restart from 'latest'")

Things to Watch in Production

Watermark too tight causes data loss; too loose causes OOM. Start with 2x your expected event latency. If events are typically 2 minutes late, set a 4-minute watermark. Monitor the eventTime.watermark metric in stream progress to tune it.

One checkpoint per stream, never share. If two streams share a checkpoint location, they'll corrupt each other's offset tracking. Always use separate ADLS paths per stream query.

maxOffsetsPerTrigger is your back-pressure valve. Without it, a traffic spike will cause a single micro-batch to process millions of records, blow your cluster memory, and fail. Set it conservatively and tune up from there.

Use Trigger.AvailableNow() for near-real-time batch. If you don't need sub-minute latency, Trigger.AvailableNow() processes all available data and stops — cheaper than a continuously running stream and easier to schedule via Databricks Workflows.

Wrapping Up

Spark Structured Streaming on Databricks gives you a solid foundation for real-time pipelines, but the production details — watermarking, checkpoint management, back-pressure, and Delta MERGE in foreachBatch — are what separate a demo from something you can actually trust at 3am. Get those right and the rest is just business logic.

References