Spark Structured Streaming: Triggers, Watermarks, and State Management

Spark Structured Streaming: Real-Time Processing at Scale

Spark Structured Streaming is a scalable, fault-tolerant stream processing engine built on the Spark SQL engine.

Why Structured Streaming?

Key Abstraction:

• Live data stream as a continuously appended unbounded table
• Express stream processing as DataFrame/Dataset operations
• Same code for batch and streaming

Advantages over Spark Streaming (DStream):

1. DataFrame abstraction — Catalyst optimizer enables SQL-level optimizations
2. Incremental processing — engine handles state management
3. Fault tolerance — built-in recovery mechanisms
4. Unified model — same code for batch and streaming

Key Insight: You write the same code for both batch and streaming — the engine handles incremental processing, fault tolerance, and state management.

Structured Streaming Micro-Batch Architecture

Watermark Diagram

Architecture Diagram

Key Concepts

Production Code

Kafka-to-Sink Streaming Pipeline with Watermarking

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

logger = logging.getLogger(__name__)


def create_streaming_session() -> SparkSession:
    """Create SparkSession optimized for Structured Streaming."""
    return (
        SparkSession.builder
        .appName("KafkaStreamingPipeline")
        .config("spark.sql.streaming.schemaInference", "true")
        .config("spark.sql.streaming.forceDeleteTempCheckpointLocation", "true")
        .config("spark.sql.shuffle.partitions", "8")
        .config("spark.streaming.stopGracefullyOnShutdown", "true")
        .config("spark.sql.streaming.metricsEnabled", "true")
        .getOrCreate()
    )


def build_streaming_pipeline(spark: SparkSession) -> None:
    """Build a production streaming pipeline with watermarking and aggregation."""

    # Define event schema
    event_schema = StructType([
        StructField("event_id", StringType(), False),
        StructField("user_id", StringType(), False),
        StructField("event_type", StringType(), False),
        StructField("amount", DoubleType(), True),
        StructField("event_time", TimestampType(), False),
    ])

    # Read from Kafka
    # Parameters explained:
    #   format("kafka") - Use Kafka as the streaming source
    #   kafka.bootstrap.servers - Comma-separated list of Kafka broker addresses
    #   subscribe - Kafka topic(s) to consume from (comma-separated for multiple)
    #   startingOffsets - "latest" (new data only) or "earliest" (replay all)
    #   kafka.security.protocol - Security protocol for authentication
    #   maxOffsetsPerTrigger - Cap on records per micro-batch (prevents OOM)
    raw_stream = (
        spark.readStream
        .format("kafka")
        .option("kafka.bootstrap.servers", "kafka-broker-1:9092,kafka-broker-2:9092")
        .option("subscribe", "user-events")
        .option("startingOffsets", "latest")
        .option("kafka.security.protocol", "SASL_SSL")
        .option("maxOffsetsPerTrigger", 100000)
        .load()
    )

    # Parse JSON payload
    parsed_stream = (
        raw_stream
        .select(
            F.col("key").cast("string").alias("kafka_key"),
            F.from_json(F.col("value").cast("string"), event_schema).alias("data"),
            F.col("timestamp").alias("kafka_timestamp"),
        )
        .select("kafka_key", "data.*")
        .filter(F.col("event_id").isNotNull())
        .filter(F.col("amount").isNotNull())
    )

    # Add watermark for late data tolerance (10 minutes)
    # Watermark defines how late an event can arrive and still be included
    # Events arriving > 10 minutes late will be dropped
    watermarked_stream = parsed_stream.withWatermark("event_time", "10 minutes")

    # Windowed aggregation: 5-minute tumbling windows
    # Parameters:
    #   F.window("event_time", "5 minutes") - Tumbling window of 5 minutes
    #   "event_type" - Group by event type within each window
    #   .agg() - Aggregation functions applied per window per group
    windowed_aggregation = (
        watermarked_stream
        .groupBy(
            F.window("event_time", "5 minutes"),
            "event_type",
        )
        .agg(
            F.count("*").alias("event_count"),
            F.sum("amount").alias("total_amount"),
            F.avg("amount").alias("avg_amount"),
            F.countDistinct("user_id").alias("unique_users"),
        )
        .select(
            F.col("window.start").alias("window_start"),
            F.col("window.end").alias("window_end"),
            "event_type",
            "event_count",
            "total_amount",
            "avg_amount",
            "unique_users",
        )
    )

    # Write to Kafka sink with checkpointing
    query = (
        windowed_aggregation
        .writeStream
        .format("kafka")
        .option("kafka.bootstrap.servers", "kafka-broker-1:9092")
        .option("topic", "event-aggregations")
        .option("checkpointLocation", "s3://checkpoints/event-aggregation")
        .option("failOnDataLoss", "false")
        .trigger(processingTime="30 seconds")
        .outputMode("update")
        .start()
    )

    logger.info(f"Streaming query started: {query.id}")

    # Monitor the streaming query
    import time
    while query.isActive:
        progress = query.lastProgress
        if progress:
            logger.info(
                f"Batch {progress['batchId']}: "
                f"input_rows={progress['numInputRows']}, "
                f"processed_rows={progress['processedRowsPerSecond']:.0f}/s, "
                f"state_rows={progress.get('stateOperator', {}).get('numRowsTotal', 0)}"
            )
        time.sleep(30)

    query.awaitTermination()

