Structured Streaming with Spark

Data Engineering & Modern Data Platforms · Lesson 32 · Big Data Processing and Apache Spark

Lesson Overview

The course-day table from Spark Performance Optimization is ready at 06:00 UTC. Product asked for a tile that shows activity in the last five minutes, for the same course_id grain, while a lesson is live. A batch job every morning cannot answer that. The events already land on a Kafka topic, learning.progress, which Module 4 will take apart as a system: brokers, partitions, consumer groups, and retention. This lesson treats the topic as a replayable source with offsets. It does not teach Kafka itself.

Someone's first Structured Streaming job grouped by course_id with no watermark, wrote with foreachBatch in append mode to a Parquet path, and stored the checkpoint on the driver's local disk. Two things happened. The state store kept every course key forever, because nothing told Spark that an old aggregate could no longer change. And when the driver restarted on another node, the checkpoint was gone, so the query either failed or, worse, started again from startingOffsets and rewrote a day of tiles. A colleague pointed at the Kafka consumer-group offset and said the group would resume. Structured Streaming does not use that commit as its recovery cursor. The checkpoint does.

Structured Streaming runs the Spark SQL engine on an unbounded table. By default it executes that table as micro-batches: each trigger takes a range of source offsets, updates a result, and commits. Continuous processing, available since Spark 2.3, is a separate experimental mode with latencies quoted by the guide as low as about a millisecond and with at-least-once guarantees. It does not run foreachBatch. The five-minute tile is a micro-batch. State, watermarks, and the checkpoint are what make the micro-batch restartable.

The programming guide for Spark 3.5 states that end-to-end exactly-once behavior comes from checkpointing and write-ahead logs, replayable sources, and sinks that tolerate reprocessing. The file sink is documented as exactly-once. foreachBatch is documented as at-least-once unless you use the batch id to make the write idempotent. The Kafka sink is not the file sink. This lesson writes the tile with the file sink in append mode after a watermark, and it shows foreachBatch only with an explicit batch id. Module 4 is where the topic, the key, and the retention window get their own design.

Examples target Apache Spark 3.5. Trigger names and sink guarantees below are taken from that guide. Confirm them if you run Spark 4.

Learning Objectives

By the end of this lesson you will be able to:

1. Describe a Structured Streaming query as an unbounded table updated by triggers, executed by default as micro-batches.
2. Contrast micro-batch execution with continuous processing, including the guarantee and the foreachBatch limitation.
3. Choose append, update, or complete output mode for a windowed count, and say which modes can drop state.
4. Define a watermark as max observed event time minus a delay, and list the conditions under which it cleans aggregation state.
5. Place withWatermark on the timestamp column before the aggregation.
6. Configure checkpointLocation on durable storage, and explain what a restart does with and without that directory.
7. Explain why startingOffsets applies when there is no checkpoint, and why a Kafka group offset is not Spark's cursor.
8. Use the file sink for an exactly-once tile, and use batch_id inside foreachBatch when you must call a batch writer.
9. Bound a trigger with maxOffsetsPerTrigger or Trigger.ProcessingTime, and use Trigger.AvailableNow for a finite catch-up.
10. Hand Module 4 a source contract: topic name, event-time field, watermark delay, and checkpoint path, without pretending this lesson configured the cluster.

Prerequisites

• Spark Joins and Data Shuffling and Spark Performance Optimization: the course_id grain, file sizing, and why a shuffle still exists inside a streaming aggregate.
• Data Formats and Serialization: the Kafka payload here is JSON lines in the value bytes. The curated tile is Parquet.
• Data Quality Engineering: a late or unknown lesson still needs a rule. Streaming does not retire the gate. It changes when the gate runs.
• PySpark 3.5.x. A local lab can use the file source. The production shape uses the official kafka format from spark-sql-kafka, which is a Spark module, not a third-party wrapper this course invents.
• No Kafka cluster is required to learn watermarks. One is required before you claim the production query was tested. Module 4 builds that cluster understanding.

Detailed Core Concepts

The unbounded table

Structured Streaming names a query the way Spark SQL names a batch query. The input is a source that can produce more data. Each trigger appends a chunk of source rows to the input table and updates a result table. The sink receives whatever the output mode says to emit. You write groupBy and count. The engine decides how to update the count from the new chunk and from state it kept.

That model is why a batch function and a streaming function can share a transformation on a static DataFrame and a streaming one, up to the point where the operation has no incremental plan. explain on a streaming DataFrame still shows the incremental plan. foreachBatch is the escape hatch that hands you a normal DataFrame for one micro-batch, and it is also where people accidentally leave the guarantees.

