Distributed Processing Reliability

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

Lesson Overview

On a Wednesday in March the TechLearningPro engagement job succeeded twice. The first attempt lost a shuffle file when an executor was removed, recomputed the stage, and wrote Tuesday's event_date. The second attempt was a human clicking "rerun" because the first attempt's metrics had not landed in the dashboard yet. Both attempts used mode("append") on the same Parquet path. Tuesday's event_count doubled. The completion product, which reads that date, moved. Nothing in the Spark UI looked failed after the second exit code. The retry did what retries do: it ran the job again. The sink did what an append does: it kept both copies.

A second incident, smaller and easier to miss, came from a streaming tile. Structured Streaming with Spark had been switched to foreachBatch during a sink experiment. The function generated a uuid for each output row and appended it. A task retry inside the micro-batch produced different ids, and the rerun of an uncommitted batch appended again. The file sink's exactly-once behavior had been the thing protecting the tile. The function replaced it with two sources of new identity: a random id, and an append.

Reliability in a distributed processor is the property that a failure and a rerun leave the curated grain true once. Spark batch provides that for computation by lineage: lost partitions are recomputed from their parents. It does not provide it for a sink that already exposed a partial write, or for a transform that calls uuid or rand or the wall clock inside a task. Structured Streaming adds a checkpoint so an uncommitted micro-batch repeats from the same offsets. The file sink is built to absorb that repeat. foreachBatch is not, until you make it so. Flink's aligned checkpoint, from Apache Flink Fundamentals, is the same idea in a continuous graph: one snapshot of state and source positions, plus a sink that commits or aborts with that snapshot. Exactly-once is a sentence about that whole chain. It is not a checkbox on the engine.

This lesson names the failure domains, the writes that can be repeated safely, and the late data you will restate rather than pretend the watermark was the business rule. The capstone turns the rules into a pipeline you can rerun for one event_date.

Learning Objectives

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

1. Define at-most-once, at-least-once, and exactly-once as claims about a specific hop, naming the source, the processor, and the sink.
2. Explain Spark lineage: which lost work can be recomputed, and which side effects a recomputation will repeat.
3. Describe what a Structured Streaming checkpoint commits, and what a Flink barrier snapshot commits, without mixing their cursors with a Kafka group offset.
4. Design an idempotent publish for one event_date: write a complete partition, then expose it, so a rerun replaces rather than appends.
5. Keep transforms deterministic: no uuid, rand, or processing-time "now" inside a task whose output is a key.
6. List the failure domains that matter here — task, executor, shuffle files, driver, checkpoint storage, source retention, sink partial write — and the recovery each one actually has.
7. Separate a watermark's drop of late streaming events from a batch restatement window that still accepts them.
8. Use a stable batch_id or the date partition as the idempotency key, and say why a random file name is not one.
9. Test a replay: run twice, compare counts, and show that the published grain did not double.
10. Refuse a design that deletes raw input in order to "make the retry safe."

Prerequisites

• Spark RDDs and Execution Model: lineage and wide dependencies. This lesson is what lineage does during a failure.
• Spark Joins and Data Shuffling: shuffle files are a failure domain. A deterministic salt stays valid across a recompute.
• Spark Performance Optimization: dynamic partition overwrite and file layout. The publish pattern below uses that overwrite on purpose.
• Structured Streaming with Spark: offset log, commit log, file sink, foreachBatch.
• Apache Flink Fundamentals: barriers, delivery guarantees, read_committed.
• Data Quality Engineering: a failed gate holds the previous curated version. Reliability is that rule under retries and crashes, not only under bad rows.

Detailed Core Concepts

Claims that name a hop

At-most-once means a failure can drop the effect and a retry will not redo it. You get this by not retrying, or by committing the input cursor before the sink is durable. It is rarely what a learning metric wants. It is what you get by accident when a consumer commits offsets and then crashes before the write.

At-least-once means the processor would rather repeat work than lose it. Spark task retries, Spark stage recomputes, streaming batches that are logged and not committed, and Flink's AT_LEAST_ONCE sink mode are all in this family. Duplicates are the engineer's problem at the sink.

