PySpark Data Engineering

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

Lesson Overview

Spark DataFrames and Spark SQL built the expressions that classify a progress line and compare canonical hashes. Those expressions are not a landing. A landing is a process the scheduler can run again: configuration for the date and the paths, a content hash of the input, a refusal that does not touch the published dataset, a quarantine for ordinary bad lines, a commit of the affected event_date partitions, and a checkpoint written after that commit. Building Batch Data Pipelines with Python was this process on one host. This lesson is the same process when the plan runs as a Spark application, with the driver still small enough to be the thing that decides and records.

TechLearningPro's published dataset remains Parquet under event_date, readable by completion-v3 and not a place where completion is computed. The input remains a closed JSONL file, or a closed set of files you hash as one identity. The prohibited fields remain email, name, and payment_token. The new machinery is the one Apache Spark Architecture and Hadoop Ecosystem and HDFS prepared: cluster deploy mode, a committer that matches the filesystem, dynamic partition overwrite so a rerun of one date does not delete the rest of the table, and a checkpoint prefix that is not inside the dataset root.

The code in this lesson is the job skeleton you can run on local[2] against a fixture and submit, with different configuration, to a cluster when the lesson 24 memo says to distribute. PySpark 3.5 is the API. Java 17 is the runtime those builds expect. Python on the driver is still the language from /learn/python and lessons 4 and 5. Python on the executors stays out of the per-row path because lesson 27 already showed the cost.

Learning Objectives

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

1. Structure a landing as configuration, pure transforms, and a driver main that performs actions in a safe order.
2. Publish by dynamic partition overwrite so one input replaces only the event dates it owns.
3. Refuse a file that spans more dates than the contract allows, before any overwrite.
4. Write the checkpoint only after the Spark write returns, and skip a rerun when the hash and the partitions are already recorded.
5. Quarantine ordinary rejects under a cap without placing them inside the dataset root.
6. Choose an output file count for a peak day instead of accepting one file per task by accident.
7. Test the contract on a local session, including the overwrite mode that would wipe a table.

Prerequisites

• Spark DataFrames and Spark SQL: classify_lines, conflict_keys, and accepted_events. This lesson calls them. It does not redefine the hash.
• Apache Spark Architecture: cluster deploy mode, actions, and a driver that must not collect the day.
• Hadoop Ecosystem and HDFS: commit visibility, _SUCCESS, and why the checkpoint prefix sits outside the dataset.
• Building Batch Data Pipelines with Python: hash, cap, replace, checkpoint last. The order is identical. The replace primitive changes.
• Big Data Fundamentals: this job exists because a memo said distribute. If the memo said stay, you are not required to submit it.
• A local PySpark 3.5 session for tests. A cluster is optional and must not be implied by a green unit test.

Detailed Core Concepts

The driver owns the decision, the executors own the rows

The production shape is three layers. Configuration is a frozen object: paths, the product name, the reject cap, the maximum number of event_date values one input may replace, and how many Parquet files to write per date. Transforms are functions from lesson 27. They do not read environment variables and they do not write. The driver in main hashes the input, builds the plan, runs the decision actions, and either returns without a publish or writes and then checkpoints.

That split is what keeps a test honest. A test constructs a BatchConfig pointed at temporary directories, calls the same run function, and asserts on files and checkpoint JSON. A test that shells out to spark-submit against a shared cluster is an integration test, and it is welcome later. It is not the contract test.

The driver is allowed to hold: the hash, the list of event dates (a handful, or a refusal), counts, and the checkpoint document. The driver is not allowed to hold the events. Every collect in this job is on an aggregate or on distinct of event_date.

Dynamic overwrite is the replace

mode("overwrite") on a path, with the session in static partition overwrite mode, deletes the dataset and writes the new rows. Static is the Spark default (spark.sql.sources.partitionOverwriteMode = static). Pointed at /data/tlp/raw/progress, a rerun of 2 March deletes January. That is the Wednesday incident at table scale.

Dynamic mode overwrites only the partitions present in the DataFrame being written. A DataFrame that contains only event_date=2026-03-02 replaces that directory and leaves other dates. This is the distributed analogue of os.replace on one day's file, with an important difference: the unit of replace is the entire partition directory, not a file beside other files. Whatever rows you write become the whole date. If the input is not the complete set of events for that date, you have just deleted the missing ones.

The contract this lesson requires of the producer is the same one lesson 12 stated for a closed file: the input is complete for the dates it will replace. A daily export named for 2 March must contain the events that belong to that publish, and a rerun of the same export replaces them. A second file of "late events only" must not be overwritten into the same partition by itself. Late events are a restatement of the full date, or a different design that merges under the same collision rule. This job does not merge two inputs. It replaces. Merging would be a new product, and it would need a primary key across files. Say so in the run book so an operator with a small correction file does not "just rerun the landing."