Micro-batch and continuous processing

A micro-batch trigger plans a small Spark job over the offset range for that trigger. The engine records the offsets before the work is considered committed, and it records a commit when the sink succeeds. On failure it restarts from the last committed point and reprocesses an uncommitted range. The programming guide describes end-to-end latencies as low as about 100 milliseconds for this engine, which is a bound under ideal conditions, not a promise for a join-heavy tile. A one-minute Trigger.ProcessingTime is an honest choice for a five-minute window. Running as fast as possible, with no trigger interval, is how a small cluster spends its life scheduling batches of a few thousand rows.

Continuous processing uses Trigger.Continuous and a checkpoint interval. The guide marks it experimental. It aims at much lower latency and provides at-least-once guarantees, not the exactly-once story of the micro-batch engine. Operators and sinks are a subset. foreachBatch does not run there, because it depends on micro-batches. This course does not use continuous mode for the activity tile. If a product requirement is a few milliseconds and the aggregation is simple enough to fit the supported operators, read the continuous-processing section of the guide for your patch version and test failure behavior before you adopt it. A requirement for reliable window state and a Parquet sink stays on micro-batch.

Trigger.Once was the old "run one batch and stop" switch. Spark's guide now demonstrates Trigger.AvailableNow, which processes all data available at start, in one or more micro-batches, and then stops. Use AvailableNow for a catch-up or a test. Do not leave it as the production trigger of a tile that is supposed to stay up.

Output modes

Append emits rows that the result table has newly added and will not change. A windowed count in append mode is emitted once, after the watermark passes the window end, so the sink sees each window one time. That delay is the late-data allowance you configured. It is not a bug.

Update emits rows that changed in this trigger. A window can be rewritten many times as late events arrive, until the watermark allows state to drop. The sink must tolerate corrections. A Parquet file sink does not. The file sink supports append.

Complete emits the entire result table every trigger. It is for aggregates. It cannot drop old state, because the result still contains every group. A complete-mode count of all courses since the beginning of the topic will grow without bound. Do not use it for the tile.

The guide's support rules that matter here:

• A watermark cleans aggregation state only in append or update, and only when the aggregation is on the event-time column or a window over it.
• withWatermark must be applied on that same column, and it must be called before the aggregation.
• Joins in streaming queries are append-oriented. Do not plan on update or complete for a stream-stream join.

Watermarks and event time

Event time is the timestamp in the event, event_time, produced by the learning client. Processing time is when the micro-batch sees the row. Clients go offline. Partitions lag. A watermark is a threshold: Spark tracks the maximum event time seen and subtracts the delay you passed to withWatermark. Rows with event time older than the watermark are late. For a windowed aggregate in append mode, windows that end before the watermark can be emitted and their state dropped. Data later than that is not applied to those windows.

textCopy
watermark = max(event_time observed) - delay

A delay of ten minutes on a five-minute window means a window can close about ten minutes after the latest event time, not ten minutes after the wall clock, and only once some later event advances the max. If events stop, the watermark does not advance, and the last window stays open. That idle behavior is a reason to watch the watermark metric and the source lag together. It is also why a test must send a row past the window, not merely sleep.

The delay is a product decision. Ten minutes keeps a mobile client that reconnects on the train inside the tile and grows state for ten minutes of windows per key. A zero delay drops anything that arrives out of order, including ordinary jitter. Write the delay next to the tile's definition so a later "real-time" request does not silently change it.

Spark 3.5 propagates watermarks between stateful operators more carefully than a single global watermark did in older versions. If you chain two stateful steps, read the 3.5 migration notes and test late rows at the second step. This tile has one aggregation.

State

The aggregate keeps partial counts per (window, course_id) in a state store. The default provider is the HDFS-backed store: an in-memory map plus files under the checkpoint. Large state belongs on RocksDB:

textCopy
spark.sql.streaming.stateStore.providerClass
  org.apache.spark.sql.execution.streaming.state.RocksDBStateStoreProvider

That class name is Spark's, in spark-sql. Use it when the number of open windows times course keys no longer fits comfortably in executor memory. A five-minute window, a ten-minute watermark, and a few thousand courses is a small state. countDistinct on learner_key per window is not small, because the state holds keys, not a counter. The tile in this lesson counts events. Distinct learners are a separate decision with a separate state budget. If product accepts an approximation, approx_count_distinct changes the definition and the memory. Do not swap it in without changing the label on the tile.

Checkpoints

