Structured Streaming: Intermediate

A streaming query is meant to run for weeks, and the processes running it will die along the way: a node gets reclaimed, a deploy restarts the driver, an executor runs out of memory. The checkpoint is how the next run learns where the last one stopped. It is a folder on durable storage, named by the checkpointLocation option on the writer, that the query updates on every micro-batch.

Inside /chk/payout_lines/, each micro-batch leaves files named by its batch number. Here is what each entry means, using batch 41 as the example:

An offset is a source's position marker. For a partitioned message log it is a record number per partition; for the file source it is an entry number in the file list it keeps under sources/. Batch 41 starts where offsets/40 ended and stops at offsets/41, so the pair pins down exactly which rows belong to the batch: here, 3 files holding 634 trip events.

When a query starts with a checkpointLocation that already has content, Spark finds the highest batch number in offsets/ and in commits/ and compares them before reading any new data.

Any state the query had built is gone as well. Spark protects most sinks from this: the file sink and foreachBatch refuse to start without a checkpointLocation, while the console sink, meant for debugging, gets a temporary folder that is deleted when the query stops.

Replay has an expiry date. If cleanup deletes trip files, or the log's retention window passes, before the query restarts, the range recorded in offsets/41 no longer exists and the contract fails at its first party.

Upserts and batch-id tracking need a target that supports them, such as a database table with a key. Overwriting by batch needs only a folder. foreachBatch calls your function with each batch's DataFrame and its batch id, and a re-run of batch 41 arrives with id 41 again. Overwriting by batch and recording the batch id both key on that repeated id. The upsert keys on trip_id instead, a business key that is the same in both copies of a trip. Here is write_payout_lines rewritten with the first pattern:

A transactional sink makes a batch's rows and its batch id visible in a single atomic step: both or neither. Spark's file sink works this way. Tasks write part files into the output folder, then the driver writes _spark_metadata/41, a log entry listing exactly those files. A re-run of batch 41 finds that entry and skips the write. Spark jobs that read the folder take their file list from _spark_metadata, so part files left by a crashed attempt stay invisible to them; a tool that lists the folder directly still sees them.

The version of write_payout_lines running that night, before the rewrite in the previous section, appended each batch to the payouts database over JDBC and stamped every row with its batch id. Here is batch 41, step by step:

The third row sits outside Spark's contract. Exactly-once means each input record lands once, and 2 deliveries of 1 event are 2 records. Phones cause this: an app that loses signal after sending an event, before the acknowledgement arrives, sends it again with the same event id. The app_event_deliveries table has that shape: 1 row per delivery, with its own delivery_id and received_ts, and event_uuid naming the event. Run the cell to compare deliveries with distinct events.

from pyspark.sql import functions as F

deliveries = spark.table("app_event_deliveries")

result = deliveries.agg(
    F.count("*").alias("deliveries"),
    F.countDistinct("event_uuid").alias("distinct_events"),
)
result.show()

Any gap between the 2 numbers is redelivery. deliveries.dropDuplicates(["event_uuid"]) keeps 1 row per event. distinct() would keep every copy, because 2 deliveries of an event differ in delivery_id and received_ts. On a stream, the dedupe step must remember event ids from every earlier batch to catch a copy that arrives a day late, which makes it stateful; the last section of this lesson prices that memory.

Spark checks the mode against the plan when you call start(), before any data is read, and raises an AnalysisException. A sort needs the full result in hand, which only complete mode provides, so orderBy is legal there and nowhere else. The 2 rejections you will meet most often read:

A mode the query allows must also be one the sink accepts. The file sink writes immutable files and can't revise a row it has already written, so it takes append only. foreachBatch takes all 3 modes and hands your function the rows the mode selects, leaving it to your code to apply them correctly. The message-log sink also takes all 3, publishing each changed row as a new record for its consumers to reconcile.

Here is a per-city leaderboard of drivers, built on the same trips stream and checkpointed to /chk/leaderboard/:

Read it top down. The filter and the join to cities, a 2MB static table under the 10MB broadcast threshold, are stateless: they pipeline inside the first stage of each batch, and the broadcast adds no shuffle. The groupBy is the stateful part. It hash-shuffles the batch's rows on city_name and driver_id, and each task after the shuffle merges its slice of the batch into the running totals it holds for those keys.

The number of state partitions is the value of spark.sql.shuffle.partitions at the query's first start, and Spark records it among the settings in each offsets file. Every later restart reuses the recorded value and ignores the one you set, because each key's state was filed under the partition its hash chose on the first day. At the default of 200, a stateful query runs 200 reducer tasks on every batch, even a batch of 12 rows, and each task loads and saves its slice of state. Adaptive query execution will not coalesce them: Spark turns it off for streaming queries and logs a warning if it's enabled. Choose the count from the expected state size before the first start.

State size follows the number of distinct keys an operator has seen, and a key stays until something removes it. The leaderboard keeps 1 row per city and driver, which levels off once every active driver has appeared. dropDuplicates(["event_uuid"]) keeps every event id it has ever seen: at 50 million events a day, a month of ids is 1.5 billion keys held across the executors and saved into the checkpoint. Bounding that growth by event time is a later lesson's subject; until then, size any stateful operator over an unbounded key before it ships.