Stream-Stream Join with Session Windows

def build_stream_stream_join(spark: SparkSession) -> None:
    """Join two streaming DataFrames with session windows."""
    from pyspark.sql.types import StructType, StructField, StringType, TimestampType

    # Click stream schema
    click_schema = StructType([
        StructField("click_id", StringType(), False),
        StructField("user_id", StringType(), False),
        StructField("page_url", StringType(), False),
        StructField("click_time", TimestampType(), False),
    ])

    # Purchase stream schema
    purchase_schema = StructType([
        StructField("purchase_id", StringType(), False),
        StructField("user_id", StringType(), False),
        StructField("product_id", StringType(), False),
        StructField("amount", StringType(), False),
        StructField("purchase_time", TimestampType(), False),
    ])

    # Read click stream
    click_stream = (
        spark.readStream
        .format("kafka")
        .option("kafka.bootstrap.servers", "kafka-broker-1:9092")
        .option("subscribe", "click-events")
        .load()
        .select(F.from_json(F.col("value").cast("string"), click_schema).alias("data"))
        .select("data.*")
        .withWatermark("click_time", "30 minutes")
    )

    # Read purchase stream
    purchase_stream = (
        spark.readStream
        .format("kafka")
        .option("kafka.bootstrap.servers", "kafka-broker-1:9092")
        .option("subscribe", "purchase-events")
        .load()
        .select(F.from_json(F.col("value").cast("string"), purchase_schema).alias("data"))
        .select("data.*")
        .withWatermark("purchase_time", "30 minutes")
    )

    # Stream-stream join: clicks within 30 minutes before purchase
    joined_stream = (
        click_stream
        .join(
            purchase_stream,
            (click_stream.user_id == purchase_stream.user_id) &
            (click_stream.click_time <= purchase_stream.purchase_time) &
            (click_stream.click_time >= F.expr("purchase_time - INTERVAL 30 MINUTES")),
            "leftOuter",
        )
        .select(
            click_stream.click_id,
            click_stream.user_id,
            click_stream.page_url,
            purchase_stream.purchase_id,
            purchase_stream.product_id,
            purchase_stream.amount,
            click_stream.click_time,
            purchase_stream.purchase_time,
        )
    )

    # Write to console for testing
    query = (
        joined_stream
        .writeStream
        .format("console")
        .option("truncate", "false")
        .option("checkpointLocation", "s3://checkpoints/click-purchase-join")
        .trigger(processingTime="10 seconds")
        .outputMode("append")
        .start()
    )

    query.awaitTermination()

Best Practices

1. Always set checkpointLocation for production streaming queries. This enables fault tolerance and exactly-once semantics.
2. Configure watermarks for any stateful operation to prevent unbounded state growth. Set the delay threshold based on observed event lateness.
3. Use maxOffsetsPerTrigger to cap the amount of data processed per micro-batch, preventing OOM errors during data spikes.
4. Enable streaming metrics (spark.sql.streaming.metricsEnabled=true) for monitoring input rates, processing rates, and state sizes.
5. Use failOnDataLoss=false for production workloads where occasional data loss is acceptable. Investigate data loss alerts separately.
6. Prefer Update mode for aggregations where only changed rows matter. Complete mode writes the full result every trigger, which is expensive.
7. Set spark.sql.shuffle.partitions to a small number (4-8) for streaming aggregations to reduce task overhead.
8. Monitor state store size via Spark UI state operator metrics. Alert on unexpected state growth.
9. Use ForeachBatch for complex sink logic that cannot be expressed with built-in formats.
10. Test with spark.sql.streaming.testing=true to enable deterministic testing of streaming queries.

Streaming Trigger Comparison

Output Mode Compatibility

See Also

• 021 - Apache Spark: RDDs, DataFrames, and the Catalyst Optimizer - Spark batch fundamentals
• 023 - Batch vs Streaming: Lambda, Kappa, and Architecture Trade-offs - Streaming architecture patterns
• 019 - Apache Kafka: Topics, Producers, and Consumers - Kafka as streaming source
• 020 - Kafka Streams: DSL, Windowed Aggregation, and Exactly-Once - Alternative streaming approach
• 030 - Capstone Project: Real-Time Streaming Pipeline - Production streaming pipeline project