checkpointLocation is a directory unique to this query. Spark writes an offset log, a commit log, and state into it. The offset log is the write-ahead record of which source range a batch covers. The commit log records which batches completed. On restart, a batch that has offsets and no successful commit is reprocessed. The directory must live on durable storage shared by any node that might run the driver: HDFS or an object store the cluster already uses, not /tmp on one machine.

Rules that avoid painful recoveries:

• One query, one checkpoint path. A second query on the same path will treat the first query's offsets as its own.
• startingOffsets (earliest, latest, or a JSON map of partitions) is used when the checkpoint does not exist. When the checkpoint exists, Spark resumes from it and ignores a conflicting start.
• Deleting the checkpoint and starting from latest skips the gap. Deleting it and starting from earliest replays the topic, as far as retention allows, into the sink. Both are operational choices. Neither is a restart.
• Changing the aggregation in a way the state schema cannot read requires a new checkpoint and a planned cutover. Do not point a new definition at an old state directory and hope.

Kafka retention has to exceed the time you need to replay after a checkpoint, and it has to cover the watermark delay plus a failure window. Module 4 sets retention. This lesson states the requirement: if the checkpoint says offset 1000 and the broker no longer has offset 1000, the query cannot reconstruct the input. failOnDataLoss on the Kafka source, defaulting so that lost offsets fail the query rather than skip, is the setting you want for this tile. Skipping is a silent hole in a five-minute count.

What Spark stores, and what Kafka stores

The Kafka source integration reads partitions. It can be given kafka.bootstrap.servers, subscribe, startingOffsets, maxOffsetsPerTrigger, minPartitions, and failOnDataLoss. Those are source options in the Spark Kafka integration guide. Spark tracks progress in the checkpoint. Committing offsets to a Kafka consumer group is not how this source recovers. You may see a group id in broker metrics and still restart from the wrong place if the checkpoint moved. When Module 4 explains group offsets, keep the two cursors apart: the group is for consumers that commit to Kafka; this query's cursor is the checkpoint.

maxOffsetsPerTrigger caps how much of a backlog one micro-batch will take. After an outage, that cap stops the first batch from trying to shuffle a full day of events in one trigger. The tile lags and then catches up. That is preferable to a batch that spills for an hour and then fails the checkpoint timeout.

Sinks and the word "exactly-once"

The Spark 3.5 sink table documents the file sink as fault-tolerant and exactly-once, in append mode, provided the checkpoint is in use. The sink writes files for a batch in a way that a reprocessed batch does not duplicate committed output. That is the sink this tile uses.

foreachBatch runs a function (batch_df, batch_id) per micro-batch. The guide states that it provides at-least-once writes by default, and that using batch_id to deduplicate can produce an exactly-once effect. The function may run more than once for the same id if the commit did not finish. A function that appends to a directory without looking at batch_id double-counts after a retry. A function that overwrites a path keyed by batch_id, or that writes a row set whose key includes the window and replaces that window, can be made idempotent. You own that logic. Spark does not inspect it.

The Kafka sink writes to a topic. Treat it as a sink that can emit duplicates on retry unless you have designed idempotence on the consumer side. This lesson does not claim an exactly-once Kafka sink API. Downstream consumers of the tile should read the Parquet path, or, if you later publish to Kafka, Module 4's delivery lesson applies to that hop.

Architecture and Design Diagrams

mermaidCopy
flowchart LR
    subgraph source [Replayable source]
        kafka[Kafka topic learning.progress]
    end

    subgraph engine [Micro-batch engine]
        off[Offset range for this trigger]
        parse[Parse JSON value]
        wm[Watermark on event_time]
        agg[Window aggregate by course_id]
        state[(State store under checkpoint)]
        commit[Commit log]
    end

    subgraph sink [Append file sink]
        parquet[Parquet tile path partitioned by window date]
    end

    kafka --> off --> parse --> wm --> agg
    agg <--> state
    agg --> parquet
    off --> commit
    parquet --> commit

The checkpoint directory is the box around the offset log, the commit log, and the state. The driver can move. That directory cannot be local to the old driver.

A single window's life:

mermaidCopy
sequenceDiagram
    participant Src as Source offsets
    participant Q as Micro-batch
    participant St as State
    participant Sink as File sink
    Src->>Q: Batch N offset range logged
    Q->>St: Update counts for open windows
    Note over Q,St: Window ends plus delay still in the future
    Q->>Sink: Append mode emits nothing for that window yet
    Q->>Src: Commit batch N
    Src->>Q: Later batch advances the watermark
    Q->>Sink: Append the closed window once
    Q->>St: Drop state for the closed window

If the driver dies after the sink writes and before the commit, batch N runs again. The file sink's exactly-once contract is what keeps the closed window from appearing twice. A hand-rolled append inside foreachBatch does not inherit that contract.