Before the write, the driver collects distinct event_date values. If the set is empty, there is nothing to publish and the behavior follows the cap and the breach flags, not an overwrite of every partition. If the set is larger than max_dates — two is a reasonable policy if you allow a file to cross midnight UTC — the job refuses. A corrupt timestamp that parses into hundreds of dates must not dynamic-overwrite hundreds of partitions. The collect of distinct dates is small only when the data is sane. A refuse cap on the count protects the driver and the table. You pay one job to compute distinct dates. Do it before write.

The write returning means the commit finished

DataFrameWriter returns after the job succeeds, which means the output committer finished. On HDFS that is the rename of attempt files into the partition and a _SUCCESS marker. On an object store it is whatever lesson 25 required you to configure, not the default rename. The driver treats a returned write as "the partitions we listed are the new bytes." It then writes the checkpoint. If the write throws, the checkpoint is untouched, and readers who honor the checkpoint still see the previous success. If the process dies after the write returns and before the checkpoint, the next run must be willing to replace those same dates again. Dynamic overwrite makes that rerun safe. Append would not.

_SUCCESS is not the checkpoint. It does not contain the input hash. A reader that globs Parquet and ignores the checkpoint will see files as soon as the committer places them, which may be before the checkpoint if another process races. The operational reader for completion-v3 waits until the checkpoint names that input and those dates. The landing does not call the warehouse. The scheduler orders the tasks: landing application, then SQL, and the SQL task checks the checkpoint. Data Pipeline Orchestration Fundamentals is where that ordering becomes a DAG. This lesson only refuses to make _SUCCESS do two jobs.

Quarantine, cap, and breach

Ordinary rejects — malformed JSON, missing fields, bad types, timestamps without offsets — are written as reason codes and line-sized metadata you are willing to retain, under a directory outside the dataset, keyed by the input hash so a rerun replaces the quarantine rather than appending. The count is an action. If it exceeds max_rejected_lines, the driver returns before the publish write and before the quarantine becomes the only copy of a bad day you needed to see. Write the quarantine on refusal too, so an operator can see why, and still do not move the checkpoint. A refusal leaves the previous publish in place.

A file breach — prohibited key, unknown key, or two canonical hashes — skips the publish write even if the reject count is under the cap. The cap is not a vote that outranks the contract. Conflict ids are not dumped into the application log in bulk. A count and, if you need a handle, a single id in a restricted log line are enough. The quarantine file for a breach can contain the reason code prohibited_field without the email value. You do not need the address to know the exporter broke the contract.

File counts are a write-time decision

A task writes a part file. If you partitionBy("event_date") on a DataFrame that still has one partition per input split, a large input becomes a large number of small files in the date directory, which is the listing problem from lesson 25 in miniature. repartition(n) before the write shuffles to n files when there is a single date. Choose n so each file lands near the size lesson 11 asked for, on the order of 128 MB to a few hundred, using the measured Parquet size of a sample day. The shuffle of the write is the full accepted row, which is wider than the collision shuffle. It is still the right place to set file count. Do not coalesce(1) to make the directory pretty. One file restores the single-threaded read for the next job.

repartition(n) when the DataFrame holds two dates does not mean n files per date. It means n tasks, and each task writes into the dates it happens to hold, so you can get up to n files per date and often fewer, unevenly. For the one-date case, which is the policy max_dates is protecting, you get n files. If you allow two dates, accept a less even split or repartition by a salted key. Document the choice. Evenness matters less than "not fifty thousand files" and "not one file."

Configuration, identity, and skip

The input identity is the SHA-256 of the bytes, computed by a streaming read on the driver, the same definition as lesson 14. The Spark read parses. The hash read does not parse. Two reads are the cost of an identity that does not depend on executor scheduling. A skip happens when the checkpoint for this product and input name records the same hash and the output partitions named in that record are still present. Presence is a filesystem check the driver performs, not a feeling that the job "usually works." If the hash matches and a partition directory is missing, the job republishes. If the hash differs, the job runs and replaces the dates it owns.

The checkpoint document stores product, input name, hash, accepted count, rejected count, the date list, and the publish root. It is JSON written with a temp file and os.replace on the driver's local view of the checkpoint directory, which may be HDFS through a client that does not give you POSIX replace. If the checkpoint store is HDFS or an object store, use that store's documented atomic create — a write to a temp name and a rename on HDFS, or a conditional write on the object store — and do not pretend os.replace works on a FUSE mount you have not tested. The lesson 14 rule survives the change of primitive: readers of the checkpoint see the old document or the new one, and they see it only after the data commit.

Logs and secrets

Log the product, the input name, the hash, the dates, the counts, the action (published, skipped, refused), and the application id. Log the refusal reason code. Do not log the JSON line, the email that triggered a breach, or the learner id list. Spark's event log will record the plan. Keep literal ids out of the SQL text, as lesson 26 said. Credentials for the filesystem come from the execution environment. They are not fields on BatchConfig that a __repr__ would print. The dataclass __repr__ of a frozen config full of paths is useful. A dataclass full of keys is an incident when a failure handler prints it.

