3.4.3. Spark Structured Streaming with Delta Tables

💡 First Principle: Spark Structured Streaming treats streams as unbounded tables, using the same DataFrame API as batch processing. Combined with Delta Lake, it provides exactly-once processing and queryable historical data.

Scenario: A retail platform ingests clickstream data continuously. The data must be immediately queryable for real-time dashboards while also available for historical analysis.

Streaming to Delta Pattern

# Read stream from Event Hub
stream_df = spark \
    .readStream \
    .format("eventhubs") \
    .options(**event_hub_config) \
    .load()

# Parse and transform
from pyspark.sql.functions import from_json, col
schema = StructType([
    StructField("deviceId", StringType()),
    StructField("temperature", DoubleType()),
    StructField("timestamp", TimestampType())
])

parsed_df = stream_df \
    .select(from_json(col("body").cast("string"), schema).alias("data")) \
    .select("data.*")

# Write to Delta table
query = parsed_df \
    .writeStream \
    .format("delta") \
    .outputMode("append") \
    .option("checkpointLocation", "/checkpoints/temperature") \
    .start("Tables/temperature_readings")

Key streaming settings:

• Trigger: how often the query processes available data: .trigger(processingTime="5 minutes") for fixed-interval micro-batches, .trigger(availableNow=True) to process everything available and stop, or the default (next micro-batch as soon as the last one finishes). It is not a pipeline trigger, and it does not handle late data (that is the watermark, 3.4.5).
• Checkpoint: checkpointLocation stores the source offsets and state-store data (metadata, not a copy of the output) so a restarted query resumes from the last committed point.
• Exactly-once vs. at-least-once: at-least-once means a failure and retry can write the same events to the sink twice. Exactly-once needs a replayable source and a checkpoint plus an idempotent or transactional sink. Delta Lake's transaction log makes it one; a plain external endpoint usually only gets at-least-once.

⚠️ Exam Trap: Not specifying checkpoint location means a restart reprocesses all data from the beginning. Checkpoints track stream progress—always configure them in production streaming jobs.

Output modes

Watermarks: without one, a stateful aggregation keeps every group's state forever and eventually runs out of memory. withWatermark bounds that state: Spark guarantees to keep data within the threshold, and later data may be dropped. How a watermark finalizes windows and handles late events is covered in 3.4.5.

Custom writes with foreachBatch: writeStream.foreachBatch(func) passes each micro-batch to func(batch_df, batch_id) as an ordinary DataFrame. Use it for MERGE upserts (dedup on write), writing to several tables, or routing rows by event type. Each streaming query needs its own checkpoint path; never share one between queries.

Delta table as a source: spark.readStream.format("delta").load(path) reads the rows already in the table, then only rows appended by later commits; the checkpoint lets each run resume where the last one stopped. startingVersion/startingTimestamp apply only to a brand-new query and are ignored once a checkpoint exists. By default the stream fails if upstream rows are updated or deleted; skipChangeCommits lets it continue (it supersedes ignoreChanges, which can re-emit unchanged rows as duplicates).