Technical Explanation

Time columns

Store event_time as a timestamp in UTC inside the JSON, and parse it with a declared schema. A string timestamp that is not parsed will not window correctly. ingest_time can ride along as a processing-time diagnostic. The watermark uses event_time only. Using ingest time as if it were event time makes a buffered mobile client look "on time" and makes a truly late event look punctual. The tile's definition says event time.

window(event_time, "5 minutes") produces a struct with start and end. Those fields belong in the output. A sink partitioned only by calendar date derived from window.start keeps listings reasonable. Do not partition by course_id.

Stream-batch joins, briefly

A tile that must drop events for unknown lessons can left-join a static manifest DataFrame. Static joins in Structured Streaming are a documented pattern: the stream is the events, the batch is the dimension. The dimension is reread according to the source's options when you want it refreshed; a simple static read is a snapshot at query start unless you rebuild the query. For a manifest that changes hourly, a join to a snapshot is acceptable if the tile labels the manifest_id it used, and a restart picks up a new snapshot. Do not shuffle-join the stream to a large enrollment fact inside this query until you have sized the state. Lesson 30's salt does not automatically apply inside a streaming join. A hot course_id still creates a hot task in the aggregate. maxOffsetsPerTrigger limits how large one batch gets. It does not split the key. If one course dominates a trigger, the stage metrics from lesson 31 still apply.

Checkpoint layout and operations

Operators should know the directory contains the query's metadata, offsets, commits, and state. They should not hand-edit those files. Safe operations are: stop the query, keep the directory, start the same query again; or stop the query, archive the directory, and start from a chosen startingOffsets as a new query with a new path. Unsafe operations are: copy a checkpoint between two different queries, delete offsets and leave state, or point a changed aggregation at the old state.

Query ids and checkpoint paths should include the environment and the query name, for example s3a://tlp-checkpoints/prod/course-activity-tile/. A developer laptop using the production checkpoint will commit production offsets. Separate them.

Triggers you will set

ProcessingTime is about the trigger clock, not the event-time window. You can trigger every minute and still window by event time. Those clocks are supposed to differ.

Observability

Watch: input rows per batch, batch duration, watermark value, state rows, shuffle write of the aggregate, and the age of the last committed offset versus the end of the topic. The last of these is source lag. A watermark that stops moving while lag is zero means no new event time is arriving. A watermark that stops while lag grows means the query is stuck or under-triggered. Spark's streaming UI shows batch lists and state operators. You do not need a custom metrics product to see the first failure. You do need the checkpoint on durable storage so the UI's "last batch" survives a driver move.

Practical Implementation or Design Walkthrough

Problem

Build a five-minute course activity tile. Count events per course_id and window. Allow ten minutes of lateness. Emit each window once, in append mode, to Parquet. Checkpoint on a durable path. Provide a file-source variant a laptop can run, and a Kafka-source variant that matches production options.

Event schema

JSON object in the Kafka value, or one JSON file per row in the file-source lab:

The tile grain is (course_id, window_start, window_end).

File-source lab

pythonCopy
"""Five-minute course activity tile. File source for a local lab.

PySpark 3.5.x. The production block below swaps in the Kafka source.
"""

from pyspark.sql import SparkSession
from pyspark.sql import functions as F
from pyspark.sql.types import StructType, StructField, StringType, TimestampType
from pyspark.sql.streaming import Trigger

EVENT_SCHEMA = StructType(
    [
        StructField("event_id", StringType(), nullable=False),
        StructField("course_id", StringType(), nullable=False),
        StructField("lesson_id", StringType(), nullable=True),
        StructField("event_time", TimestampType(), nullable=False),
    ]
)


def build_spark() -> SparkSession:
    return (
        SparkSession.builder.appName("tlp-course-activity-tile")
        .master("local[*]")
        .config("spark.sql.shuffle.partitions", "4")
        .config("spark.sql.adaptive.enabled", "true")
        .getOrCreate()
    )


def tile(events):
    # Watermark before the group. Append mode emits a window once it is closed.
    return (
        events.withWatermark("event_time", "10 minutes")
        .groupBy(F.window("event_time", "5 minutes"), F.col("course_id"))
        .agg(F.count("*").alias("event_count"))
        .select(
            F.col("course_id"),
            F.col("window.start").alias("window_start"),
            F.col("window.end").alias("window_end"),
            F.col("event_count"),
        )
    )