Exactly-once means the effect of each input is visible once, including across the failures you designed for. Spark's programming guide attaches that phrase to replayable sources, checkpoints, and sinks that tolerate reprocessing. The file sink is in that table. A Kafka sink on Spark is not automatically in it. Flink's CheckpointingMode.EXACTLY_ONCE covers state. Flink's Kafka DeliveryGuarantee.EXACTLY_ONCE covers the topic only for consumers that read committed records. A Parquet path written with append covers none of it.

If you cannot name the reader that would observe a duplicate, you do not yet have a claim. "The job is exactly-once" is not a claim. "A rerun of event_date=2026-03-02 replaces that partition, and readers of other dates are unaffected" is a claim.

Lineage and its boundary

A Spark DataFrame remembers its parents. If an executor dies before a task finishes, the task runs again. If shuffle files are lost and the source files remain, the map stage runs again. Deterministic tasks produce the same rows. The job's result matches a run that never failed, until a task has a side effect: an append to an external table, an email, a call to an HTTP endpoint, a write that is already visible. Lineage will do the side effect again. Speculative execution (spark.speculation) can run a second copy of a slow task. Whichever copy commits, the other may already have written. Speculative execution and external side effects do not belong in the same task.

spark.task.maxFailures (default 4) is how many times a task may fail before the stage fails. Those retries are at-least-once from the task's point of view. Idempotent tasks make the number uninteresting. Tasks that append make the number a duplicate counter.

Lineage cannot recompute what you deleted. If the job removes raw files at the start, a failed stage has nothing to read. Raw stays until the curated publish for that date is durable and superseded by policy, not by the first attempt's optimism. Data Quality Engineering made the same rule for a failed gate. A crash is not a reason to weaken it.

Deterministic tasks

A recompute is the same result only when the function of the input is the same. These are not:

• F.uuid() or any call that mints an identifier inside the task, used as a primary key.
• F.rand() used as a salt that is then stored, or as a sample without a fixed seed.
• F.current_timestamp() stored as event_time. Processing time changes on retry. Event time was already on the row.
• A read of an external service that can answer differently the second time, such as "current enrollment" from a live database, baked into the fact without a snapshot id.
• Map-side writes that append to a shared path.

The salt in lesson 30 used pmod(hash(event_id), n) and then dropped the column. A retry hashes the same event_id the same way. That is the standard. A snapshot of enrollments, identified by a date or a manifest, is the standard for the live-service problem. You join the snapshot you named. You do not join "whatever the database says during the task."

dropDuplicates without a subset of columns uses the whole row. It is deterministic. dropDuplicates on a frame that includes a uuid column you just added will not collapse anything, because every retry minted different ids and, within one attempt, every row is already unique. Deduplicate on the business key: event_id for events, (learner_key, course_id) for the enrollment grain.

Streaming cursors, restated once

A Structured Streaming checkpoint records offset ranges and commits. An uncommitted range is processed again. The file sink is built so that repeat does not add a second committed output. foreachBatch must overwrite a path that includes batch_id, or replace rows by the window key. Flink records source offsets and keyed state in the checkpoint barrier. The Kafka sink's transaction commits on the completed checkpoint or aborts. In both engines the Kafka group offset is a display. Lesson 32 and lesson 33 own the details. The reliability rule that spans them: never operate two cursors as if they were one. Resetting a group, deleting a checkpoint, or retaining a checkpoint and also replaying from earliest into an append sink are three different incidents. Pick one cursor, write it in the runbook, and test a kill during the sink.

Source retention is part of the cursor. A checkpoint that points at an offset the broker has deleted cannot be restored. failOnDataLoss set to true turns that into a failed query instead of a hole. The human decision — new checkpoint and a declared skip, or a topic that still has the bytes — belongs in an incident, not in a default of false.

Idempotent publish for a date partition

The batch pattern this course uses for course_day and for the capstone fact:

1. Read raw for the dates in scope. Do not delete raw.
2. Transform deterministically. Carry event_date and a published_at you set once on the driver, not per row inside a task, if you need a publish timestamp. Better: store the job id as a column from a single lit value passed into the job, so every row in the attempt shares it.
3. Write the attempt to a staging path that includes the job id: .../staging/job_id=20260303T0600Z/event_date=2026-03-02/.
4. Validate counts against the quality gate. If the gate fails, do not publish. The previous published partition stays.
5. Publish by replacing the target partition for that event_date only. spark.sql.sources.partitionOverwriteMode=dynamic is the documented session setting that overwrites partitions present in the DataFrame and leaves other partitions. Writing to a path that already ends in event_date=2026-03-02 with mode("overwrite") is the other documented approach. Both require a test that a neighbor date survives.
6. Readers are pointed at the published path, never at the staging path.