Architecture and Design Diagrams

The driver sequences jobs. Executors see only the plans. The checkpoint sits off to the side, written last.

mermaidCopy
flowchart TD
    hash[Stream hash on the driver]
    skip{Checkpoint hash matches and partitions exist?}
    plan[Build classify plan]
    breach{File breach or reject cap or too many dates?}
    q[Write quarantine outside the dataset]
    pub[Dynamic overwrite of those event_dates]
    mark[Write checkpoint]
    sql[Later SQL task reads checkpoint then Parquet]

    hash --> skip
    skip -->|yes| mark
    skip -->|no| plan --> breach
    breach -->|yes| q
    breach -->|no| q --> pub --> mark
    mark --> sql

A refusal still writes quarantine and does not publish. The previous Parquet remains. The arrow from breach to quarantine is the operator's visibility. There is no arrow from breach to pub.

Dates in the input bound the overwrite. Other dates in the dataset are not in the write.

mermaidCopy
flowchart LR
    input[Closed JSONL complete for its dates]
    dates[Distinct event_date values]
    partA[event_date=2026-03-02 replaced]
    other[event_date=2026-03-01 untouched]
    input --> dates --> partA
    other -.->|not in this DataFrame| other

Static mode would have drawn a delete over both partitions. Dynamic mode draws a delete only over partA, and only if this input is the full replacement for that date.

Technical Explanation

What dynamic mode actually deletes

For a partitioned write, static overwrite removes existing data under the root before writing. Dynamic overwrite removes existing data only under partition directories that this write will recreate. Unpartitioned overwrite of a single file path is a different operation and is not what this job does. The job always partitionBy("event_date") and always sets the mode explicitly in the session so a cluster default cannot surprise it:

pythonCopy
spark.conf.set("spark.sql.sources.partitionOverwriteMode", "dynamic")

Set it on the session you write with, not in a notebook you once ran. Confirm in the submitted application's config. A review that reads the source and not the submit flags can miss a cluster default of static. Log the resolved value at startup. spark.conf.get("spark.sql.sources.partitionOverwriteMode") is enough.

Empty accepted set: do not call overwrite "to clear the day" unless the product owner agrees that an empty closed file means an empty date. An empty overwrite in dynamic mode removes the partitions you name. If you name them by writing an empty DataFrame with an explicit partition, you can delete a good day with a bad empty input. The job treats zero accepted rows and zero breaches and zero rejects as a publish of an empty partition only when configuration publish_empty is true. The default is false: refuse, quarantine nothing, leave the previous day, and log refused with reason empty_input. Silence plus delete is worse than a stale day.

Hash and Spark read are different passes

The hash function streams bytes. It does not import PySpark. Tests of the hash do not start a JVM. The parse is the Spark read of the same path. If the file changes between the hash and the parse, the checkpoint can name bytes you did not publish. The producer contract is a closed file that is not rewritten in place. If you cannot get that promise, hash after a copy to an immutable inbox key that the producer cannot open. The landing reads only the immutable key. That copy is an ingestion concern from lesson 15. This job documents the assumption: the path is stable for the life of the application.

Committers

The write uses Spark's file source and the Hadoop output committer configured for the scheme. For hdfs://, the default rename committer matches lesson 25. For s3a://, set the committer the Hadoop AWS document describes for your versions, in the same submit configuration as deploy mode. This lesson does not embed a committer class name that might be wrong for your patch. It embeds the requirement: a killed job leaves readers, who obey the checkpoint, on the previous partition, and a listing of the date does not mix two generations. Prove that on the target scheme with a small file before enrollment week. A local[2] test proves the driver order. It writes the local filesystem, where rename is real. It does not prove S3A.

Idempotent rerun

Same hash, partitions present: log skipped, do not start a Spark job, exit zero. Same hash, partition missing: run the write again. New hash: run the write, replace dates, replace checkpoint. Overlapping applications: still forbidden. Spark will not lock the path. The scheduler must. A skipped run must not return success if the checkpoint is corrupt JSON. Fail the application so the scheduler pages a human. A skip that swallows a parse error in the checkpoint will serve a stale mart and look healthy.

Tests without a cluster

Use one SparkSession for the module, local[2], UI off, dynamic overwrite on, ANSI on, UTC. Build fixtures with tmp_path. Assert:

