Building Batch Data Pipelines with Python

Data Engineering & Modern Data Platforms · Lesson 14 · Data Ingestion, ETL and Orchestration

Lesson Overview

The placement decision in ETL and ELT Architecture said TechLearningPro's progress events should land as a projected raw partition, and that completion-v3 should be SQL over that partition. The first Python job that tried to do the landing was a script on a laptop: open the day's JSONL, append to a file in the raw prefix, print "done." On a Wednesday retry, after a disk-full error halfway through Tuesday, the script appended the same events again. The checkpoint was the word "done" in a terminal scrollback. The curated replace from lesson 13 then counted two copies until someone noticed the grain had been violated upstream of the primary key, in a file the SQL job trusted.

A batch pipeline is a program that can be run again. Configuration says which window and which paths, without editing source. Validation decides which lines are facts and which are breaches. Logging makes a run reconstructable without opening the data file. A checkpoint records a successful input identity only after the output is durable. Errors are a quarantine with a reason, or a failed run that leaves the previous output in place. The scheduler, lesson 19, can call this program every night. The program does not become correct because a scheduler called it.

This lesson builds that program for one product: the daily progress landing that lesson 12 chose as an incremental file and lesson 13 placed at the boundary. The standard library is enough. Lessons 4 and 5 and /learn/python are the language. This lesson is the run contract. It does not extract from PostgreSQL, does not call a cloud SDK, and does not schedule itself. Fifty million events on a peak day, the standing planning assumption, will not be parsed as one giant list. The job streams line by line. If a measured run still misses 06:00 UTC, that measurement belongs to Module 3, not to a bigger list.

Learning Objectives

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

1. Separate a batch run's configuration, input identity, validation, durable publish, and checkpoint so a retry is safe.
2. Stream a JSONL progress file without holding the day in memory, and reject a prohibited field without publishing the file.
3. Advance a checkpoint only after an atomic replace of the output, and skip a rerun whose input hash is unchanged.
4. Quarantine malformed lines with a reason, and refuse to treat the quarantine as a successful metric.
5. Log a run so an operator can see input hash, accepted rows, rejected rows, and whether the checkpoint moved.
6. Explain what a crash at each stage leaves on disk, and how the next run recovers.

Prerequisites

• Data Integration Patterns: partition replace and a completeness marker that moves after durability.
• ETL and ELT Architecture: this job is the boundary load into raw, not the completion-v3 SQL.
• Lessons 4 and 5: dataclasses, type hints, streaming reads, and the habit of not building a list of every event before you decide. /learn/python if the syntax is rusty.
• Lesson 1's event contract: event_id, learner_id, lesson_id, event_type, attempt_number, event_time. Email, name, and payment tokens fail the batch.
• A command line and Python 3.11 or newer. No third-party packages. /learn/sql is not required to run the demo. The production checkpoint table is shown as SQL you can read.

Detailed Core Concepts

A run is an identity plus a result

A batch run names an input and produces an output or a refusal. The input identity this lesson uses is the file name plus the SHA-256 of the bytes. The name alone is not an identity. A producer can overwrite progress-2026-03-02.jsonl with a repaired file. The hash changes. The job must run again. The same hash means the bytes did not change, and a second run should not rewrite for sport and should not append.

The output identity is a published path the downstream SQL is allowed to read, plus a checkpoint row that points at the input hash. Downstream reads the published path only. It does not read a temporary file, and it does not infer success from a process exit it did not see. The checkpoint is the job's memory. The published file is the data product's input. Both exist, or the run did not succeed.

Configuration is data

Paths, the product name, and the definition of "too many rejects" belong in a configuration object loaded from the environment or from an argument, not in constants scattered through the function. A retry of Tuesday points at Tuesday's file. Changing the code to do that is how Tuesday's logic and Wednesday's logic diverge. The walkthrough uses a frozen dataclass. A production job can build that dataclass from environment variables. The processing functions accept the dataclass. They do not call os.environ themselves, which keeps tests honest.

Secrets are not configuration you log. This job has no database password. The moment it does, the log line that prints the configuration object is an incident. Log paths and hashes. Do not log payloads. A progress event is enough to reconstruct a learner's evening.

Validation is part of the run, not a later query

Each line is one JSON object. The contract fields are required. attempt_number is an integer, not a boolean and not a string. event_time is an ISO-8601 timestamp with an offset, normalized to UTC in the output so two writers of Z and +00:00 do not look like two times. event_type is checkpoint or completed. A prohibited key fails the file, not the line. One email means the producer joined a profile "for convenience." Quarantining that line and publishing the rest trains the producer to keep sending it. The file stays unpublished. The previous day's published partition remains the latest success.