A rerun uses a new job id, a new staging directory, and the same published partition. The second publish replaces the first. Append at the published path is forbidden. Appending at staging is harmless if each job id is unique and readers cannot see staging.

This is idempotent overwrite. It does not require a transaction manager. It requires that readers do not observe the partition mid-overwrite. On a filesystem where rename of a directory is atomic, write the new partition beside the old one and rename. On object storage, directory rename is not a reliable atomic operation. Hadoop's S3A committers exist for that problem; this course does not configure a fictional one. The practical pattern on object storage is: write the new partition to a complete prefix, validate it, then update a small manifest object that names the prefix readers should use. The manifest write is one object. Readers that load the manifest and then the prefix either see the old prefix or the new one. They do not see a mixture, provided they do not list the staging parent and pick files themselves. The capstone uses a manifest pointer for that reason and labels it as an application convention, not as a Spark built-in.

foreachBatch maps onto the same idea. The micro-batch id is the job id. Overwrite batch_id=N or replace the window keys that batch owns. Do not append.

Late data

A watermark drops events from a streaming window so state can end. That drop is a live-tile rule. It is not a statement that the event never happened. The batch fact restates a trailing window of dates: for example, each morning job recomputes event_date for the last three days from raw, and publishes each date with the overwrite pattern. An event that arrived a day late, with an event_time on Monday and an ingest_date on Tuesday, enters Monday's partition on Tuesday's run. The tile for that five-minute window may already have closed without it. The two outputs are allowed to differ by the late events. They are not allowed to differ because one of them appended a retry.

How long the restatement window stays open is a product decision. Three days covers a weekend laptop. It does not cover a client that syncs a month later. Those rows can land in a late partition with a flag, or extend the window, if the metric owner signs the definition. Silently extending forever means Monday's number never settles. Write the horizon down.

Failure domains

The orchestrator, Airflow or otherwise, can rerun a batch job. It cannot invent idempotence the job did not have. A green "retries: 2" on an appending job is how Wednesday doubled.

Architecture and Design Diagrams

mermaidCopy
flowchart TD
    raw[(Raw Parquet kept)]
    job[Deterministic Spark job]
    stage[Staging prefix for one job_id]
    gate{Quality gate}
    manifest[Published manifest pointer]
    old[Previous prefix for that event_date]
    reader[Readers of course_day and facts]

    raw --> job --> stage --> gate
    gate -->|pass| manifest
    gate -->|fail| old
    manifest --> reader
    old --> reader
    raw -.->|recompute on lost shuffle| job

The reader follows the manifest. A failed gate does not move it. A second job id writes a different staging prefix and, on success, moves the pointer. The first staging prefix can be deleted after the pointer has moved and a retention delay has passed. It is not deleted before the publish.

A streaming hop beside it:

mermaidCopy
flowchart LR
    topic[Replayable topic]
    chk[(Engine checkpoint)]
    state[Processor state]
    sink[Idempotent or transactional sink]
    topic --> state
    chk --> state
    state --> sink
    sink -->|commit agrees with checkpoint| topic

The arrow from sink back toward the source is the agreement of cursors: Spark's commit log with the file sink's batch, or Flink's completed checkpoint with a Kafka transaction. An append path has no such arrow. It only grows.

Technical Explanation

Where to mint identifiers

Mint the job id on the driver, once, and pass it as a literal column:

pythonCopy
job_id = "20260303T0600Z-engagement"  # argument, not F.uuid()
frame = frame.withColumn("job_id", F.lit(job_id))

Every task sees the same job_id. A retry of the task does not change it. A second orchestrated attempt gets a new argument and a new staging prefix. If you need a row identity and event_id already exists, use event_id. If you are aggregating, the grain (course_id, event_date) is the identity. Adding a surrogate key in the task helps nothing the grain did not already name.

Dynamic overwrite, tested