def main() -> None:
    spark = build_spark()
    # maxFilesPerTrigger keeps a lab from swallowing the whole directory in one batch.
    raw = (
        spark.readStream.schema(EVENT_SCHEMA)
        .option("maxFilesPerTrigger", 1)
        .json("/tmp/tlp/stream/in")
    )
    query = (
        tile(raw)
        .writeStream.outputMode("append")
        .format("parquet")
        .option("path", "/tmp/tlp/stream/tile")
        .option("checkpointLocation", "/tmp/tlp/stream/chk-course-tile")
        .trigger(Trigger.AvailableNow())
        .start()
    )
    query.awaitTermination()


if __name__ == "__main__":
    main()

/tmp checkpoints are for the lab. The production query changes two options and the trigger:

pythonCopy
def kafka_events(spark, brokers: str):
    raw = (
        spark.readStream.format("kafka")
        .option("kafka.bootstrap.servers", brokers)
        .option("subscribe", "learning.progress")
        .option("startingOffsets", "earliest")
        .option("failOnDataLoss", "true")
        .option("maxOffsetsPerTrigger", "500000")
        .load()
    )
    # value is binary. startingOffsets is ignored once the checkpoint exists.
    parsed = raw.select(
        F.from_json(F.col("value").cast("string"), EVENT_SCHEMA).alias("e")
    ).select("e.*")
    return parsed


def start_production(spark, brokers: str, warehouse: str, checkpoint: str):
    return (
        tile(kafka_events(spark, brokers))
        .writeStream.outputMode("append")
        .format("parquet")
        .option("path", f"{warehouse}/course_activity_tile")
        .option("checkpointLocation", checkpoint)
        .trigger(Trigger.ProcessingTime("1 minute"))
        .start()
    )

The checkpoint argument must be a durable directory unique to course-activity-tile. startingOffsets of earliest matters on the first start only.

An idempotent foreachBatch, if you leave the file sink

Use this only when a batch writer is required. The file sink does not need it.

pythonCopy
def write_batch(batch_df, batch_id: int) -> None:
    # One directory per batch id. A retry replaces that directory instead of appending.
    path = f"/tmp/tlp/stream/batches/batch_id={batch_id}"
    (
        batch_df.write.mode("overwrite")
        .parquet(path)
    )


# query = tile(raw).writeStream.foreachBatch(write_batch)
#     .option("checkpointLocation", checkpoint).start()

A retry of batch_id 7 overwrites batch 7's path. It does not append a second copy. That is the idempotence the guide points at. Clean up abandoned batch directories with the same id scheme. Do not append into one growing folder inside this function.

Tests you can run without a cluster

Write three JSON files into the input directory: two events inside the same five-minute window, one event thirty minutes later. Run AvailableNow. The first window should appear, because the later event advances event time past the watermark delay. Then run again with no new files and the same checkpoint. The first window must not appear a second time in the sink. Delete the checkpoint, run again, and record that the window is produced again from the source. That third run is the lesson about what a deleted checkpoint does. It is not a production procedure.

Enterprise Real-World Scenarios

Scenario 1: The tile that double-counted after a driver failure

The first production tile used foreachBatch to append Parquet into tile/event_date=.... The checkpoint lived on the driver pod's emptyDir volume. A deploy moved the driver. The query started with startingOffsets latest because the new pod had no checkpoint, so the team thought they had only lost the gap. They had. A week later a node died after the append and before the commit, the checkpoint this time was on object storage, and the batch ran again. The append wrote a second copy of the same windows. The five-minute counts were high during the incident window and the daily course_day job, which does not read this tile, disagreed.

The repair is the file sink with a durable checkpoint, or foreachBatch that overwrites by batch_id or by window. The driver pod must not be the only place the checkpoint exists. startingOffsets stays earliest for a new query so a missing checkpoint is obvious and loud, and the deploy process refuses to start if the checkpoint path is empty in production without an incident ticket. Lag is monitored. The daily table remains the system of record for "how many events on this date." The tile is a live view with a watermark delay in its definition. They are not averaged together.

Scenario 2: State that never shrank

A second query computed a "courses seen today" set in complete mode so a dashboard could read one row per course with a running total. It worked in the demo. By the end of the month the state store held every course-day since launch, checkpoints took longer than the trigger interval, and batches queued. Nobody had set a watermark, and complete mode would not have dropped state anyway.

The repair splits the products. The live tile is append mode, a five-minute window, a ten-minute watermark, event counts only. The running total for a calendar day is the batch course_day job, or an update-mode stream with a watermark that emits corrections into a sink that can replace a key, if product truly needs a correction before morning. Complete mode stays off this query. RocksDB is enabled only after a measurement of state rows, not because a blog recommended it for every job. With a few thousand courses and two open windows, the default state store is enough.