• Identical retries publish one row and a checkpoint whose hash matches the file.
• A second run with the same file logs skip and does not change the checkpoint timestamp if you store one, or does not rewrite Parquet mtime if you can read it. Easiest assertion: the function returns action skipped.
• An email field returns refused, writes no Parquet under the date, and leaves an existing good partition byte-for-byte. Seed a good partition first, then refuse a bad file aimed at the same date, but only if that bad file would have been a full replace. Because you refuse before write, the seed remains. That is the assertion.
• Two hashes refuse.
• Static mode is not used. A regression test can set static mode in a separate session and show that a write of one date deletes a second date you planted, which is the motivation for the production session's explicit dynamic setting. Keep that test obviously labeled so nobody copies it into main.
• Reject cap: three bad lines and a cap of two refuses and does not publish the good lines of that file.

These tests are the landing. The cluster run is a scale and commit test, not a place to discover that email publishes.

Practical Implementation or Design Walkthrough

Problem

Wire lesson 27's transforms into a run function that publishes Parquet, quarantines reasons, and checkpoints last. Exercise it on a local filesystem fixture. State the submit-time settings that differ in production.

Configuration

pythonCopy
from __future__ import annotations

import hashlib
import json
import os
import tempfile
from dataclasses import asdict, dataclass
from pathlib import Path

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


@dataclass(frozen=True)
class BatchConfig:
    product: str
    input_path: Path
    publish_root: Path
    quarantine_root: Path
    checkpoint_path: Path
    max_rejected_lines: int
    max_dates: int
    files_per_date: int
    publish_empty: bool = False

Paths are local in tests. In production they may be HDFS URIs stored as strings. If you need URIs, use str in the config and keep Path only in the local test. Do not pass a Path("s3a://...") and expect os.replace to mean something. The checkpoint writer below is the local and same-filesystem version. The production note beside it says when to switch primitives.

Hash and checkpoint primitive

pythonCopy
def sha256_file(path: Path) -> str:
    digest = hashlib.sha256()
    with path.open("rb") as handle:
        for chunk in iter(lambda: handle.read(1024 * 1024), b""):
            digest.update(chunk)
    return digest.hexdigest()


def atomic_json(path: Path, payload: dict) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    descriptor, temporary_name = tempfile.mkstemp(dir=path.parent, prefix=".ckpt.", suffix=".tmp")
    temporary = Path(temporary_name)
    try:
        with os.fdopen(descriptor, "w", encoding="utf-8") as handle:
            json.dump(payload, handle, sort_keys=True)
            handle.write("\n")
            handle.flush()
            os.fsync(handle.fileno())
        os.replace(temporary, path)
    except Exception:
        temporary.unlink(missing_ok=True)
        raise

The run order

classify_lines, conflict_keys, and accepted_events are imported from the lesson 27 module. file_breach is coalesced so a malformed line is not a null flag.

pythonCopy
def run(spark: SparkSession, config: BatchConfig) -> str:
    spark.conf.set("spark.sql.sources.partitionOverwriteMode", "dynamic")
    digest = sha256_file(config.input_path)
    previous = _read_checkpoint(config.checkpoint_path)
    if (
        previous
        and previous.get("input_sha256") == digest
        and _partitions_exist(config.publish_root, previous.get("event_dates", []))
    ):
        return "skipped"

    lines = spark.read.text(str(config.input_path))
    classified = classify_lines(lines).withColumn(
        "file_breach", F.coalesce(F.col("file_breach"), F.lit(False))
    )
    if classified.where(F.col("file_breach")).limit(1).count():
        _write_quarantine(classified, config, digest)
        return "refused"

    conflicts = conflict_keys(classified)
    if conflicts.limit(1).count():
        _write_quarantine(classified, config, digest)
        return "refused"

    rejected = classified.where(F.col("reject_reason").isNotNull()).count()
    if rejected > config.max_rejected_lines:
        _write_quarantine(classified, config, digest)
        return "refused"

    accepted = accepted_events(classified).where(F.col("event_date").isNotNull())
    dates = [row["event_date"] for row in accepted.select("event_date").distinct().collect()]
    if len(dates) > config.max_dates:
        _write_quarantine(classified, config, digest)
        return "refused"
    if not dates and not config.publish_empty:
        return "refused"

    _write_quarantine(classified, config, digest)
    if dates:
        (
            accepted.repartition(config.files_per_date)
            .write.mode("overwrite")
            .partitionBy("event_date")
            .parquet(str(config.publish_root))
        )
    atomic_json(
        config.checkpoint_path,
        {
            "product": config.product,
            "input_name": config.input_path.name,
            "input_sha256": digest,
            "accepted_rows": accepted.count(),
            "rejected_rows": rejected,
            "event_dates": dates,
        },
    )
    return "published"

Read this order carefully, because it is the design. Breach and conflict and cap all return before write.partitionBy. Quarantine writes happen on those paths and do not move the checkpoint. The publish write happens, and only then atomic_json. accepted.count() after the write is a second action on accepted. That recomputes unless you cached accepted. On a large day, cache the narrow accepted DataFrame before the write, unpersist after the count, or count from the Parquet you just wrote with a scan of only those dates. A second full parse is the kind of serial surprise lesson 26 warned about. The fixture is small, so the skeleton is clear. The production patch is:

pythonCopy
accepted = accepted.persist()
try:
    # write, then accepted.count()
    ...
finally:
    accepted.unpersist()

persist without a measured reuse is still a cost. Here the reuse is real: write and count. Cache that. Do not cache lines.

_write_quarantine writes reject_reason and event_id where the reason is not null, mode overwrite, to quarantine_root / digest. It does not write the raw line. A prohibited row can record event_id and prohibited_field if the id parsed. If the line was so broken there is no id, record the reason alone. _partitions_exist lists each event_date=... directory and returns false if any is missing. _read_checkpoint returns None when the file is absent, and raises when the file is present and not valid JSON.

There is a subtle bug to avoid when you copy this skeleton: accepted.count() after a refusal is not reached, which is correct. dates is collected before the write, which is correct. If repartition changes row counts, you have a Spark bug, not a landing bug. Counts must match.

Submit configuration

Local tests call run(session, config). Production uses cluster deploy mode, the landing principal, an event log, and an explicit overwrite mode even though run sets it again. The belt and braces is intentional. A future edit that removes the line in run should still be safe on the submit. Executor sizing comes from the lesson 24 measurement, not from this chapter. Shuffle partitions for the collision aggregate are set for the hash volume. files_per_date is the write repartition, independent of spark.sql.shuffle.partitions, though both are shuffles and both show up in the UI.

textCopy
spark-submit \
  --master yarn \
  --deploy-mode cluster \
  --name tlp-progress-landing \
  --conf spark.sql.sources.partitionOverwriteMode=dynamic \
  --conf spark.sql.shuffle.partitions=64 \
  --conf spark.sql.session.timeZone=UTC \
  --conf spark.sql.ansi.enabled=true \
  --conf spark.eventLog.enabled=true \
  progress_landing.py --config /etc/tlp/progress-landing.json

The JSON config is paths and caps. It is not the event file. Arguments select Tuesday versus Wednesday. The image or the conda environment pins pyspark to the 3.5 line you tested. An unpinned "latest" on the cluster is how ANSI mode and overwrite defaults drift.

A local assertion sketch

pythonCopy
def test_email_does_not_replace_a_good_day(spark, tmp_path: Path) -> None:
    good = tmp_path / "good.jsonl"
    good.write_text(
        '{"event_id":"e1","learner_id":"l1","lesson_id":"les1",'
        '"event_type":"completed","attempt_number":1,'
        '"event_time":"2026-03-02T15:00:00Z"}\n',
        encoding="utf-8",
    )
    publish = tmp_path / "publish"
    checkpoint = tmp_path / "ckpt.json"
    config = BatchConfig(
        product="progress",
        input_path=good,
        publish_root=publish,
        quarantine_root=tmp_path / "quar",
        checkpoint_path=checkpoint,
        max_rejected_lines=10,
        max_dates=2,
        files_per_date=1,
    )
    assert run(spark, config) == "published"
    bad = tmp_path / "bad.jsonl"
    bad.write_text(
        '{"event_id":"e2","email":"a@example.com","learner_id":"l1",'
        '"lesson_id":"les1","event_type":"completed","attempt_number":1,'
        '"event_time":"2026-03-02T16:00:00Z"}\n',
        encoding="utf-8",
    )
    refused = BatchConfig(
        product="progress",
        input_path=bad,
        publish_root=publish,
        quarantine_root=tmp_path / "quar",
        checkpoint_path=checkpoint,
        max_rejected_lines=10,
        max_dates=2,
        files_per_date=1,
    )
    assert run(spark, refused) == "refused"
    saved = spark.read.parquet(str(publish)).select("event_id").collect()
    assert [row["event_id"] for row in saved] == ["e1"]
    assert json.loads(checkpoint.read_text())["input_sha256"] == sha256_file(good)

This is the regression that matters. The bad file names the same date the good file published. A write that forgot to return early would replace e1 with nothing or with a row that failed the key check. The assertion reads Parquet and the checkpoint hash. It does not trust the return string alone. The return string can be "refused" while a bug still wrote. Assert the files.

Use files_per_date=1 only in the test, so the read is simple. Production configuration uses the sized count. A test that sets production file counts does not make the contract safer.

Enterprise Real-World Scenarios

Static overwrite of the quarter

The first scheduled run of this job used mode("overwrite") and did not set the partition overwrite mode. The default is static. The fixture in CI wrote into an empty directory, so the tests passed. The first production rerun was a single closed day in March, aimed at a root that already held the quarter restatement from the backfill. The job succeeded. The checkpoint recorded March 2. The root contained March 2 and nothing else. completion-v3 for January and February disappeared between one scheduler interval and the next. The event log showed a successful write. The success was the incident.