spark.sql.sources.partitionOverwriteMode set to dynamic changes overwrite so that only partitions in the write are replaced. The default static replaces the set of partitions the write is aimed at, and if you aimed at the table root you can delete the month. Set the mode on the session that submits the job, not in a notebook you might forget. The test is four steps: write date A, write date B, rewrite A, read B unchanged and A equal to the second write. Put that test in the capstone. A manifest pointer makes the same promise when object-storage rename is the weak step: the pointer object names A's prefix, and rewriting A publishes a new prefix for A only.

ForeachBatch idempotence

pythonCopy
def publish_windows(batch_df, batch_id: int) -> None:
    # Readers query the published window table, which is replaced per window key
    # inside this batch. A second call with the same batch_id writes the same keys.
    keys = batch_df.select("course_id", "window_start").distinct()
    # The delete-then-append must itself be repeatable. Overwrite a batch path instead
    # when the downstream reader can use that path via a manifest.
    (
        batch_df.write.mode("overwrite")
        .parquet(f"{warehouse}/tile_batches/batch_id={batch_id}")
    )

Prefer the overwrite of batch_id to a delete-then-append against a shared table. Delete-then-append has a crash window where the delete succeeded and the append did not. Overwrite of a batch-specific path has a crash window too, but the previous published manifest still points at the last good batch. The pointer moves after the listing of the new path matches the expected row count.

Speculation and commits

Leave spark.speculation at its default off for these jobs. If a platform turns it on globally, the engagement job should set it off. Two attempts of a task that both try to write staging files for the same task id will corrupt a naive file writer. Spark's own file output for a DataFrame write commits task output through Hadoop's output committer so that a speculated duplicate does not publish two copies for that Spark write. That protection is the committer's, for that action. It does not cover a Python function that opened its own client and posted rows. Keep external writes on the driver, after an action, and make them idempotent.

Late events in the batch grain

pythonCopy
# event_date is the business date from event_time, not the ingest date.
# The job's --dates argument lists the restatement horizon.
events = spark.read.parquet(raw_root).where(F.col("ingest_date").between(horizon_start, horizon_end))
events = events.withColumn("event_date", F.to_date("event_time"))
events = events.where(F.col("event_date").isin(restate_dates))

Filtering only on ingest_date and also partitioning the fact by ingest_date hides late events inside the wrong day. The fact's partition is event_date. The read filter must be wide enough on ingest time to catch stragglers, then the write publishes the business dates those stragglers belong to. Otherwise Monday never picks up Tuesday's backpack sync, or Tuesday contains Monday's events and the days drift.

Quality gate as a publish bit

The gate from lesson 17 returns a publish bit. Reliability means the bit is consulted after staging and before the manifest moves. A job that writes straight to the published path and then checks quality has already exposed the bad partition. Check, then move the pointer. On a gate failure, leave the pointer. Alert. Do not delete staging until a human has looked, and do not delete raw at all in this step.

Practical Implementation or Design Walkthrough

Problem

Publish course_day for one or more event_date values so that a rerun and a lost executor leave each date's counts equal to one deterministic computation. Show the test.

Code

pythonCopy
"""Idempotent publish of course_day for an explicit list of event dates.

PySpark 3.5.x. The manifest is a JSON file the job writes after a successful gate.
It is an application convention: Spark does not provide this pointer as a built-in.
"""

import json
from pathlib import Path

from pyspark.sql import DataFrame, SparkSession
from pyspark.sql import functions as F

RESTABLE_DATES = ("2026-03-02", "2026-03-03")


def build_spark() -> SparkSession:
    return (
        SparkSession.builder.appName("tlp-course-day-publish")
        .master("local[*]")
        .config("spark.sql.adaptive.enabled", "true")
        .config("spark.sql.sources.partitionOverwriteMode", "dynamic")
        .getOrCreate()
    )


def course_day(events: DataFrame) -> DataFrame:
    # event_date comes from event_time. No uuid, no rand, no current_timestamp.
    prepared = events.withColumn("event_date", F.to_date("event_time")).where(
        F.col("event_date").isin(*RESTABLE_DATES)
    )
    return prepared.groupBy("course_id", "event_date").agg(
        F.count("*").alias("event_count"),
        F.count("manifest_id").alias("matched_lesson_count"),
    )