Scenario 3: Handoff to the Kafka module

Platform engineering asks the streaming owners what they need from the topic before Module 4 freezes the contract. The answer from this query is concrete. The topic is learning.progress. The value is JSON with event_id, course_id, lesson_id, and event_time in UTC. Offsets must be replayable for longer than the checkpoint recovery window plus the ten-minute watermark. failOnDataLoss is true, so retention cannot be shorter than the replay need. Spark will not commit the consumer group as its cursor; group lag in Kafka is not this query's lag. The sink is Parquet under a named warehouse path. A later chapter can replace the transport. The watermark, the window, and the checkpoint ownership stay with this job.

Best Practices

• Use micro-batch for this tile. Adopt continuous mode only after the guide's operator limits and at-least-once behavior fit the product, and after a failure test.
• Put withWatermark on event_time before the window aggregate. Keep append mode with the file sink so each closed window is written once.
• Give every production query a durable, unique checkpointLocation. Refuse a production start that points at a local disk.
• Treat startingOffsets as a first-start policy. Document what you do if the checkpoint is lost: new path and a chosen offset, with a person approving the replay or the skip.
• Set failOnDataLoss to true and size source retention with Module 4 so the checkpoint's offsets still exist.
• Set maxOffsetsPerTrigger so a backlog cannot become one unbounded batch.
• Keep countDistinct out of the hot tile unless the state budget is signed. Count events, and let the batch job own distinct learners if that is the contract.
• Prefer the file sink's exactly-once behavior over a custom append. If you use foreachBatch, key the write by batch_id or by window and overwrite.
• Watch watermark, batch duration, state rows, and source lag. A stopped watermark with a growing lag is an incident. A stopped watermark with no new events is an idle source.
• Keep the daily course_day grain as the historical record. Do not redefine it as the sum of tiles after the fact without an explicit reconciliation.

Common Mistakes and Anti-Patterns

No watermark on a windowed count. State and, in update mode, the result keep every window. Append mode may refuse the query or be unable to emit finalized windows. Set a delay the product accepts.

Watermark after the aggregation, or on a different column. The guide states that state cleanup will not apply. Call withWatermark("event_time", ...) on the frame you aggregate, before groupBy.

Complete mode for a live tile. State cannot be dropped. The checkpoint grows for the life of the query. Use append with a window, or update with a sink that replaces keys, and keep a watermark.

Checkpoint on local disk. A new driver is a new empty checkpoint. You replay or skip, and you may not know which. Use shared durable storage.

Trusting the Kafka group offset as Spark's cursor. The checkpoint holds the range. A group commit, if one exists for display, can move independently. Operate the checkpoint.

foreachBatch that appends. Retries duplicate windows. Overwrite by batch_id or use the file sink.

startingOffsets latest as a safety default. A lost checkpoint silently drops history. Prefer a loud failure and a deliberate replay decision.

Partitioning the tile by course_id. You recreate the small-file problem inside a streaming sink. Partition by the window date if you must partition at all, and keep file counts in mind. Streaming file sinks can create many small files because each batch writes. Compact closed dates with the batch layout from lesson 31.

Using the tile as the only copy of the day's counts. A watermark drops late data from that window. The batch job, reading landed Parquet with a restatement window, is what finance and completion metrics should read. The tile is allowed to be incomplete relative to tomorrow's batch.

Continuous mode because the word "real-time" appeared in a ticket. The mode is experimental in the 3.5 guide, at-least-once, and it will not run this foreachBatch. Stay on micro-batch until the requirement is latency the micro-batch cannot meet and the operator set fits.

Troubleshooting and Design Review

Symptom: Windows never appear in the append sink. The query is running and input rows are non-zero.

Likely cause: the watermark has not advanced past the window end plus the delay. The test never sent a later event. Or withWatermark is missing and append cannot finalize.

Investigation: read the last watermark in the streaming UI. Compare it with window.end. Check operator order in the plan.

Corrective action: send or wait for an event that moves max event time. Confirm the watermark column is event_time. Do not switch to complete mode to "see rows" and leave it there.

Symptom: After a restart, counts jumped and the Parquet path contains duplicate windows.

Likely cause: foreachBatch appended, and a batch reran; or the checkpoint was lost and earliest replayed into an append path that is not idempotent.

Investigation: checkpoint directory presence, last committed batch id, and whether the sink path is the file sink or a custom append.

Corrective action: stop, decide the cursor, and rewrite the affected windows by overwrite. Move to the file sink or batch-id overwrite. Put the checkpoint on durable storage before the next start.