The repair was the explicit dynamic setting in both run and spark-submit, a test that plants two dates and rewrites one, and a checkpoint schema that lists event_dates so an operator can see the intended scope. The lost quarter was restored from the backfill inputs, which had been kept, and republished date by date. The restore was possible because the input files were immutable. The Parquet had been the only copy of the publish, not of the source. That is an argument for retaining closed inputs for the restatement window, which is a storage cost lesson 24 already put in the lens. The squad wrote the default down in the run book in one sentence: overwrite mode static deletes the dataset root.

No learner payload was added to the ticket. The ticket cited dates and counts. The email incident of lesson 27 stayed a separate test, and it would not have caught this. Contract fixtures and filesystem-mode fixtures are both required. One of them is not a substitute for the other.

Checkpoint first, because the write "usually works"

A later revision wrote the checkpoint before write.parquet so the SQL task could start sooner. The write threw after the committer had removed the old partition and before the new one was visible, on a full disk on the executor side. The checkpoint said the new hash was published. The SQL task checked the checkpoint, found it fresh, and scanned a missing or partial date. The morning number dropped. The skip logic then made the next landing a no-op, because the hash matched and the existence check had been implemented as "checkpoint file exists," not "each date directory exists."

The repair restored the order: write, then checkpoint. The existence check lists every date in the checkpoint. A drill kills the driver after a successful local write and before atomic_json, reruns, and expects action published and one copy of the rows, not two. The drill is in the run book for enrollment week. The SQL task was changed to require both the checkpoint and a non-empty partition, and to fail rather than publish a zero. A zero that is a real empty day is the publish_empty flag, which remains false. Starting SQL early is an orchestration wish. It does not outrank the crash window.

Best Practices

• Keep transforms free of paths and environment variables. Pass a BatchConfig into run.
• Set dynamic partition overwrite in code and in the submit configuration. Log the resolved value.
• Collect distinct event_date values and refuse if the count exceeds max_dates, before any overwrite.
• Treat the input as a full replacement for those dates. Do not use this job to apply a fragment of late events.
• Return before write on breach, conflict, cap, or unexpected empty input. Assert in tests that seeded Parquet survives a refusal.
• Write quarantine outside the dataset, without raw payloads, overwrite by input hash.
• persist the accepted DataFrame only across the write and the count, then unpersist.
• Checkpoint after the write returns, with an atomic primitive the filesystem actually provides. Include the hash and the date list.
• Skip only when the hash matches and every listed partition exists.
• One application per input, cluster deploy mode, scheduler-enforced. Pin the PySpark version.
• Log action, hash, dates, and counts. Do not log lines.
• Leave completion-v3 to the next task. Pass it a checkpoint it can trust.
• Size files_per_date from measured Parquet volume. Use 1 only in tests.

Common Mistakes and Anti-Patterns

Static overwrite on a shared root. A one-day rerun deletes the quarter. Tests on an empty directory do not catch it. The repair is dynamic mode and a two-date regression test.

Late-file overwrite. A file of corrections for a handful of learners replaces the entire date with that handful. The repair is a full restatement of the date, or a designed merge with the collision key. This job's write is a replace. Call it that in the operator instructions.

Checkpoint before write, or skip without looking at the directories. The next run does nothing while the mart reads a hole. The repair is the crash order from lesson 14 and an existence check per date.

Quarantine inside the date directory. The next Parquet read fails on a JSON file, or a glob ingests it. The repair is a separate root and a schema on the publish path that only reads Parquet.

collect of accepted rows to write them with the Python open from lesson 14. The driver dies, and you have rebuilt the single process inside the expensive application. The repair is DataFrameWriter. The Python file primitive remains appropriate for the small checkpoint.

publish of breaches that "only failed one column." Filtering email out and writing the struct publishes the rest of a bad extract. The repair is the early return. The test seeds a good day and shows it still there.

files_per_date left at the input split count. Tens of thousands of part files make the SQL scan slower than the landing. The repair is an explicit repartition sized from bytes, reviewed after the first real day in the UI.

Unpinned Spark on the cluster. ANSI mode, overwrite defaults, and json_object_keys behavior are versioned. The repair is the pin you tested, including the Hadoop committer jars for the object store.

Two configs, one in a notebook and one in submit. The notebook was dynamic. The scheduler was static. The repair is one config object and a logged resolved setting at the start of run.

Troubleshooting and Design Review

Symptom: A rerun of one day removes other dates. The job status is success.

Likely cause: static partition overwrite, or a write that was not partitioned and replaced the whole directory.

Investigation: the logged overwrite mode, explain is less useful here than the writer call and spark.conf.get, and the checkpoint's event_dates versus the directories that remain.

Corrective action: restore from closed inputs, switch to dynamic mode, add the two-date test, and rerun the missing dates one complete input at a time.

Symptom: The job returns refused and the good partition is gone anyway.