Malformed JSON, a missing field, or a bad type quarantines the line. Identical event_id values with identical payloads collapse to one accepted event. The same event_id with a different payload fails the file: that is the collision lesson 1 refused to resolve with "latest wins." A quarantine file is written for the run so an operator can see the reasons. The checkpoint records rejected counts. The curated SQL does not read the quarantine. A green count of accepted rows with an unexplained reject pile is not success if you also set a threshold. This job fails the file when any prohibited field or identity collision appears, and it publishes when ordinary rejects stay under a configured absolute cap. Lesson 17 generalizes thresholds. Here the cap is part of configuration so "we logged it" cannot hide a file that is half garbage.

Checkpoints move last

The dangerous orders are easy to write.

1. Append to the published file, then crash. The next run appends again.
2. Write the checkpoint, then crash while publishing. The next run skips, and the output is missing.
3. Publish, then crash before the checkpoint. The next run repeats the work. If publish is a replace, the repeat is safe. If publish is an append, the repeat doubles.

The safe order for this lesson: write the accepted lines to a temporary file in the same directory, write the quarantine the same way, replace both into their published names with os.replace, then write the checkpoint the same way. os.replace is atomic on the same filesystem. A reader sees the old published file or the new one, not a torn half. A crash before the checkpoint causes a repeat, and the repeat replaces the same bytes. A crash before replace leaves a temp file that the next run ignores and overwrites. The checkpoint is never ahead of the bytes.

Skipping when the hash matches avoids needless work and, more importantly, avoids a second publish that could race a reader if you had used append. Skip only when the checkpoint hash matches and the published output exists. A checkpoint that points at a deleted file is a lie. The run republishes.

Logs are for the operator, not for the metric

Each file logs the product, the input name, the hash, the accepted count, the rejected count, and the action: published, skipped, or refused. Logs go to stderr. stdout stays available for a caller that wants a machine-readable summary. A log line is not a checkpoint. Shipping logs to a system that deletes them in three days is fine. The checkpoint has to outlive the rerun you will attempt next week.

Include a run id so two overlapping invocations can be separated in the log. Overlapping invocations are still a bug. The job does not take a lock in this lesson. Production adds a lock or a scheduler that will not start a second run of the same product and window. Document that limit. Two writers replacing the same path can leave the checkpoint describing the other writer's bytes.

What "batch" excludes

This program is one process, one machine, local files standing in for object prefixes. It is the right shape for a day of events when the parse and the write fit the host and the morning bound. It is not a stream consumer. It is not a distributed shuffle. It does not open a connection pool. The production landing may be an object store. The contract to preserve is the same: identity of the input, temp object, publish, then checkpoint. A vendor SDK does not change the order. This lesson refuses to invent that SDK. When you adopt one, you map os.replace to that vendor's documented conditional write, and you test the crash window again.

Architecture and Design Diagrams

One file moves from an inbox the producer closes, through a refuse-or-publish decision, to a raw prefix the SQL job may read. The checkpoint sits beside the data. It is not inside the data file, because a reader of events should not have to parse a footer to know the file is finished. The manifest of success is the checkpoint.

mermaidCopy
flowchart LR
    inbox[Closed JSONL in the inbox]
    hash[SHA-256 and checkpoint lookup]
    parse[Stream parse and validate]
    refuse[Refuse: no publish]
    tmp[Temp accepted and quarantine files]
    publish[Atomic replace into the raw prefix]
    mark[Checkpoint written last]
    sql[completion-v3 reads published files only]

    inbox --> hash
    hash -->|same hash and output exists| mark
    hash -->|new or missing output| parse
    parse -->|prohibited field or id collision| refuse
    parse -->|within reject cap| tmp --> publish --> mark
    mark --> sql
    refuse -.->|previous publish remains| sql

The crash diagram is the one to rehearse before you add a scheduler. Each X is a process death. The recovery column is the next invocation of the same command.

mermaidCopy
flowchart TD
    start[Start run]
    wroteTmp[Temp file written]
    replaced[Published path replaced]
    marked[Checkpoint updated]
    start -->|crash| r1[No change. Rerun does the work]
    wroteTmp -->|crash| r2[Temp ignored. Rerun replaces]
    replaced -->|crash| r3[Rerun replaces the same bytes, then marks]
    marked -->|crash| r4[Next run sees the hash and skips]

A third picture is the boundary with lesson 13. This job does not compute completion. It refuses to be the place where checkpoint is dropped.

mermaidCopy
flowchart LR
    job[This batch job]
    raw[(Published raw JSONL)]
    curated[Window replace SQL]
    job -->|keeps contract fields including checkpoint| raw --> curated
    job -.->|does not write| curated

Technical Explanation

Streaming and the peak-day assumption