def write_staging(summary: DataFrame, staging_path: str) -> None:
    (
        summary.write.mode("overwrite")
        .partitionBy("event_date")
        .parquet(staging_path)
    )


def gate(summary: DataFrame) -> bool:
    """A stand-in for the lesson 17 publish bit.

    Fail closed: null course ids, or a date in the request that produced no rows
    while the caller expected data, are decided by the caller. Here, null keys fail.
    """
    bad = summary.where(F.col("course_id").isNull() | F.col("event_date").isNull()).count()
    return bad == 0


def publish_manifest(staging_path: str, manifest_path: str, job_id: str) -> None:
    pointer = {"job_id": job_id, "staging_path": staging_path, "dates": list(RESTABLE_DATES)}
    Path(manifest_path).write_text(json.dumps(pointer), encoding="utf-8")


def run(events: DataFrame, job_id: str, warehouse: str) -> bool:
    summary = course_day(events)
    staging = f"{warehouse}/staging/job_id={job_id}"
    write_staging(summary, staging)
    # Read back from staging so the gate sees the bytes that would be published.
    published_view = events.sparkSession.read.parquet(staging)
    if not gate(published_view):
        print(f"gate_failed job_id={job_id}")
        return False
    publish_manifest(staging, f"{warehouse}/manifest/course_day.json", job_id)
    return True

publish_manifest writes a local file in the lab. On object storage the same bytes go to one object, written by the driver after the Spark action, not by each task. A crash before the manifest write leaves readers on the previous pointer. A crash during the staging write leaves an incomplete staging prefix that the gate should reject because the read-back counts fail your reconciliation, or because the job never reaches the manifest step. Do not point the manifest at a staging path you have not read back.

The replay test

Construct a fixture of one thousand events for 2026-03-02 and run run twice with job ids a and b. After the second success, the manifest's job_id is b. Read the Parquet at that staging path and assert sum(event_count) equals one thousand, not two thousand. Read job a's staging path and accept that it still holds a thousand as an abandoned attempt. Readers that follow the manifest do not see both. Then kill the process before publish_manifest in a third run and assert the manifest still names b.

That test is the reliability acceptance test. A unit test of course_day without a second publish does not catch the append bug, because the bug is in the publish.

Streaming replay, in one paragraph of procedure

Start the file-sink tile from lesson 32. Kill the driver after at least one batch has committed. Restart with the same checkpoint. Counts of closed windows stay equal to one computation. Repeat with a foreachBatch that appends, and record the duplicate. The pair is the demonstration you want in a design review, and it uses only the Spark APIs lesson 32 already named.

Enterprise Real-World Scenarios

Scenario 1: The doubled Tuesday

Orchestration retried the engagement job after a spot-instance executor disappeared. The first attempt had already appended Tuesday before the driver reported success to the scheduler — the write completed, the metric post failed, the scheduler retried. Append made two Tuesdays. The dashboard showed a completion rate the content team could not explain, and the on-call response was to divide by two, which happened to be right once and will be wrong the next time the first attempt had not finished the write.

The repair is staging plus a manifest, a gate that runs against staging, and a scheduler retry that is safe because the publish replaces. The metric post becomes a reader of the manifest, not a side effect inside the Spark tasks. Raw files for Tuesday stay. A recompute after a lost shuffle reads them. The incident review's artifact is the replay test: two job ids, one visible sum.

Scenario 2: A streaming retry with new identity

The tile's foreachBatch added F.uuid() so the warehouse would have a primary key. The key changed every retry, so the downstream merge inserted new rows instead of matching old ones. Window counts climbed through the evening. The checkpoint was healthy. The sink was "idempotent" in a design doc that assumed the primary key was the window.

The repair drops the uuid. The key is (course_id, window_start). The write overwrites batch_id. A replay test kills the query during the sink. The Spark file sink is the alternative if the team does not want to own that function. Flink's transactional Kafka sink is the alternative if the reader is a Kafka consumer that can set read_committed. Each alternative has a reader. The uuid had none.

Scenario 3: Late backpack sync