Symptom: The first batch after an outage runs for hours and then fails.

Likely cause: no maxOffsetsPerTrigger, so one micro-batch ingested the whole backlog. The shuffle and the checkpoint grew with it.

Investigation: input rows on that batch, compared with the cap you thought you set.

Corrective action: set a cap, restart from the existing checkpoint so you do not skip or double the committed prefix, and let the query catch up over several triggers. Watch lag.

Symptom: failOnDataLoss aborts the query. The broker no longer has the checkpoint's offset.

Likely cause: retention is shorter than downtime, or a topic was recreated.

Investigation: the offset in the checkpoint log and the earliest offset on the partition. This is a Module 4 retention fact meeting a Spark cursor.

Corrective action: do not flip failOnDataLoss to false as a habit. Choose a new starting offset with the product owner, use a new checkpoint directory, and accept the hole or the replay explicitly. Lengthen retention so the next outage fits.

Symptom: State rows climb for weeks. Checkpoint duration exceeds the trigger interval.

Likely cause: complete mode, a missing watermark, or a countDistinct on learners.

Investigation: state operator metrics and the output mode in the plan.

Corrective action: move to append and a bounded watermark, drop distinct from this query, and start a new checkpoint because the old state schema is the unbounded one. Compact or ignore the partial tile. The batch table covers history.

A design review asks: What is the window, the watermark delay, and the output mode? Where is the checkpoint, and who is allowed to delete it? What happens on a retry of the sink? What is startingOffsets for, and what is it not? How much of a backlog can one trigger take? Which table is the system of record the next morning? What must Module 4 guarantee about retention and replay? A review that says "it's streaming so it's real-time" has not answered these.

Interview Questions and Answers

Q1. What is the default execution model in Structured Streaming, and how does a trigger use offsets?

The default is micro-batch. Each trigger selects a range of source offsets, runs a Spark plan over that range, updates state, writes the sink according to the output mode, and records a commit in the checkpoint. The offset range is logged so an uncommitted trigger can be repeated. The query looks like a SQL query on an unbounded table. The execution is a sequence of batch jobs with durable progress. I set Trigger.ProcessingTime when I want a bound on how often those jobs start.

Q2. How does continuous processing differ, and why is it not the course-activity tile?

Trigger.Continuous is the experimental low-latency engine described in the Spark 3.5 guide. The guide associates it with about one-millisecond-class latency and with at-least-once guarantees, and it supports a narrower set of operations. foreachBatch does not run in that mode. The tile needs window state, a watermark, and a Parquet file sink in append mode, which is the micro-batch design. I would revisit continuous mode only for a simpler query whose latency the micro-batch cannot meet, and only after reading the operator limits for the patch version we run.

Q3. Why does append mode delay a windowed count?

Append may emit a row only when that row will not change. A windowed count can still change until late data is no longer accepted. The watermark is max event time minus the delay. Once the watermark passes the window end, the count is final, Spark emits it once, and it can drop that window's state. The delay you see in the sink is the lateness you asked to keep. Update mode would emit intermediate counts and would need a sink that overwrites. The file sink is append.

Q4. What has to be true for a watermark to clean aggregation state?

The output mode is append or update, not complete. The aggregate is on the event-time column or a window of it. withWatermark names that same column and is called before the aggregation. Complete mode keeps the full result, so it cannot drop state. A watermark placed after groupBy, or on a different column, does not give the planner the information the guide requires. I then test with a late row and a row that advances event time, rather than trusting the configuration string.

Q5. What does the checkpoint directory represent, and what does startingOffsets do after it exists?

The checkpoint holds the query's offset log, commit log, and state. It is the cursor. startingOffsets on the Kafka source selects the start when no checkpoint is present. After the directory has commits, restart continues from those commits, even if the option says something else. Deleting the directory makes startingOffsets matter again, which is either a full replay or a skip, depending on earliest or latest. I keep the directory on durable shared storage, one path per query.

Q6. Why is the Kafka consumer-group offset the wrong recovery cursor for this query?

Structured Streaming records the ranges it has processed in its checkpoint and replays from there. The Kafka source integration does not use a group commit as that record. A group offset can be absent, ahead, or behind. Operating it as if it were the Spark cursor either skips batches or reprocesses them without the state that matches. Module 4's group-offset lesson applies to consumers that commit to Kafka. This job's runbook names the checkpoint path.

Q7. What exactly-once claim can you make for the file sink, and what claim can you make for foreachBatch?