Fifty million lines, at a few hundred bytes each, are tens of gigabytes. A list of parsed objects will try to hold that in RAM and will fail on a host that could have streamed the file in one pass. The walkthrough iterates lines, keeps a dictionary of event_id to payload identity for collision detection, and writes accepted lines as it goes. The dictionary is the memory cost of the contract. Fifty million short keys is a real memory number, not a free abstraction. If that map does not fit, the honest designs are a sort-and-scan for duplicate keys, or a database unique constraint on a staged table. Do not "fix" it by skipping the collision check. Record the assumption: this teaching job holds the key map, and a production review measures it. Lesson 5's warning against building a list of every event is the same warning, applied to the value map. Keys are required. Values are not all held if you hash the canonical payload and store the hash.

The implementation below stores a canonical JSON hash per event_id, not the full object. Identical retries share the hash. A collision is a different hash. Memory stays proportional to distinct ids, which on a clean day is the event count. That is still large on a peak day. Say so in the run book. A host that cannot hold the set is a measured reason to stage into a store with a primary key, which is still this checkpoint contract: stage, validate, publish, then mark.

Canonical identity for a duplicate

Two JSON objects can differ in key order and whitespace and still be the same event. Hashing the raw line makes those look like collisions and fails a file a retry wrote with a different encoder. The job parses JSON and hashes a canonical dump with sorted keys and no extra spaces, over the contract fields only. Extra unknown fields are a failure, not a thing to ignore silently, because an unknown field is how email arrives under a new name you did not list. The prohibited list is the known set. Unknown keys fail the line into quarantine, and a threshold can still fail the file. This is stricter than "ignore extra fields," and it is the right strictness for a boundary that once leaked a column through a helpful exporter.

Time normalization

2026-03-02T14:02:11Z and 2026-03-02T14:02:11+00:00 are the same instant. The published line stores event_time as UTC with a Z suffix so downstream partitions do not split one instant across two strings. Naive timestamps, with no offset, fail the line. The player clock is not invited to guess. Event time remains the business clock. This job does not decide which UTC day the completion belongs to. It preserves the instant. The SQL in lesson 13 applies the window.

Atomic replace and filesystems

os.replace is atomic when source and destination are on the same filesystem. A temporary file in /tmp and a destination on another mount is a copy, and a crash during the copy tears the destination. The walkthrough creates the temporary file in the destination directory. Object stores differ: a completed object appears, a multipart upload may not, and "rename" may be a copy plus delete. The pattern is "readers cannot observe a partial object." Map it to the store's documented complete-upload or conditional write. Do not assume POSIX rename over a FUSE mount you have not tested.

The checkpoint record

A JSON file is the demo store so the lesson runs without a database. Its shape is the row you would insert:

sqlCopy
CREATE TABLE ingestion_checkpoint (
    product text NOT NULL,
    input_name text NOT NULL,
    input_sha256 text NOT NULL,
    accepted_rows integer NOT NULL,
    rejected_rows integer NOT NULL,
    published_path text NOT NULL,
    committed_at timestamptz NOT NULL,
    PRIMARY KEY (product, input_name)
);

The commit of that row is the last step, in the same sense as the JSON replace being last. If the checkpoint lives in PostgreSQL and the file lives in object storage, you cannot have one transaction across both. The safe order is still: data durable, then checkpoint. A crash in between redoes an idempotent replace. The unsafe order is checkpoint first. There is no distributed transaction in this lesson, and inventing one would be a fictional API. The redo window is the design.

Reject caps

A cap of zero means any bad line fails the file. That is correct for a small catalog extract and harsh for a peak day where one truncated line at byte offset 40 GB should not hide fifty million good events. The cap is a product decision. This lesson's demo uses a small cap and fails closed on prohibited fields regardless of the cap. Prohibited fields are not "bad lines" in the same sense. They are a contract breach of the file. Lesson 17 will talk about rates and burn. Do not preview a statistical model here. A count and a named reason are enough.

Concurrency and the scheduler you do not have yet

Two processes can pass the hash check before either writes the checkpoint. Both publish. The last replace wins, and the checkpoint may describe the first. Until lesson 19, the operational rule is one run of a given product at a time, enforced by the human or by a lock file created with O_EXCL. A lock file is local and wrong across machines. It is an honest limit. Write it on the run book. Do not claim the job is safe under overlapping cron entries.

Practical Implementation or Design Walkthrough

Problem

Ingest one closed JSONL file of progress events into a published raw file. Retrying the same bytes skips. A crash before the checkpoint redoes a replace. An email field publishes nothing. A malformed line is quarantined. A changed event_id payload fails the file.

Configuration and code

Save this as progress_batch.py. It uses only the Python 3.11 standard library.

pythonCopy
from __future__ import annotations

import hashlib
import json
import logging
import os
import tempfile
from dataclasses import dataclass
from datetime import datetime, timezone
from pathlib import Path
from typing import Mapping

LOG = logging.getLogger("progress_batch")