A mobile client delivers Monday's events on Wednesday. The streaming tile for Monday's 10:00 window has already closed on a ten-minute watermark. The batch job's restatement horizon is three days, and Wednesday's run includes event_date=Monday because event_time says Monday even though ingest_date says Wednesday. Monday's published partition moves forward once, via the manifest. The tile is not rewritten. Product copy on the live tile says "counts update as events arrive, and may lag a reconnecting client." Product copy on the daily report says "restated for three days." Those sentences are the reliability design as the business sees it. Without them, someone will "fix" the discrepancy by lengthening the watermark until state never drops, or by shortening the batch horizon until Monday is wrong.

Best Practices

• Name the hop when you say exactly-once: source, processor, sink, reader.
• Keep raw immutable for the restatement horizon. Recompute from it. Do not delete it to simplify a retry.
• Mint job ids and publish timestamps once, on the driver, and pass them with F.lit. Do not mint keys with uuid inside tasks.
• Publish a date by replacing that date: dynamic overwrite on a filesystem you trust, or a manifest pointer when the store does not give you an atomic directory replace.
• Run the quality gate against the staged bytes. Move the pointer only after the gate passes.
• Test two publishes and a crash before the pointer. The visible sum must match one computation.
• Keep streaming checkpoints on durable storage. Overwrite by batch_id if you leave the file sink.
• Turn speculative execution off for any job that performs its own external writes.
• Restate a bounded set of event_date values from event time. Let the watermark own the live tile, not the historical fact.
• Write the failure domain next to the alert: shuffle loss, checkpoint loss, and doubled append are different pages.

Common Mistakes and Anti-Patterns

Append as the retry story. The scheduler's retry is at-least-once. Append turns that into user-visible duplicates. Replace the partition or move a pointer.

uuid or rand in the task. Recomputes and speculative copies become new rows. Hash a stable id or use the grain.

Deleting raw at the start of the job. Lineage has nothing to read after a lost shuffle. Delete published staging only after the pointer has moved and aged.

Two cursors. Resetting a Kafka group and also restoring a checkpoint, or restoring a checkpoint into an append path, double-applies. One cursor, one sink behavior.

failOnDataLoss=false as a permanent setting. You trade a loud offset gap for a quiet hole in the tile. Keep the failure, and make retention long enough that the failure is rare.

Gate after publish. Readers already saw the bad day. Gate the staging prefix, then publish.

Static overwrite of the table root. A one-date repair deletes the rest. Dynamic mode or a date-specific pointer, plus the neighbor-date test.

Using the watermark as the only lateness policy. The live window must close. The business date must restate. One number for both will either drop Monday's backpack or hold streaming state open for days.

Side effects inside foreachPartition. A retry posts the payment or the push notification again. Write curated data idempotently. Send notifications from a consumer that records the key it already sent.

Calling Flink's checkpoint mode the end of the story. State restores once. The sink and the reader still decide whether the effect is once. The same limit applies to Spark's file sink: a reader that copies files out of staging without the manifest sees abandoned attempts.

Troubleshooting and Design Review

Symptom: A date's event_count is a multiple of the source count. The job log shows two successful attempts.

Likely cause: append, or a manifest that was never consulted and a reader listing every staging prefix.

Investigation: count attempts in the scheduler, list staging prefixes, read the manifest job_id, sum the source events for that event_date.

Corrective action: point readers at the latest good manifest, republish once from raw, and remove append from the published path. Do not divide the metric by the number of attempts.

Symptom: After an executor loss, the job fails with missing shuffle blocks and cannot recompute.

Likely cause: source files were deleted, or the input was a streaming source that cannot replay that range.

Investigation: the stage's input paths. Whether raw for that date still exists.

Corrective action: restore raw from the archive if you have it, and change the job so it does not delete raw. If the bytes are gone, the date is a declared hole, not a silent zero.

Symptom: The streaming query restarts and closed windows duplicate.

Likely cause: foreachBatch append, or a new checkpoint directory with startingOffsets earliest writing into the same sink.

Investigation: checkpoint path identity across the restart, sink mode, batch ids present in the output.

Corrective action: one checkpoint path, file sink or batch-id overwrite, compact the duplicated windows by reprocessing from the checkpoint's last good offset if you still have it.

Symptom: Monday's daily count is missing events that Wednesday's raw files contain. The tile matches the lower number.

Likely cause: the batch job filtered on ingest_date equal to event_date, so late events never enter Monday's publish. The tile dropped them by watermark, which is expected.