With a replayable source and a checkpoint, the file sink is documented as exactly-once: a reprocessed batch does not add a second committed copy of the output. foreachBatch is at-least-once. The function can run again for the same batch_id. I get an exactly-once effect only if the function overwrites or otherwise deduplicates on that id or on the window key. An append inside the function doubles the tile after a failure. I use the file sink for this tile so the idempotence is the sink's, not a custom path I might get wrong.

Q8. What is maxOffsetsPerTrigger for?

It caps the source records, in offsets, that one micro-batch will read. After an outage the topic may hold hours of events. One uncapped batch turns into a very large shuffle and a long time before the next commit. The cap lets the query catch up in steps and keeps checkpoint progress moving. It is not a substitute for a watermark, and it does not split a hot course_id inside the batch. I still look at task skew the way lesson 30 taught.

Q9. When would you choose update or complete mode instead of append?

Update, when consumers need corrections before the watermark closes the window, and the sink can replace a key. A file sink cannot. Complete, when the result is a small aggregate you truly want rewritten every trigger, and you accept that state never drops. A dashboard of a handful of lifetime counters can be complete. A per-course window since the start of the company cannot. For the five-minute tile I stay on append so the file sink and the state cleanup both apply.

Q10. How should this tile relate to the morning course_day table?

The tile counts events in event-time windows and drops data later than the watermark. The morning table is built from landed files with an explicit restatement window and is the historical record. I do not sum the tiles to audit the morning table. I may reconcile a closed hour on a good day to catch a sink bug, and I expect late events to explain small gaps. The schemas share course_id and a time bucket. They do not share a promise about lateness.

Q11. What do you hand the Kafka module as requirements, without designing the cluster?

The topic name, the JSON fields, and UTC event time. A retention and replay guarantee that covers checkpoint recovery plus the watermark delay. Permission for this job to read the topic. Confirmation that lost offsets should fail the query. I do not ask Module 4 to commit this job's cursor into a consumer group. I do not embed broker counts in the Spark job. I do document failOnDataLoss, maxOffsetsPerTrigger, the checkpoint path, and the sink path so their on-call and ours can debug one incident.

Hands-On Exercises

Exercise: A tile that survives a second run

Problem statement. Implement the five-minute course activity tile against the file source. Demonstrate watermark closure, a clean restart, and a duplicate caused by a naive append.

Tasks.

1. Define the schema and the grain (course_id, window_start, window_end).
2. Build the aggregate with a ten-minute watermark and append mode. Use Trigger.AvailableNow in the lab.
3. Show that the window does not appear until a later event advances event time, and that it appears once it does.
4. Restart with the same checkpoint and no new input. Show that the sink does not gain another copy of the window.
5. Implement a foreachBatch that appends, force the function to run twice for one batch_id in a unit test or by calling it directly, and show the duplicate. Then overwrite by batch_id and show that the second call leaves one copy.
6. Write the production option list for the Kafka source, including failOnDataLoss and maxOffsetsPerTrigger, and state where the checkpoint must live.
7. State, in one paragraph, what the morning batch table promises that this tile does not.

Expected deliverables. The script, the input files, the output listing after one run and after restart, and the paragraph.

Evaluation rubric.

You pass when a second AvailableNow on the same checkpoint does not double the file-sink output, and when you can point at the foreachBatch path that would have.

Lesson Summary

Structured Streaming evaluates a SQL-shaped query as micro-batches over source offsets, with state and an offset commit stored in a checkpoint. Continuous processing is a different, experimental, at-least-once engine and is not this tile. Append mode plus a watermark emits each five-minute course window once and allows state to be dropped. The watermark has to be on event_time, before the aggregation. The file sink, with a durable checkpoint, is the exactly-once path documented for files. foreachBatch is at-least-once until the write is idempotent on batch_id. Kafka group offsets are not the cursor. startingOffsets matters when the checkpoint is absent. The morning course_day table remains the historical record, because the tile is allowed to drop late events.

Apache Flink Fundamentals is the next lesson, for the case where sub-second event time, large keyed state, or a transactional sink is the actual requirement. Module 4 takes the topic contract this query just named and designs the log.

Further Reading

• Structured Streaming Programming Guide, Spark 3.5.9 — output modes, watermarks, checkpoints, sinks, foreachBatch, and continuous processing. The guarantees in this lesson follow that guide.
• Spark Kafka integration guide (Structured Streaming section of the same programming guide) — kafka.bootstrap.servers, subscribe, startingOffsets, failOnDataLoss, and maxOffsetsPerTrigger.
• PySpark Trigger — ProcessingTime, AvailableNow, and Continuous.
• Module 4, starting at Streaming Data Architecture — the log this source assumes.
• Spark Performance Optimization — file counts and skew still apply inside a micro-batch.