REQUIRED = (
    "event_id",
    "learner_id",
    "lesson_id",
    "event_type",
    "attempt_number",
    "event_time",
)
PROHIBITED = ("email", "name", "payment_token")
EVENT_TYPES = frozenset({"checkpoint", "completed"})
ALLOWED_KEYS = frozenset(REQUIRED + ("checkpoint",))


@dataclass(frozen=True)
class BatchConfig:
    product: str
    inbox: Path
    publish_dir: Path
    quarantine_dir: Path
    checkpoint_path: Path
    max_rejected_lines: int


@dataclass(frozen=True)
class FileOutcome:
    input_name: str
    input_sha256: str
    action: str
    accepted_rows: int
    rejected_rows: int


def _sha256(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 _canonical(payload: Mapping[str, object]) -> str:
    return json.dumps(payload, sort_keys=True, separators=(",", ":"))


def _parse_line(payload: Mapping[str, object]) -> dict[str, object]:
    for key in PROHIBITED:
        if key in payload:
            raise ValueError(f"prohibited field {key}")
    extra = set(payload) - ALLOWED_KEYS
    if extra:
        raise ValueError(f"unknown field {sorted(extra)[0]}")
    missing = [key for key in REQUIRED if key not in payload]
    if missing:
        raise ValueError(f"missing {missing[0]}")
    attempt = payload["attempt_number"]
    if not isinstance(attempt, int) or isinstance(attempt, bool) or attempt < 1:
        raise ValueError("attempt_number must be a positive integer")
    event_type = payload["event_type"]
    if event_type not in EVENT_TYPES:
        raise ValueError("event_type is not in the contract")
    raw_time = payload["event_time"]
    if not isinstance(raw_time, str):
        raise ValueError("event_time must be a string")
    parsed = datetime.fromisoformat(raw_time.replace("Z", "+00:00"))
    if parsed.tzinfo is None:
        raise ValueError("event_time must include an offset")
    checkpoint = payload.get("checkpoint")
    if checkpoint is not None and (
        not isinstance(checkpoint, int) or isinstance(checkpoint, bool)
    ):
        raise ValueError("checkpoint must be an integer")
    event_id = payload["event_id"]
    learner_id = payload["learner_id"]
    lesson_id = payload["lesson_id"]
    if not all(isinstance(value, str) and value for value in (event_id, learner_id, lesson_id)):
        raise ValueError("identifiers must be non-empty strings")
    normalized = parsed.astimezone(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
    return {
        "event_id": event_id,
        "learner_id": learner_id,
        "lesson_id": lesson_id,
        "event_type": event_type,
        "attempt_number": attempt,
        "event_time": normalized,
        "checkpoint": checkpoint,
    }


class ContractBreach(Exception):
    """The file must not be published."""


def _consume(path: Path, max_rejected: int) -> tuple[list[dict[str, object]], list[dict[str, object]]]:
    accepted: list[dict[str, object]] = []
    rejected: list[dict[str, object]] = []
    seen: dict[str, str] = {}
    with path.open("r", encoding="utf-8") as handle:
        for line_number, line in enumerate(handle, start=1):
            if not line.strip():
                continue
            try:
                payload = json.loads(line)
                if not isinstance(payload, dict):
                    raise ValueError("line is not an object")
                event = _parse_line(payload)
            except (ValueError, TypeError, json.JSONDecodeError) as exc:
                message = str(exc)
                if message.startswith("prohibited field"):
                    raise ContractBreach(message) from exc
                rejected.append(
                    {"line_number": line_number, "reason": message}
                )
                if len(rejected) > max_rejected:
                    raise ContractBreach(
                        f"rejected lines exceeded {max_rejected}"
                    ) from exc
                continue
            identity = _canonical(event)
            previous = seen.get(str(event["event_id"]))
            if previous is None:
                seen[str(event["event_id"])] = identity
                accepted.append(event)
            elif previous != identity:
                raise ContractBreach(
                    f"duplicate event_id with a different payload: {event['event_id']}"
                )
    return accepted, rejected


def _atomic_write(path: Path, text: str) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    descriptor, temporary_name = tempfile.mkstemp(
        dir=path.parent,
        prefix=f".{path.name}.",
        suffix=".tmp",
    )
    temporary = Path(temporary_name)
    try:
        with os.fdopen(descriptor, "w", encoding="utf-8") as handle:
            handle.write(text)
            handle.flush()
            os.fsync(handle.fileno())
        os.replace(temporary, path)
    except Exception:
        temporary.unlink(missing_ok=True)
        raise


def _load_checkpoint(path: Path) -> dict[str, dict[str, object]]:
    if not path.exists():
        return {}
    data = json.loads(path.read_text(encoding="utf-8"))
    files = data.get("files")
    if not isinstance(files, dict):
        raise ValueError("checkpoint is missing files")
    return files


def _store_checkpoint(path: Path, files: Mapping[str, Mapping[str, object]]) -> None:
    body = json.dumps({"files": files}, indent=2, sort_keys=True)
    _atomic_write(path, body + "\n")


def ingest_file(config: BatchConfig, input_path: Path) -> FileOutcome:
    """Publish one closed file or refuse it. Checkpoint updates only after replace."""
    digest = _sha256(input_path)
    published = config.publish_dir / input_path.name
    files = _load_checkpoint(config.checkpoint_path)
    prior = files.get(input_path.name)
    if (
        isinstance(prior, dict)
        and prior.get("sha256") == digest
        and published.exists()
    ):
        LOG.info(
            "product=%s input=%s sha256=%s action=skipped",
            config.product,
            input_path.name,
            digest,
        )
        return FileOutcome(input_path.name, digest, "skipped", int(prior["accepted_rows"]), int(prior["rejected_rows"]))

    try:
        accepted, rejected = _consume(input_path, config.max_rejected_lines)
    except ContractBreach as exc:
        LOG.error(
            "product=%s input=%s sha256=%s action=refused reason=%s",
            config.product,
            input_path.name,
            digest,
            exc,
        )
        return FileOutcome(input_path.name, digest, "refused", 0, 0)

    accepted_text = "".join(json.dumps(row, sort_keys=True) + "\n" for row in accepted)
    rejected_text = "".join(json.dumps(row, sort_keys=True) + "\n" for row in rejected)
    _atomic_write(published, accepted_text)
    quarantine = config.quarantine_dir / f"{input_path.name}.rejects.jsonl"
    _atomic_write(quarantine, rejected_text)
    files[input_path.name] = {
        "sha256": digest,
        "accepted_rows": len(accepted),
        "rejected_rows": len(rejected),
        "published_path": str(published),
    }
    _store_checkpoint(config.checkpoint_path, files)
    LOG.info(
        "product=%s input=%s sha256=%s action=published accepted=%s rejected=%s",
        config.product,
        input_path.name,
        digest,
        len(accepted),
        len(rejected),
    )
    return FileOutcome(input_path.name, digest, "published", len(accepted), len(rejected))


def _demo() -> None:
    logging.basicConfig(level=logging.INFO, format="%(message)s")
    root = Path(tempfile.mkdtemp(prefix="progress-batch-"))
    inbox = root / "inbox"
    inbox.mkdir()
    good = {
        "event_id": "evt_1001",
        "learner_id": "lrn_1042",
        "lesson_id": "les_07",
        "event_type": "completed",
        "attempt_number": 1,
        "event_time": "2026-03-02T14:02:11+00:00",
        "checkpoint": None,
    }
    retry = dict(good)
    bad_line = '{"event_id": "evt_1002", "learner_id": "lrn_1042"}\n'
    source = inbox / "progress-2026-03-02.jsonl"
    source.write_text(
        json.dumps(good) + "\n" + json.dumps(retry) + "\n" + bad_line,
        encoding="utf-8",
    )
    config = BatchConfig(
        product="completion_raw_events",
        inbox=inbox,
        publish_dir=root / "raw",
        quarantine_dir=root / "quarantine",
        checkpoint_path=root / "checkpoint.json",
        max_rejected_lines=10,
    )
    first = ingest_file(config, source)
    second = ingest_file(config, source)
    assert first.action == "published"
    assert first.accepted_rows == 1
    assert first.rejected_rows == 1
    assert second.action == "skipped"
    published = (config.publish_dir / source.name).read_text(encoding="utf-8")
    assert published.count("evt_1001") == 1
    assert "14:02:11Z" in published

    leaked = inbox / "progress-2026-03-03.jsonl"
    leaked.write_text(json.dumps(dict(good, email="learner@example.com")) + "\n", encoding="utf-8")
    refused = ingest_file(config, leaked)
    assert refused.action == "refused"
    assert not (config.publish_dir / leaked.name).exists()
    print(f"root={root} first={first.action} second={second.action} leaked={refused.action}")


if __name__ == "__main__":
    _demo()

What the demo asserts

The retry line collapses to one published event. The truncated line is quarantined, and the file still publishes because the cap allows one reject. The second call skips. The timestamp is normalized to Z. The file that contains email is refused and does not appear in the publish directory. The checkpoint is what makes the skip possible. Delete the published file and the next call republishes even if the hash matches, because the skip condition requires the output to exist. You can confirm that by moving the published file aside and calling ingest_file again.

How you would run it

bashCopy
python3 progress_batch.py

The log shows one published line and one skipped line. The printed summary names a temporary directory you can inspect. Nothing is installed. tempfile.mkdtemp leaves that directory on disk so you can look at the quarantine and the checkpoint. Delete it when you are finished. Production does not use a temporary root. It uses the configured prefixes and the checkpoint table.

Operational notes

Run one product at a time. Point the SQL job only at publish_dir. Treat a refused action as a page if the file is the day's completeness. A skip is success only when you expected the same bytes. A new hash that publishes is a restatement of that file name. Lesson 13's window replace should then rerun for the event dates inside the file, which this landing job does not compute. The handoff is "this input name is new," not "Tuesday's metric moved."

Holding every accepted object in a list, as the demo's _consume does, is acceptable for the fixture and the wrong production shape once the file is large. Stream writes to the temp file inside the loop, and keep only the identity hash map, before you point this at a peak day. The checkpoint order does not change.

Enterprise Real-World Scenarios

The Wednesday retry that doubled Tuesday

The laptop script appended. The retry appended. Distinct event_id counts in the published file were about twice the source for the overlap, under the planning assumption of a multi-gigabyte day, and the SQL primary key had not yet been added because the team "would dedupe in the query." The Monday number moved on Wednesday with no product change.

The replacement is the job in this lesson. Tuesday's file hash is the identity. The publish is a replace. The checkpoint records accepted and rejected counts. A rerun with the same hash skips. A rerun after a crash during replace writes the same bytes again. The operator log shows action=published once and action=skipped once. The SQL job sees one line per surviving event_id. The quarantine from the truncated final line is a counted reject, not a silent drop, and it stayed under the cap the squad set for ordinary parse errors. An email in a later file refused the publish. Tuesday's good file stayed in place.

The squad wrote the memory assumption in the run book: the identity map must be measured before enrollment week. If it does not fit, the same contract moves onto a staged table with a primary key. They did not respond to the double-count by deleting rows by hand in the published file.

A partial publish during enrollment week

A host ran out of disk while writing the temp file. os.replace had not been called. The previous published file, from an earlier partial day the producer had closed too soon, was still what SQL read. That earlier file was itself a problem of completeness: the producer had dropped a file that was still open. The batch job cannot invent rows the file does not contain. The repair on the producer side is lesson 12's rule: only a closed file is an input. The repair on the job side is that a failed write did not move the checkpoint and did not tear the published path. When the producer delivered the closed file, the hash differed, the job published, and the checkpoint moved. The curated restatement for those event dates ran because the checkpoint change was the signal, not because an operator remembered Wednesday.

The security clause showed up the same week. A debug export added email. The job refused. The landing inbox still had the file, which is private to the ingestion role in the lesson 13 zone diagram. The published raw prefix did not gain a column. The incident was the inbox file, retained and deleted under the landing retention rule, not a warehouse column to scrub from three downstream copies.

Best Practices

• Identify an input by content hash, not by file name alone. Skip only when that hash is checkpointed and the published file exists.
• Publish by atomic replace in the destination directory. Append is a retry bug.
• Write the checkpoint after the publish is durable. A redo must be safe.
• Stream the file. Do not build a list of every event on a peak day. Keep the duplicate check, and measure its memory.
• Fail the file on prohibited fields and on event_id payload collisions. Quarantine ordinary bad lines under an explicit cap.
• Normalize timestamps to UTC. Reject naive timestamps.
• Reject unknown fields at the boundary so a new personal column cannot ride through as extra JSON.
• Log action, hash, and counts. Do not log payloads or secrets.
• Point downstream readers at the published path only.
• Run one invocation per product at a time until a real lock exists. Say that in the run book.
• Keep contract fields the SQL definition needs, including checkpoint.
• When the host cannot meet the bound, record the measurement. Do not drop validation to go faster.

Common Mistakes and Anti-Patterns

Append as the write mode. Retries and overlaps double the metric, and a unique count in SQL hides the damage without stopping it. The repair is replace, with a hash checkpoint.

Checkpoint first. The next run skips a window that never became readable. The repair is the crash diagram: data, then mark. Test it by killing the process, or by raising after publish and before the mark in a drill.

Reading the temp file from SQL. A curious catalog includes every object in the prefix. The temp name becomes a second, partial day. The repair is a publish directory the SQL role can read and a temp prefix it cannot, plus names that are not valid inputs until replace.

Loading the day into a list. The job dies on memory at 02:00 UTC after looking fine on a sample. The repair is a streaming write and a measured identity index. A sample of ten thousand lines is not a peak-day test.

Quarantine as a silent success. Rejected lines are written and never counted, and a segment of learners disappears from completion. The repair is a cap that refuses the file, a log field, and a lesson 17 check that compares accepted counts with source counts.

Logging the event. Support asks for "more debug," and the log pipeline becomes a second warehouse of learner activity with none of the zone grants. The repair is identifiers in logs only when you have a privacy review, and no payloads by default.

Two crons. Overlapping runs race the replace and the checkpoint. The repair is a single scheduler slot. A local lock file is a partial measure on one host, not a cross-host design.

Editing the script to rerun a day. The path is hardcoded, the edit is deployed, and Wednesday uses Tuesday's constant. The repair is configuration. The code stays the same. The input path changes.

Troubleshooting and Design Review

Symptom: Distinct event_id in the published file is higher than in the source, and a retry was run.

Likely cause: the writer appends, or two overlapping runs each replaced and a downstream job concatenated both copies.

Investigation: compare hashes and line counts. Read the checkpoint. Look for two published log lines for one hash. Check whether SQL reads a glob that includes temp files.

Corrective action: stop readers. Replace the published file from a single consume of the source. Fix the writer. Do not DISTINCT in the dashboard and leave the file doubled.

Symptom: The job logs skipped and the published file is missing.

Likely cause: the checkpoint was written and the file was deleted later, or a bug skipped without checking the file exists. The walkthrough's condition checks existence. A homemade variant often does not.

Investigation: open the checkpoint row and the path it names.

Corrective action: republish when the file is absent. Make existence part of the skip condition. Treat deletion of a published file as an incident with a retention rule, not as a cleanup.

Symptom: A file with one bad line publishes zero events and the morning mart is stale.

Likely cause: the reject cap is zero, or a prohibited field failed the file, or a collision failed it.

Investigation: read the refused log reason. Open the inbox file's offending line in a private setting. Do not paste the line into a chat if it may contain email.

Corrective action: if the cap is wrong for ordinary parse errors, change the cap deliberately and rerun. If the field is prohibited, the file stays unpublished until the producer sends a clean file. Stale curated data is the correct consumer experience. Publishing the file to look fresh is not.

Symptom: The host memory climbs linearly and the job dies near the end of the file.

Likely cause: accepted events are retained in a list, or the identity map exceeded the host.

Investigation: look at resident set size against file size and distinct event_id estimates.

Corrective action: stream to the temp file. If the key set still does not fit, stage into a store with a unique key and keep this publish-then-checkpoint order. Take the runtime number to the Module 3 decision if the bound is missed after that.

Symptom: Two hashes for the same file name on the same morning, and completion moved twice.

Likely cause: the producer rewrote a file that was supposed to be closed. The job did what the hash rule says and republished.

Investigation: producer logs for the rewrite. Checkpoint history if you keep it. This demo's checkpoint keeps the latest hash only, which is a limit: add a history row in production if you need to see the first publish.

Corrective action: stop the producer from rewriting closed files, or version the file name. Restate the SQL window once the bytes are stable. Do not freeze the checkpoint and ignore a new hash. The new bytes are a real change.

A design review asks: What is the input identity? What does the consumer read? What is on disk if we die before replace, after replace, and after the checkpoint? What fails the file rather than the line? What is the memory behavior on the peak-day assumption? Can two runs overlap? Where are logs, and what do they omit? The review does not start by adding a framework.

Interview Questions and Answers

Q1. What makes a batch job restartable?

The same input produces the same published output without doubling, and a failure leaves either the previous publish or a replace that can be repeated. I implement that as a content hash, a write to a temp object in the destination, an atomic publish, and a checkpoint written after the publish. A retry with the same hash skips only if the published file is still there. A crash before the checkpoint redoes the replace. Append mode and a checkpoint written first both break the promise in opposite directions.

Q2. Why is a file name an insufficient checkpoint?

Producers overwrite names. progress-2026-03-02.jsonl can mean two different byte strings an hour apart. The second is a legitimate repair or an illegitimate rewrite, and either way the job must notice. SHA-256 of the bytes is the identity. The name is the address. I store both. Skipping on the name alone serves stale bytes after a repair and looks successful.

Q3. How do you handle a duplicate event_id?

I hash a canonical form of the contract fields. The same hash is one event, counted once, and the duplicate is not a reject. A different hash fails the file, because I will not choose a winner inside the loader. That matches the contract from the foundations lesson. DISTINCT downstream would hide the collision and publish one of the payloads essentially at random, depending on file order.

Q4. Where should a prohibited field be stopped in this job?

Before publish, as a file-level refusal. Quarantining the single line and publishing the rest of a file that included email means the exporter keeps adding it and some rows have already been accepted from a bad extract. The inbox copy is still a private landing object to delete under the retention rule. The published raw path stays on the previous good file. I log the reason, not the value.

Q5. What do you log, and what do you refuse to log?

Product, input name, hash, action, accepted count, rejected count, and a run id. I do not log the JSON payload, the learner's timeline, or any secret that arrives when the job later grows a database connection. Logs are an operational store with their own retention and access. They are not a second raw zone. If I cannot debug from hash and counts, I reproduce against the inbox file in a private path.

Q6. How does this job relate to the completion SQL?

It publishes raw events and keeps the fields the definition needs, including checkpoint. It does not decide whether an attempt completed. The SQL window replace reads only the published path, after the checkpoint says the file is that hash. If I compute completion in this process and drop unused fields, I recreate the ETL failure from lesson 13 inside a better checkpoint. Placement and restartability are both required. Neither substitutes for the other.

Q7. A peak day is fifty million events. What breaks first in the obvious Python implementation?

A list of every parsed object, then the identity map if I store full payloads, then a single-threaded parse if the morning bound is tighter than one host's read speed. I stream writes, store hashes for identity, and measure resident memory and elapsed time. If the key set does not fit, I stage to a table with a primary key. If one host misses the bound after that, I take the numbers to a distributed engine. I do not solve memory pressure by turning off the collision check.

Q8. Why is os.replace part of the design, and when is it the wrong primitive?

On one filesystem it makes the published name switch from the old file to the new file without a reader seeing a half-written object. A temp file on another mount degrades to a copy. An object store wants its own complete-and-promote operation, documented by that store, not a pretend rename. I keep the visibility rule and replace the primitive. I test by killing the process during the write and confirming the published name is either the old object or the new one.

Q9. How do you decide the reject cap?

With the product owner, as a count or a rate that means "this file is not the day." Prohibited fields and identity collisions ignore the cap and fail the file. Ordinary truncated lines count toward the cap. Zero is appropriate when every line is supposed to be well formed and the file is small. A non-zero cap needs a quarantine someone reads and a metric on the count. An infinite cap is the silent-drop bug. Lesson 17 is where the threshold becomes a suite. The loader still needs a default that fails closed.

Q10. What happens if two schedulers start the job together?

Both can pass the skip check, both can publish, and the checkpoint can describe the loser of the race or the winner, depending on timing. This lesson's contract is single execution per product. I want the scheduler to enforce that, or a lock the platform actually provides across hosts. I will not describe the job as exactly-once under two crons. The publish replace limits the damage relative to append, and it does not make the checkpoint linearizable by itself.

Q11. Why normalize event_time in the loader?

So two encodings of the same instant do not become two raw values and so a missing offset cannot be interpreted as local time on whatever host ran the job. I store UTC. I reject naive timestamps. I do not truncate to a date in this job, because the business day is the SQL window's decision and a late event still needs its instant. Normalization is not the completion rule.

Q12. What evidence, after a month, shows the landing job is fit?

Retries of an unchanged file do not change the published hash or the accepted count. A drilled crash before the checkpoint results in one publish, not two. A fixture containing email never appears in the publish prefix. Reject counts are non-zero only when the quarantine explains them. The SQL grain constraint does not fire because of loader doubles. Peak-day memory and runtime are written down. A green exit code with an append-mode file and no checkpoint is not evidence.

Hands-On Exercises

Exercise: Make a one-day landing job restartable

Problem statement. Implement a batch landing for progress events, or a similarly shaped file you define, with a checkpoint that survives a retry and a crash window you can explain.

Tasks.

1. Define the input identity, the published path, and the checkpoint fields. State the skip rule.
2. Specify which conditions refuse the file and which quarantine a line. Include one prohibited field.
3. Implement a streaming or small-file version with atomic replace and a checkpoint written last. Use the standard library.
4. Test an unchanged rerun, a payload collision, a prohibited field, and a malformed line under your cap.
5. Describe the disk state after a crash in each segment of the crash diagram, and how the next run behaves.
6. State the memory assumption and what you would change if the identity map does not fit.
7. Name the downstream reader and the path it must not read.
8. Apply the six-part lens in short paragraphs. Label the peak volume as an assumption.

Expected deliverables. The program, the tests or demo assertions, the crash note, and the lens notes.

Evaluation rubric.

You pass when a reviewer can kill the process in their head at two points and predict the next run without asking you which file is "the real one."

Lesson Summary

A batch pipeline is restartable when the input has a content identity, the publish is a replace a reader cannot observe half-written, and the checkpoint moves only after that publish. Validation refuses a file that contains a prohibited field or a conflicting event_id, and it quarantines ordinary bad lines under a cap. Logs tell the operator what happened. They do not carry the events.

TechLearningPro's progress landing is this job. The completion definition stays in SQL over the published files. Peak-day volume is a memory and runtime measurement, not a reason to drop the collision check or to hold every object in a list. One run of a product at a time is part of the contract until the scheduler exists.

API and File-Based Data Ingestion is the next source shape: pages and manifests, where the "closed file" this lesson assumes has to be created by the ingestion job itself.

Further Reading

• Python documentation: os.replace and hashlib — the publish and the identity used here. The language primer is /learn/python.
• Python documentation: json — dumps with sort_keys is the canonical form for the duplicate check. It is not a schema language.
• Python documentation: logging — stderr logs for operators. Configure a handler in the runner you deploy. The demo uses basicConfig.
• Lessons 4 and 5 of this course — streaming reads and the failure modes of holding a batch in a list.
• Martin Kleppmann, Designing Data-Intensive Applications — idempotent sinks and the difference between a successful side effect and a recorded position. The checkpoint-last rule is that chapter in miniature.
• Continue this course: API and File-Based Data Ingestion, then Change Data Capture Fundamentals.