Investigation: count raw rows with to_date(event_time) = Monday and ingest_date > Monday.

Corrective action: include those ingest dates in the restatement read, and publish Monday again through the manifest. Leave the tile closed. Document the horizon.

Symptom: Two readers disagree during the publish, one high and one low, for a few minutes.

Likely cause: they listed the partition while an overwrite was replacing files, or one read staging and one read the old prefix.

Investigation: whether readers use the manifest object.

Corrective action: readers load the pointer, then the prefix it names. They do not list a parent that contains every attempt. The pointer moves only after the new prefix is complete.

A design review asks: What is the hop, and who is the reader? What is the idempotency key? What happens if the process dies before the pointer moves? What happens if it dies after? Can a task recompute without raw? Which functions are non-deterministic? How long can a business date change, and what does the live tile refuse to change? Which failure domain is this alert for? A review that says "we retry three times" has described the scheduler. It has not described the sink.

Interview Questions and Answers

Q1. What do at-most-once, at-least-once, and exactly-once mean if you are not allowed to wave at the engine?

Each one is a claim about a hop with a source, a processor, and a reader. At-most-once: a crash can drop the effect, and we do not redo it. At-least-once: we redo work, and the reader may see duplicates unless the sink folds them. Exactly-once: the reader observes one effect per input across the failures we planned for. Spark lineage gives at-least-once execution of tasks. The file sink plus a streaming checkpoint is documented as exactly-once for that sink. An appending Parquet path is exactly-once for nobody. I name the reader before I use the phrase.

Q2. What can lineage recompute, and what will it repeat that you did not want repeated?

It can recompute lost partitions from parent RDDs or DataFrame operators, including redoing a shuffle's map stage when the source files still exist. It will also repeat any side effect the task performed: an append, an HTTP post, a notification. Speculative execution can start that side effect twice. I keep tasks free of external writes, mint no random keys, and do the publish from the driver after the action, idempotently. If the source was deleted, lineage has no parent bytes, and the stage cannot be saved by a retry count.

Q3. Why is F.uuid() as a primary key incompatible with retries?

A retried task mints new ids. A downstream merge treats them as new rows. dropDuplicates on that column drops nothing. The business key was already available: event_id, or (course_id, event_date), or (course_id, window_start). I pass a job id with F.lit from the driver when every row of an attempt should share a publish token. The id is stable across task retries and different across orchestrated attempts, which is what staging paths need.

Q4. How do you publish one event_date so a rerun does not double it?

Compute into a staging prefix named by the job id. Run the quality gate on the bytes you wrote. If it fails, leave the published pointer where it is. If it passes, point a single manifest object at that prefix. Readers follow the manifest. A second attempt writes a new prefix and moves the pointer. On object storage I do not depend on a directory rename being atomic. Dynamic partition overwrite is the alternative when the filesystem provides the replacement semantics and I have tested that a neighbor date survives. Append at the published location is the failure we already had.

Q5. What is the crash window of your publish?

If the job dies during staging, the manifest still names the previous prefix, and readers see the last good date. If it dies after the manifest write, readers see the new prefix, which the gate already read back. The bad window is a manifest write that names a prefix whose files are not durable yet. I write the manifest only after a read-back and the gate. I do not write the manifest from inside tasks. Abandoned staging prefixes are invisible to readers and can be deleted by a later retention job that refuses to delete the prefix the manifest currently names.

Q6. How does a Structured Streaming checkpoint differ from a Flink checkpoint in one reliability sentence each?

Spark's checkpoint logs the offset range of a micro-batch and a commit when the sink finishes, and the file sink is built so a repeated batch is not a second output. Flink's barrier snapshots operator state and source positions together; the Kafka sink commits or aborts a transaction with that snapshot when the delivery guarantee says so. In both cases a Kafka consumer-group offset is not the cursor. Losing the checkpoint storage is a human decision about a new cursor. I do not restore by resetting the group and also keeping the old checkpoint.

Q7. Where should speculative execution sit relative to external writes?

Off, for any job whose tasks talk to a system outside Spark's committer. Spark's file committer already handles speculated task output for a DataFrame write. A Python hook that posts rows does not. The engagement job sets speculation off and publishes from the driver. I would rather have a slow task than a second notification.