Likely cause: the write is not actually after the return, or an exception handler writes, or another application overlapped.

Investigation: application ids in that hour, and the line order in run. A stack trace that includes DataFrameWriter on a refusal path is the bug.

Corrective action: restore the partition, fix the order, and keep the seeded-Parquet test.

Symptom: Action skipped and the date directory is missing.

Likely cause: the existence check was removed, or it checks the root and not the date, or trash/retention deleted the directory after the checkpoint.

Investigation: checkpoint JSON and a list of event_date= paths.

Corrective action: republish when any listed date is absent. Fix the check. Do not mark skip because the checkpoint file exists.

Symptom: The morning scan is slow and the date directory has tens of thousands of small files.

Likely cause: no repartition before partitionBy, on a finely split input.

Investigation: file sizes under the date. UI task count on the write stage.

Corrective action: set files_per_date from the measured output size and rewrite the date. Confirm the next SQL stage's task count dropped.

Symptom: Rejected rows appear inside the published Parquet.

Likely cause: accepted_events was not the DataFrame written, or a filter on reject_reason was dropped.

Investigation: a row count of null event_id or a sample schema. The quarantine directory should hold those reasons. If the publish contains them, the select list is wrong.

Corrective action: rewrite from the input. Assert in CI that every published row has a non-null event_id and a parseable event_time.

Symptom: Disk fills on executors during overwrite and the partition is missing.

Likely cause: the committer materialized a second copy, or dynamic overwrite still writes new files before deleting old ones, so peak disk is old plus new.

Investigation: executor disk metrics and the filesystem. Object-store copies from a rename-style committer make this worse.

Corrective action: free space, rerun with the documented committer, and size disks for two copies of a date during commit. Do not switch to append to save disk. Append saves the disk and spends the metric.

A design review asks: What is deleted if this input contains one date? What is deleted if it contains forty? What is left if we kill the driver after write and before checkpoint? Where is quarantine? What does the skip check list? What is files_per_date and what did the first real day actually write? Which test plants a good day and sends email next? The review is the run function on one page, plus that test. It is not a new diagram of executors.

Interview Questions and Answers

Q1. Why is a DataFrame transform not yet a landing job?

A transform has no input identity, no commit, and no record of success. I can recompute it twice and publish twice if the writer appends. The landing adds configuration, a hash, a refusal path that does not write the dataset, a dynamic partition replace, and a checkpoint after the write returns. The expressions from the DataFrame lesson stay pure so tests can call them without HDFS. main is the only place that sequences actions.

Q2. What does dynamic partition overwrite replace, and what does it leave alone?

It replaces partition directories that are present in the DataFrame and leaves other event_date directories in place. Static mode, the default, replaces the whole root. I set dynamic mode in the session and in spark-submit, and I log it. I also refuse an input whose distinct date count exceeds max_dates, because a bad timestamp must not be a license to replace a year of partitions. The input must be the complete data for the dates it does replace. A fragment overwrites the date with the fragment.

Q3. Where do you put the checkpoint relative to the Parquet write?

After write returns without throwing, and not inside the dataset directory. The write's return means the committer finished. The checkpoint records the hash, the counts, and the dates. If I checkpoint first, a failed write still looks published and the next run skips. If I die between the two, the next run replaces the same dates and writes the checkpoint. That rerun is safe only because the write is a replace. I test the refusal path by seeding a good day and proving a bad file does not change it.

Q4. What do you collect to the driver?

Distinct event_date strings, counts, and a one-row existence check for breaches and conflicts. I persist the accepted DataFrame across the write and the count so I do not parse the JSON twice for that pair, and I unpersist afterward. I do not collect events, and I do not write Parquet from a Python loop over rows. The hash is a streaming read of bytes on the driver, which is serial and is not a collect of parsed rows.

Q5. How do you keep a rerun of the same file from being a second copy?

The checkpoint stores the SHA-256. If it matches and every listed partition exists, I return skipped and I do not start the Spark jobs. If the hash matches and a partition is gone, I publish again. If the hash differs, I replace the dates in the new file. Append mode is not used. Two overlapping applications are prevented by the scheduler, because nothing in DataFrameWriter locks the path against another driver.

Q6. Why is files_per_date a configuration value?

Each write task produces part files. Left alone, the input split count becomes the file count, which on a compacted day might be fine and on a raw small-file tree is a disaster for the next scan. I repartition to a chosen n for the one-date case and I pick n from measured Parquet size so files sit near a block-sized read. Tests use 1 for simplicity and do not pretend that is the production setting. spark.sql.shuffle.partitions does not set this. Both are shuffles. They are different knobs.

Q7. What is in the quarantine, and where does it live?

Reason codes and identifiers I am willing to retain, outside the publish root, overwritten per input hash. Not the raw JSON, and not the email value. A refusal still writes quarantine so the operator can see prohibited_field or collision, and it does not move the checkpoint. If the reject count exceeds the cap, I quarantine and refuse rather than publish a partial good set. Publishing the good lines of a file that exceeded the cap hides how broken the extract was.