Q8. A watermark dropped a late event and the daily fact later included it. Is that a bug?

Not by itself. The tile's watermark is how long a live window stays open. The batch horizon is how long a business date may be restated from raw. They answer different questions. It is a bug if the batch job partitions by ingest date and never revisits the event date, or if product documentation claims the tile and the daily fact are the same number. I restate the event date from raw through the idempotent publish, and I leave the closed tile closed.

Q9. What do you do when failOnDataLoss fires because the broker no longer has the checkpoint's offset?

I do not flip the flag to false as the repair. The bytes are gone. I record the gap, agree with the owner whether the tile may skip or whether an upstream archive can replay, and I start a new checkpoint at a declared offset. I also fix retention so the replay window is longer than the realistic outage plus the watermark. A quiet skip becomes a permanent undercount that the daily fact, if it still has the raw files, will disagree with.

Q10. How do you test reliability without causing a production incident?

With a fixture and two publishes: the visible sum equals one computation, the neighbor date is unchanged, and a crash before the manifest write leaves the previous pointer. Separately, for the stream, kill the query after a commit and restart on the same checkpoint, then show that an appending foreachBatch doubles and the file sink or the batch-id overwrite does not. I include a non-deterministic function in a negative test so uuid fails code review with evidence. I do not call a green scheduler retry a test. The scheduler did not look at the sum.

Q11. Why is deleting raw a reliability defect even when the publish succeeded?

Because the next failure mode is a bad publish you need to restate, a late file, or a lost derived partition. The restatement horizon is exactly the period raw must remain. A successful job that deletes its inputs has converted the next bug into data loss. Staging prefixes can expire. Raw expires on the retention the owner signed, after the horizon, not at the end of main().

Hands-On Exercises

Exercise: One date, two attempts, one sum

Problem statement. Publish a course-day aggregate so retries and a crash before the pointer move do not change the number readers see, except when the second attempt is the new good publish.

Tasks.

1. Build a fixture with a known count for one event_date and a second date that must survive untouched.
2. Publish twice with different job ids using staging and a manifest. Assert the manifest names the second id and the visible sum equals the fixture once.
3. Crash or return before the manifest write on a third attempt. Assert the visible sum is unchanged.
4. Add an appending variant and show the doubled sum. Keep that variant only as a negative test.
5. Include one late event whose ingest date differs from event_time. Show that a restatement read picks it up and a filter on ingest date equal to event date does not.
6. Replace any uuid or rand in the aggregation path. Explain where a job id may be minted.
7. Write the failure-domain row for "scheduler retried the job" and for "shuffle files lost and raw remains."

Expected deliverables. The script, the assertions, and the two failure-domain sentences.

Evaluation rubric.

You pass when the appending variant is the only one that doubles, and the late event changes Monday only after a deliberate republish.

Lesson Summary

A retry repeats work. An appending sink turns that repeat into a second Tuesday. Lineage recomputes lost partitions from raw that you kept, and it repeats any side effect you hid in a task. Deterministic functions, a job id minted once, a staged prefix, a gate on those bytes, and a manifest the reader must follow are the batch shape that makes two attempts equal one sum. Streaming checkpoints in Spark and barrier snapshots in Flink are the cursors for their hops. Kafka group offsets are not those cursors. Watermarks close live windows. A bounded restatement of event_date from raw is how late events still change the historical fact, once, through the same publish.

The capstone, Big Data Processing Capstone, asks you to build that pipeline at the scale of the engagement job: schemas, shuffles, file sizes, the gate, the replay test, and the monitors that tell you which failure domain you are in.

Further Reading

• Structured Streaming Programming Guide, Spark 3.5.9 — checkpointing, the file sink's exactly-once note, and foreachBatch at-least-once behavior.
• Spark 3.5 configuration — spark.task.maxFailures, spark.speculation, and spark.sql.sources.partitionOverwriteMode.
• Flink checkpointing and the Kafka connector delivery guarantees — the other hop's snapshot and reader rules.
• Hadoop S3A committers — documented work on object-storage commit semantics. This lesson uses a manifest pointer so it does not depend on a specific committer class.
• Data Quality Engineering — the publish bit the manifest is waiting for.