Q8. How do you test overwrite behavior without a cluster?

A local session writes real Parquet to a temporary directory. I plant two dates, write one date in dynamic mode, and assert the other remains. I plant a good day, run a file with email, and assert the Parquet and the checkpoint hash still describe the good file. I set static mode only in a test that documents the footgun. local[2] does not prove the S3A committer. I say that next to the test so a green CI run is not described as an object-store proof.

Q9. What does _SUCCESS mean here?

The Hadoop committer finished the output. It does not contain the input hash, and it is not what completion-v3 should trust if a glob would read partial files anyway. The checkpoint is the business record. The SQL task runs after the landing application and checks the checkpoint, then reads Parquet. I do not start SQL in parallel with the write to save minutes. Those minutes are the crash window.

Q10. A closed file contains events from 23:30 UTC and 00:30 UTC. What happens?

Those are two event_date values. If max_dates allows it, dynamic overwrite replaces both dates with whatever subset of those dates is in this file. If the file is not the full story of both dates, I have deleted events that arrived elsewhere. I set max_dates to the producer's real contract, usually one, and I treat a midnight span as a producer decision that must be explicit. I do not let every file quietly rewrite two days because a few events crossed the boundary. The partition column is the event instant's date, not the file name.

Q11. Why persist, and why not persist the raw text?

The accepted projection is used by the write and by the count. Without persist, that is two parses and two collision shuffles if both actions sit on the same lineage. Persist saves the reuse. The raw text is large, used once, and caching it competes with shuffle memory. I persist the small side of the plan, the accepted rows I am about to write, and I unpersist in a finally so a failed write does not pin the block in a long-lived session. The scheduled application exits, which also releases memory. A shared notebook session does not. Production is not a shared notebook.

Q12. What would you show an auditor who asks how a day is reproduced?

The closed input, still retained, and its hash. The checkpoint naming that hash and the dates. The job version pin. The refusal tests for email and for two payloads. A rerun of that input replaces the same dates and yields the same accepted count. I would not show a dashboard screenshot as the reproduction. I would not show _SUCCESS alone. If the input has been deleted, I cannot reproduce the day, and the retention policy is the audit gap. Spark does not remember the file for me.

Hands-On Exercises

Exercise: Order the actions so a bad file cannot replace a good day

Problem statement. Implement run for the progress landing on local Parquet. The good day must survive an emailed file and a static-mode footgun must be visible in a test you do not ship.

Tasks.

1. Define BatchConfig with the cap, max_dates, and files_per_date.
2. Hash the input with a streaming SHA-256. Checkpoint after a successful write, atomically, outside the dataset root.
3. Call the lesson 27 transforms. Refuse on breach, conflict, or cap before write.
4. Use dynamic overwrite. Collect dates and refuse if there are too many.
5. Seed a published day, refuse a bad file for that date, and assert the original event_id and hash remain.
6. Show that a same-hash second run skips, and that a missing partition does not skip.
7. Write one test that demonstrates static mode deleting an unrelated date. Keep it out of main.
8. State what local[2] did not prove about object-store committers, and what the driver is allowed to collect.

Expected deliverables. The job module, the tests, and a short run-book note on replace versus late fragments.

Evaluation rubric.

You pass when a reviewer can kill the process in their head between write and atomic_json and predict a safe rerun, and can say what static mode would have done to January.

Lesson Summary

A PySpark landing is the lesson 14 contract on a Spark plan: hash the closed input, classify with the DataFrame expressions, refuse without writing when the contract breaks, otherwise dynamically overwrite only the event dates that input fully replaces, and checkpoint after that write. Static overwrite and a checkpoint written first are the two failures that make a green application worse than a stale mart. Quarantine lives outside the dataset. File counts are chosen. Tests seed a good day and try to destroy it with email and with a second date. completion-v3 reads the result. It is not computed here.

Spark RDDs and Execution Model opens the lineage under this plan. The job you submit stays a DataFrame job. The RDD lesson is how you see what the engine retries, and when a lower-level API would make this landing worse.

Further Reading

• Spark SQL data source options and Parquet — partitioned writes. Pair with the configuration page entry for spark.sql.sources.partitionOverwriteMode.
• Spark configuration — overwrite mode, shuffle partitions, ANSI mode. The submit snippet in this lesson sets the ones the landing depends on.
• Hadoop AWS committers — required reading before the publish root is an s3a:// bucket. Local tests do not replace this page.
• Spark submitting applications — cluster deploy mode for the scheduled run.
• Building Batch Data Pipelines with Python — the checkpoint-last order this job keeps.
• Continue this course: Spark RDDs and Execution Model, then Spark Joins and Data Shuffling.