Advanced Python for Data Pipelines

Data Engineering & Modern Data Platforms · Lesson 05 · Data Engineering Foundations

Lesson Overview

The strict parser from lesson 4 was correct on a fixture and unusable on a peak day. It built a list of every JSON object before it reviewed uniqueness, it opened files without a guarantee they would close if a line failed, and the only record of a rejection was a print on someone's terminal. When the job was wrapped in a cron entry, two overlapping runs processed the same batch id and both believed they were the writer. The raw zone, which lesson 1 said must not contain a partial opinion, contained two copies and a traceback with a learner id in it.

Those are not new business rules. They are the operational shape of the rules you already accepted: stream the file, close the handle, log a code rather than a payload, test the contract, and make a rerun of the same batch id a no-op rather than a second copy. This lesson is that shape. Generators, context managers, concurrency, packaging, logging, and tests are taught only as far as a batch ingest worker needs them. /learn/python remains the language primer. Orchestrators are lesson 19 and lesson 20. This worker is something an orchestrator can call, not an orchestrator itself.

The planning assumption is unchanged and still labeled. A peak day is on the order of fifty million progress events. A single batch file is a slice of that day, large enough that holding every parsed object in a list is the failure mode, small enough that an index of event_id to a payload hash can live in memory. If the index itself cannot, the batch is the wrong size and the fix is smaller files, not a more clever process. One machine, one process, several files read sequentially is the default. Threads appear only for overlapping IO you have measured. They are not a badge.

Learning Objectives

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

1. Stream an NDJSON file with a generator so a rejected row does not require the whole file as a list.
2. Use a context manager to close files and to represent a batch lease that releases on failure.
3. Explain when a generator's laziness changes correctness, not just memory, for a uniqueness check.
4. Choose threads, processes, or a single thread for a pipeline step, using the GIL and the actual bottleneck.
5. Log contract codes and batch ids without logging prohibited values, through the standard logging module.
6. Package the worker so a command and a library import are not the same side effect.
7. Write tests that cover conflicts, coercion, cleanup, and idempotent reruns.
8. State the concurrency design you refused, and the measurement that would reopen it.

Prerequisites

• Lesson 4's strict types, ContractError, frozen events, and the rule that a conflict rejects the batch. This lesson streams that parser. It does not loosen it.
• Lessons 1 and 3: batch identity, one writer for the raw zone, and exit status as a severity signal.
• You can read a with statement and a for loop. Generators and concurrent.futures are introduced here. If function syntax is still unsteady, use /learn/python and return.
• Python 3.11 or newer, standard library only, plus a test runner. The tests below use unittest, which ships with CPython. pytest is a reasonable alternative and is not required.

Detailed Core Concepts

Generators are a control-flow contract

A generator function contains yield. Calling it does not run the body. It returns a generator object. The body runs up to the next yield when the consumer pulls a value. That laziness is the point and the hazard.

For NDJSON, laziness means you can parse a line, hand it to the reviewer, and drop it before reading the rest. Peak-day memory stays proportional to one line plus the uniqueness index, not to the file. The hazard is that a generator is single-pass. You cannot rewind it to "look again." Any rule that needs two passes must either store what it needs on the first pass or open the file twice. Uniqueness needs the first-seen payload, or a hash of it, stored in the index. It does not need the file stored twice.

A generator that is not fully consumed leaves the rest of the body unexecuted. If the with block that owns the file sits inside the generator, the file stays open until the generator is closed or garbage-collected. Context managers and generators meet in yield from inside with, or in contextlib.closing, or in a consumer that always exhausts or closes the generator. The walkthrough puts the file ownership in the caller and lets the generator only wrap a text stream. Ownership stays visible. A generator that opens a path as a side effect of iteration hides a handle in an object someone stored in a list "for later."

Do not use a generator to swallow errors. If parse_event raises ContractError, the generator can yield a result object that is either an event or a violation. That is an explicit stream of outcomes. Catching the error inside the generator and yielding nothing drops the row and looks like a short file. The batch policy is "reject," not "skip." The stream has to carry the violation so the consumer can stop and discard the output.

Result objects instead of mixed exceptions

Once the stream crosses a function boundary, exceptions are a poor protocol for "this row is bad and here is the code," because a consumer that wants to count and continue must know which exceptions are policy. Keep ContractError for the row, and yield a small frozen result:

• Parsed(event) for a good row
• Failed(error) for a contract failure

The consumer applies batch policy. The generator does not decide to skip. This is the same split as lesson 4, extended to a stream. Programmer bugs still propagate. The generator does not catch Exception.

Context managers

A context manager is an object with __enter__ and __exit__, usually written with @contextmanager or as a class, and used in with. __exit__ runs when the block ends, including when the block raises. That is the whole feature. Use it for anything you must release: a file handle, a temporary directory, a lock, a batch lease.

The standard library already provides the file object as a context manager. Use it. Do not call open and rely on the process exiting to close the descriptor. A worker that processes hundreds of files will exhaust descriptors before it exits.

A batch lease is the application-level version of the same idea. Before reading, the worker creates a marker that batch id b_2026_03_02_14 is in progress, with an owner token. The context manager writes the marker on enter and, on a successful exit, renames it to a committed marker. On an exception, it writes a failed marker and removes any partial output file. The next run sees the committed marker and returns without appending. The next run sees the failed marker and may retry. The next run sees the in-progress marker held by a live owner and refuses to start a second writer. The lesson's implementation uses local files as the marker store so you can run it without a database. The protocol is the design. A database row with a unique constraint on batch_id is the production form of the same protocol, written in lesson 6's engine when you are ready to hold the lease there. Do not invent a distributed lock service for a single worker.

contextlib.contextmanager is enough for the lease. You do not need a class unless the lease has several methods the caller will use inside the block. Prefer the shorter form when the only operations are enter and exit. Always try / yield / finally if the cleanup must run. Code after yield does not run if the block raised, unless it is in finally or you handle the error with the decorator's usual pattern: a try around yield.

Idempotent batch runs

Idempotent means a second successful execution with the same inputs does not change the outcome. For this worker, the identity is the batch id, not the clock. The producer, or the extract job, supplies source_batch_id. The worker does not invent one from datetime.now(), because two retries would then be two ids and both would land.

The output file is named by the batch id. The commit is a rename of a temporary file in the same directory onto the final name, after the review has accepted every row, and after the lease moves to committed. Rename on the same filesystem is atomic on the platforms this course assumes for local execution (POSIX rename). A reader that looks only at the final name either sees the previous complete file or the new complete file, not a half-written one. That is the local version of the publication rule in lesson 2.

If the process dies after rename and before the marker is committed, a retry might rewrite the same bytes and commit again. That is acceptable if the output is replaced, not appended. Appending is the defect. The walkthrough writes a new temp file and replaces. It never opens the final path in append mode.

A conflict or a bad row deletes the temp file and does not rename. The failed marker records the first violation code. It does not record the payload.

Concurrency, the GIL, and honesty about speed

CPython runs Python bytecode on one thread at a time because of the global interpreter lock. IO operations release the lock while they wait. CPU-bound Python, including json.loads of a huge line and a pure Python uniqueness check, generally does not speed up across threads. Processes do, at the cost of copying data and of a harder failure story.

For this ingest worker:

concurrent.futures.ProcessPoolExecutor is the standard-library tool if you cross into multiple files. Each process parses one file to its own temp output. The parent commits leases. Do not share a dict across processes and hope. multiprocessing requires the arguments to be picklable. A generator is not an argument you pass to a pool. A path is.

asyncio excels when you have many waiting sockets. This worker has files. Using asyncio to read a local disk adds a loop and no overlap the OS is not already providing through a buffered read. Refuse it here. The reopening condition is a measured profile that shows the worker blocked on many concurrent network reads, which is a different program.

The design you should be able to say in review: one file, one process, a generator, a lease. Parallelism across files only after a timing run shows the sequential worker misses the freshness bound and the disk is idle. Lesson 2's reopening habit applies to threads.

Logging

print goes wherever the process's stdout goes and has no severity. A cron job's print is a mail from the 1990s or a log line without a level, depending on the machine. The logging module lets the orchestrator set the handler. The library code asks for a logger with logging.getLogger("progress.ingest") and does not configure handlers itself. Configuring handlers in the library makes tests emit to the console and makes a second configuration fight the first. The command entrypoint may call logging.basicConfig if nobody else did.

Log the batch id, the line number, and the error code. Do not log the mapping. Do not log learner_id at info level in a loop: fifty million info lines of identifiers are a storage and privacy incident of their own. A single summary at the end, with counts, is the info log. Warnings are for a retry of a failed marker you are about to replace. Errors are for a rejected batch. Exceptions that are bugs should use logger.exception in the command entrypoint so the traceback is kept, outside the row loop.

Levels are a contract with the on-call engineer. If everything is error, nothing is. If contract rejections are debug, the page will say only "exit status 2" and someone will add a print of the payload to "see what's wrong." Give them the code and the line in the error log so they do not need the payload.

Packaging

A pipeline that lives as ingest_final_v2.py on a laptop will be the one production does not run. The minimum package is a directory with pyproject.toml, a package folder, and a console script. You do not need to publish it to a package index. You need to install it in the worker's virtual environment from that directory so the import path is the package, not whatever file the shell's current directory happens to shadow.

pyproject.toml with the standard [project] table and a setuptools build backend is enough. Pin the Python version to >=3.11. Do not add pandas "because the file is right there." Dependencies you do not import are an attack surface and a resolution problem. The lesson's worker imports the standard library only.

The console script calls main. main returns an int. It does not parse files at import time. Tests can import progress_ingest.review without executing a batch. That was a lesson 4 rule. Packaging is how the rule survives installation.

Tests

A test is a function that fails when the contract breaks. unittest is enough. Each test builds a small string or a temp directory. None of them read a production bucket. None of them depend on the clock except where the test freezes the batch id itself.

Mandatory cases, added to lesson 4's list:

• A generator yields a failure and stops being consumed; the file handle is closed. You can assert this by using a context manager around the source and checking that exiting the consumer's with does not raise, and that a second open of the path succeeds. On a stricter test, use a custom stream that records close.
• Two runs with the same batch id produce one output file and the same bytes. The second run logs that it skipped.
• A failure leaves no final output file.
• An overlapping lease raises and does not append.
• Log records captured with assertLogs do not contain an email address planted in a bad row.

Tests that read the real raw zone are not unit tests. They are drills. Keep them out of this module.

Architecture and Design Diagrams

The worker's data flow is a single file in and a single committed file out, or nothing.

mermaidCopy
flowchart TD
    src[NDJSON path and batch id]
    lease{Lease state}
    skip[Return success without writing]
    busy[Refuse: another writer holds the lease]
    read[Generator yields parsed or failed rows]
    temp[Temp file of accepted ids and hashes]
    reject[Delete temp, mark failed, exit 2]
    commit[Rename temp onto output, mark committed]
    src --> lease
    lease -->|committed and same checksum| skip
    lease -->|in progress by a live owner| busy
    lease -->|absent or failed| read
    read -->|any Failed| reject
    read -->|all Parsed| temp --> commit

The checksum in the skip branch is how you notice that someone reused a batch id for a different file. Same id and different bytes is ID_REUSE, a contract failure at batch level, not a skip. Same id and same bytes is the idempotent retry. The diagram's "same checksum" box is the decision. Without it, a reused id silently drops a new file.

Process shape, including what you are not running:

mermaidCopy
flowchart LR
    subgraph one [One process per file]
        gen[Generator]
        idx[In-memory id index]
        log[Logger]
        gen --> idx
        gen --> log
    end
    pool[Process pool across files, only if measured]
    pool -.-> one
    asyncio[asyncio loop]
    threads[Threads sharing one index]
    one -.->|refused by default| asyncio
    one -.->|refused| threads

Dotted arrows are refusals. They belong on the diagram so a later performance patch does not add them as a drive-by.

Lifecycle of the lease, which is the control plane next to the data flow:

mermaidCopy
stateDiagram-v2
    [*] --> Absent
    Absent --> InProgress: enter lease
    InProgress --> Committed: output renamed
    InProgress --> Failed: contract error or crash recorded
    Failed --> InProgress: retry
    Committed --> Committed: same checksum, skip
    Committed --> Failed: same id, different checksum

A crash that dies inside InProgress without running __exit__ leaves a stale marker. The lease stores a timestamp and a token. A retry may take over a marker older than a stated age, and it must log the takeover. Choosing the age is an operability decision: too short and a slow file is stolen by a second writer; too long and a dead worker blocks the day. For teaching, the walkthrough uses an explicit break_lease=True in tests rather than a wall clock, so the suite does not sleep. Production code can compare mtime and must document the minutes. This lesson records the state machine so the minutes have a place to live.

Technical Explanation

Hashing a payload without storing it

The uniqueness index maps event_id to a SHA-256 of the canonical JSON of the accepted fields, not of the raw line. Canonical means the parsed fields you care about, serialized with sorted keys and no insignificant whitespace, via json.dumps(fields, sort_keys=True, separators=(",", ":")). Two lines that differ only in key order are the same event. Two lines that differ in event_type hash differently and conflict.

Do not hash the raw line. A producer that emits the same fields in a different order would look like a conflict. Do not include prohibited keys in the hash input by spreading the whole mapping. The parser has already rejected those keys. Hash the dataclass fields you stored.

hashlib.sha256 is the standard library. You do not need a non-cryptographic hash for this volume at the file level, and you should not invent a shorter checksum that collides on the one day two payloads differ. Store the hex digest as a str in the dict. Fifty million digests is a memory budget you must check: 64 bytes of hex plus dict overhead is multiple gigabytes. That is why the batch file is a slice, not the day. Compute the budget before you brag that the generator "uses constant memory." The generator does. The index does not.

Stopping early

If the policy is "any violation rejects the batch," the consumer may stop reading after the first failure, delete the temp file, and mark failure. That saves time. It also hides the count of later violations, which the on-call engineer wants. A practical compromise: stop after a cap, such as 100 violations, and log that the scan stopped. Record the cap in the failed marker. Do not stop at one unless the file is enormous and the first error is a framing error that makes the rest meaningless, such as a binary file accidentally dropped in the directory. UnicodeDecodeError is that kind of failure. Fail the batch immediately. There is no second opinion in a file that is not UTF-8.

Partial output is a bug

Writing accepted rows to the final path as you go, and deleting them if a later row fails, is a race with any reader of the final path. Write the temp file. If you stop early, delete the temp file in the lease's failure path. The finally block is not optional. An exception between "decided to fail" and "deleted temp" is how partial files survive. The context manager's failure path performs the delete. The review function does not know the file name. That ignorance is useful. It keeps IO out of the pure review, which stays testable with strings.

Pickling and process pools

If you add a process pool, the worker function must be importable at module level. Nested functions and lambdas are a poor target for ProcessPoolExecutor because they must be pickled by name. Pass pathlib.Path or str, and a batch id, and return a small summary dataclass. Do not return the generator. Do not pass the open file handle.

Start with max_workers equal to a small number you can explain, often the count of files you actually have or a cap below the disk's ability to seek itself to death. A pool of 32 workers on one spinning disk, or on one small SSD with 32 uncompressed JSON readers, can be slower than 2. The measurement is elapsed time and IO wait, not CPU percent in a screenshot. This lesson does not include the pool in the default command. The explanation is here so the refusal is informed.

Logging configuration and tests

assertLogs captures records without depending on global handler state if you attach them to the named logger. Tests should not call basicConfig. If a test fails only when the suite runs after a test that configured logging, you have a leaked global. The idempotent worker's skip path logs at info. The rejection logs at error with code and line. Parameterize nothing through the root logger.

Versioning the package

The package version in pyproject.toml is the version the job runs. Log it at startup. When a parser rule changes, bump the version. The raw zone's batch marker can store that version. A replay question then has a chance of answering "which checker accepted this file?" Definition version of the metric remains a different number, owned by the analytics hat. Do not overload one version for both. A checker change that still accepts the same events does not bump the metric. A checker change that rejects events the old checker stored is an incident, and the version in the log is how you find the first batch that used it.

Practical Implementation or Design Walkthrough

Problem

Stream a progress NDJSON file, reject the batch on the first hundred contract problems or on any conflict, commit the output by rename, and do nothing harmful when the same batch id is submitted again.

Layout

textCopy
progress_ingest/
  pyproject.toml
  src/progress_ingest/__init__.py
  src/progress_ingest/worker.py
  tests/test_worker.py

pyproject.toml minimum:

tomlCopy
[project]
name = "progress-ingest"
version = "0.1.0"
requires-python = ">=3.11"
description = "Stream and lease a progress-event batch"

[project.scripts]
progress-ingest = "progress_ingest.worker:main"

[build-system]
requires = ["setuptools>=68"]
build-backend = "setuptools.build_meta"

[tool.setuptools.packages.find]
where = ["src"]

Install for development with pip install -e . from that directory, in a virtual environment. You do not need network access if setuptools is already present. If it is not, the install uses the Python packaging index, which is a normal dependency fetch, not a data upload.

Worker

The parser functions from lesson 4 are assumed imported as parse_event and ContractError. They are inlined in shortened form only where the new code needs a type. The new code is the stream, the lease, and the commit. Save as worker.py inside the package.

pythonCopy
from __future__ import annotations

import hashlib
import json
import logging
import os
import tempfile
from contextlib import contextmanager
from dataclasses import dataclass
from pathlib import Path
from typing import Iterator, TextIO

from progress_ingest.contract import ContractError, parse_event  # lesson 4 module, moved into the package

logger = logging.getLogger("progress.ingest")
VIOLATION_CAP = 100


@dataclass(frozen=True)
class Accepted:
    event_id: str
    digest: str


@dataclass(frozen=True)
class RowFailure:
    line_number: int
    code: str


def iter_outcomes(handle: TextIO) -> Iterator[Accepted | RowFailure]:
    """Yield one outcome per non-empty logical line. Framing errors propagate as ContractError."""
    index: dict[str, str] = {}
    for line_number, line in enumerate(handle, start=1):
        if line.strip() == "":
            yield RowFailure(line_number, "BAD_FRAMING")
            continue
        try:
            payload = json.loads(line)
        except json.JSONDecodeError:
            yield RowFailure(line_number, "BAD_FRAMING")
            continue
        if type(payload) is not dict:
            yield RowFailure(line_number, "BAD_FRAMING")
            continue
        try:
            event = parse_event(payload)
        except ContractError as exc:
            yield RowFailure(line_number, exc.code)
            continue
        digest = hashlib.sha256(
            json.dumps(
                {
                    "event_id": event.event_id,
                    "learner_id": event.learner_id,
                    "lesson_id": event.lesson_id,
                    "event_type": event.event_type,
                    "attempt_number": event.attempt_number,
                    "event_time": event.event_time.isoformat(),
                },
                sort_keys=True,
                separators=(",", ":"),
            ).encode("utf-8")
        ).hexdigest()
        previous = index.get(event.event_id)
        if previous is None:
            index[event.event_id] = digest
            yield Accepted(event.event_id, digest)
        elif previous != digest:
            yield RowFailure(line_number, "ID_CONFLICT")
        # identical retry: yield nothing new


@contextmanager
def batch_lease(root: Path, batch_id: str) -> Iterator[Path]:
    root.mkdir(parents=True, exist_ok=True)
    final = root / f"{batch_id}.ndjson"
    marker = root / f"{batch_id}.lease"
    if marker.exists() and marker.read_text(encoding="utf-8").startswith("committed"):
        yield final
        return
    if marker.exists() and marker.read_text(encoding="utf-8").startswith("in-progress"):
        raise RuntimeError(f"batch {batch_id} is already in progress")
    marker.write_text("in-progress\n", encoding="utf-8")
    try:
        yield final
    except Exception:
        marker.write_text("failed\n", encoding="utf-8")
        raise
    else:
        marker.write_text("committed\n", encoding="utf-8")


def commit_batch(source: Path, root: Path, batch_id: str) -> str:
    """Return 'skipped', 'committed', or 'rejected'."""
    digest = hashlib.sha256(source.read_bytes()).hexdigest()
    done = root / f"{batch_id}.sha256"
    if done.exists() and done.read_text(encoding="utf-8").strip() == digest:
        logger.info("batch %s skipped; checksum unchanged", batch_id)
        return "skipped"
    if done.exists():
        logger.error("batch %s reused with a different checksum", batch_id)
        return "rejected"
    temp_dir = root / "tmp"
    temp_dir.mkdir(exist_ok=True)
    failures: list[RowFailure] = []
    fd, temp_name = tempfile.mkstemp(dir=temp_dir, suffix=".part")
    os.close(fd)
    temp_path = Path(temp_name)
    try:
        with source.open(encoding="utf-8") as handle, temp_path.open("w", encoding="utf-8") as out:
            for outcome in iter_outcomes(handle):
                if isinstance(outcome, RowFailure):
                    failures.append(outcome)
                    logger.error("batch %s line %s code %s", batch_id, outcome.line_number, outcome.code)
                    if len(failures) >= VIOLATION_CAP:
                        break
                else:
                    out.write(f"{outcome.event_id}\t{outcome.digest}\n")
        if failures:
            temp_path.unlink(missing_ok=True)
            return "rejected"
        final = root / f"{batch_id}.accepted"
        os.replace(temp_path, final)  # atomic on the same filesystem
        done.write_text(digest + "\n", encoding="utf-8")
        logger.info("batch %s committed", batch_id)
        return "committed"
    except Exception:
        temp_path.unlink(missing_ok=True)
        raise

os.replace is the atomic rename. marker in batch_lease is the teaching lease; commit_batch uses the checksum file as the idempotency record so the demo can run without nesting both mechanisms incorrectly. In a production wiring, call commit_batch inside batch_lease and treat a raised RuntimeError as exit status 1 (operational) and a "rejected" return as exit status 2 (contract). The split is deliberate: a held lease is not a bad record.

Test

pythonCopy
import unittest
from pathlib import Path
from tempfile import TemporaryDirectory

from progress_ingest.worker import commit_batch


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:11Z"}\n'
)


class CommitTests(unittest.TestCase):
    def test_second_run_skips(self) -> None:
        with TemporaryDirectory() as tmp:
            root = Path(tmp)
            source = root / "in.ndjson"
            source.write_text(GOOD + GOOD, encoding="utf-8")
            self.assertEqual(commit_batch(source, root / "out", "b1"), "committed")
            self.assertEqual(commit_batch(source, root / "out", "b1"), "skipped")
            accepted = (root / "out" / "b1.accepted").read_text(encoding="utf-8")
            self.assertEqual(accepted.count("evt_1001"), 1)

    def test_conflict_writes_nothing(self) -> None:
        with TemporaryDirectory() as tmp:
            root = Path(tmp)
            source = root / "in.ndjson"
            bad = GOOD + GOOD.replace("completed", "checkpoint")
            source.write_text(bad, encoding="utf-8")
            self.assertEqual(commit_batch(source, root / "out", "b2"), "rejected")
            self.assertFalse((root / "out" / "b2.accepted").exists())


if __name__ == "__main__":
    unittest.main()

The second line of the conflict fixture changes event_type and keeps evt_1001, so the digest differs. commit_batch must return rejected and leave no .accepted file. The identical two-line file commits one id because iter_outcomes yields the duplicate delivery as nothing.

Commands

bashCopy
python3 -m unittest tests/test_worker.py
python3 -m progress_ingest.worker --help

Wire main to parse a path and a batch id, call commit_batch, and return 0 for committed and skipped, 2 for rejected. Logging configuration belongs in main only.

What this walkthrough refuses

It does not start a process pool. It does not import asyncio. It does not delete a committed file on a second run. It does not log the JSON line. It does not call an orchestrator API. Airflow, in lesson 20, can shell out to progress-ingest. The DAG is not an excuse to move the lease into a notebook.

Enterprise Real-World Scenarios

Overlapping cron and a double Tuesday

The first production schedule was 0 * * * * with no lease, because the file "usually finishes in four minutes." On enrollment week it finished in seventy. The next hour started a second process. Both appended to tuesday.ndjson. The mart job, which lesson 2 said must replace a day, ingested a file that was a concatenation of two partial opinions and one traceback dumped by a wrapper script. Completion for a subset of learners doubled until the restatement, and the restatement doubled them again because the raw file itself was duplicated.

The repair was the lease plus os.replace, the checksum skip, and the removal of append mode. The cron overlap became harmless: the second process saw in-progress and exited 1, which paged, which was correct, because an overlap means the schedule is wrong. The data engineer owned the worker. Platform owned the schedule. Both names were already in the lesson 3 matrix, and the incident was the first time the page condition "two writers" was added. Analytics was informed. The definition version did not change. Recomputing Tuesday from the replica, not from the concatenated file, was the backfill. The replica was still the system of record. The concatenated file was a derived artifact that had been treated as raw. The worker's commit protocol is what keeps derived files from impersonating a source.

A thread pool that made the index lie

A performance review suggested eight threads parsing one file, sharing the index dict, because a laptop had eight cores. JSON parsing held the GIL often enough that elapsed time got worse, and the shared dict lost updates: two threads accepted the same new event_id before either stored the digest. The uniqueness rule, which is the correctness rule, was now probabilistic. The bug did not show on the small fixture because the race lost the race.

The repair was to delete the pool and to write down the refusal. A later measurement, with four separate hourly files on a machine whose disk was idle, justified a process pool of two, each with a private index and a private temp file. That patch was a different design, reviewed as a design, with a test that runs two batch ids and checks they cannot see each other's leases. "Threads share memory, so the index is easy" is the sentence this scenario exists to retire.

Best Practices

• Stream lines with a generator. Keep the uniqueness index explicit and size the batch so the index fits.
• Put file handles and leases in with blocks. Release them on failure, not only on success.
• Commit output by replacement in the same directory. Do not append to a published path.
• Make the batch id an input. Treat the same id and the same checksum as success with no write. Treat the same id and a different checksum as a rejection.
• Log codes, line numbers, and batch ids. Do not log payloads or identifier values at row level.
• Configure logging in the entrypoint. Libraries only get a logger.
• Keep the worker installable with a pyproject.toml and a console script. Import side effects are a defect.
• Test skip, reject, and "no final file" with temporary directories. Capture logs and assert the prohibited value is absent.
• Default to one process per file. Add processes across independent files only after a measurement, with private indexes.
• Cap violation collection so a bad file cannot emit a million log lines, and record that you stopped.
• Leave orchestration, retries with backoff, and queues to later lessons. A clean exit status is the interface they need.

Common Mistakes and Anti-Patterns

Reading the file twice by accident, or not being able to. A generator was consumed in a logging pass that "just counts," and the real pass sees nothing. The repair is one pass that yields outcomes, or an explicit second open. Do not store the generator in a variable and iterate it in two places.

A context manager that commits in __exit__ unconditionally. Cleanup that renames the temp file even when the block raised will publish a partial batch. The repair is try / yield / except that marks failure and deletes, and an else that commits only when the block did not raise. Test the exception path.

Append mode as a checkpoint. Writing rows out as you parse feels recoverable. A crash and a retry duplicate the front of the file. The repair is temp file plus replace, and an idempotency key at the batch level. Row-level resume is a later optimization with its own index. It is not open(path, "a").

Threads around a shared dict. The uniqueness index is shared mutable state. The GIL does not make dict updates atomic as a business operation: check-then-set is two steps. The repair is one thread per index, or one process per index. Locks around the whole parse negate the threads.

logging.basicConfig inside the library, and print of the bad line beside it. Tests become order-dependent, and production logs fill with emails the moment a prohibited field appears. The repair is a named logger, assertLogs in tests, and messages that contain codes. The prohibited-field test from lesson 4 moves up a level: the log record is what must stay clean.

A package that is really a script with a relative path to a home directory. The job works on one laptop. The repair is pyproject.toml, a src layout, and configuration of the input path by argument. A path baked into the module is a credential-adjacent habit even when the path is not a secret. The next path will be.

Catching Exception in the generator and continuing. The stream looks complete. The mart is thin. A MemoryError or a bug becomes a short Tuesday. The repair is to yield RowFailure only for ContractError and framing, and to let everything else abort the lease through the context manager.

Troubleshooting and Design Review

Symptom: The second scheduled run doubles the accepted file.

Likely cause: append mode, or a batch id that includes the clock, so the retry is a new id.

Investigation: read the open mode and the batch id source. Compare checksums of the two outputs. Look for two leases.

Corrective action: switch to replace and a caller-supplied batch id. Rebuild the raw file from the replica for the affected window. Do not dedupe the concatenated file by hand unless you can prove the lines. Prefer re-extract.

Symptom: Descriptors leak and the worker eventually cannot open files.

Likely cause: a generator opens the path and the consumer breaks out of the loop without closing the generator.

Investigation: lsof on the worker, or a test stream that counts close. Confirm with wraps the open, and that breaking out of the consumer still exits the with.

Corrective action: open in the consumer's with, not inside an abandoned generator. Call generator.close() if the generator owns other resources. Add the test before the next schedule.

Symptom: A retry of a failed batch refuses to start and the marker says in-progress for hours.

Likely cause: the process was killed with a signal that skipped __exit__, or the lease age is infinite.

Investigation: read the marker timestamp and the process table. Confirm the worker pid is gone.

Corrective action: a documented takeover after a stale age, logged as a warning, with the same temp-file protocol so a late zombie cannot rename after the new owner. If you cannot prove the zombie is dead, do not steal the lease. Kill the schedule first.

Symptom: Logs contain email addresses after a prohibited-field rejection.

Likely cause: a formatter included the line, or an exception message included the value.

Investigation: assertLogs with a fixture that contains a sentinel address. Read the handler formatter in the entrypoint.

Corrective action: log code and line only. Scrub the retained logs according to the privacy playbook. Add the sentinel test to the suite so the formatter cannot regress.

Symptom: Eight worker processes are slower than one, and the disk is saturated.

Likely cause: a process pool was sized to CPU count for a workload that is read and parse of many small uncompressed files.

Investigation: compare elapsed time at 1, 2, and 4 workers on a realistic file. Watch IO wait.

Corrective action: reduce the pool or remove it. Revisit only when files are independent and a measurement says the bound is missed. Do not compensate with threads inside each process.

A design review asks: what is held in memory besides one line, what is the batch id, what happens to the temp file on the first violation, what does the second run do, what is logged, and which concurrency options were refused? A review that opens with "can we asyncio this" has not looked at the bottleneck.

Interview Questions and Answers

Q1. Why stream NDJSON with a generator instead of readlines()?

readlines() and a list of parsed objects hold the file in memory, which fails the peak-day assumption even when the contract logic is right. A generator parses one line at a time and lets the caller discard it. The uniqueness structure is still in memory, so I do not claim constant memory overall. I claim the event bodies are not retained. Generators are single-pass, so I either keep the digest index as I go or I open the file again on purpose. I do not iterate the same generator twice and expect rows.

Q2. How do a context manager and a generator interact if both touch the file?

If the generator opens the file, the handle lives until the generator is closed or collected, and a consumer that breaks early can leak it. I prefer the caller to open the file in with and pass the handle into the generator, so the block's exit closes the handle even when iteration stops. If a generator must own the resource, I wrap it so __exit__ or close() runs, and I test that path. The lease context manager is separate: it marks failure if the block raises, and it must not rename a partial output in that path.

Q3. What does idempotency mean for a batch ingest, concretely?

The caller supplies a batch id. If that id was already committed and the file checksum matches, the second run logs a skip and writes nothing. If the checksum differs, the run rejects the reuse instead of replacing history silently. If the first run failed, a retry may proceed and still only publishes via an atomic replace. Idempotency is not "append and hope the mart dedupes." The mart should not be asked to clean up a raw file that double-wrote.

Q4. When do threads help a Python ingest worker, and when do they threaten correctness?

Threads can overlap waiting on IO because CPython releases the GIL during many blocking calls. They do not speed up pure parsing reliably, and they break a shared uniqueness dict if two threads interleave check-and-set. For one NDJSON file I use one thread. For many sockets, which this worker does not have, threads or asyncio might be justified by a profile. I do not put the id index in shared memory without a design that makes the check-and-set one critical section, and I would rather give each worker its own file and its own index.

Q5. Why not asyncio for local files?

Asyncio pays for itself when many tasks are waiting on the network or another process. A local file read is already buffered, and the CPU work is json.loads, which is not an await point that frees the event loop to do more Python parsing. I would be adding a loop, cancellation semantics, and a new way to leak a generator, without changing the freshness bound. I reopen the question when the worker's profile shows many concurrent network waits. That program is an HTTP ingest, and it belongs in a later lesson with its own retry policy.

Q6. How do you keep a contract rejection out of the logs as personal data?

The log line carries batch id, line number, and error code. The exception type's message is written, in lesson 4, so it names the field and not the value. I test with assertLogs and a sentinel email planted in the bad row, and I assert the sentinel is absent from every record. I do not attach the raw line as extra. I do not log at info per row. A summary count at the end is enough for a healthy file. The on-call engineer can open the line in a restricted tool if they must. The log platform is not that tool.

Q7. What belongs in pyproject.toml for this worker, and what does not?

The project name, a Python requirement of 3.11 or newer, a version the process logs at startup, and a console script pointing at main. A src layout so tests import the installed package. No pandas, no cloud SDK, no orchestrator client, until a later lesson has a reason and a reviewer. Dependencies are not a wishlist. I also want main to return a status code rather than parse a file at import. The package version is the checker version, not the metric definition version. Those numbers change for different reasons.

Q8. How would you test that a failed batch does not publish?

A temporary directory, a two-line file whose second line reuses event_id with a different event_type, and an assertion that commit_batch returns rejected and that the .accepted path does not exist. A second test writes a clean file twice and asserts one line of output and a skip. I also want a test that an exception mid-write unlinks the temp file. I do not point tests at a shared development bucket. If the test needs a network, it is not this unit.

Q9. A crash leaves the lease in-progress. What should the next run do?

Not start a second writer beside a possibly living one. Check the process or the marker age. If the owner is demonstrably dead and the age exceeds a documented threshold, take over, log the takeover, and write to a new temp file. If you cannot prove the owner is dead, page a human. Stealing a live lease is how the double Tuesday returns. The threshold is configuration, tested with an injected clock or an explicit break in unit tests, not with time.sleep in the suite.

Q10. Why is os.replace part of a data lesson rather than a trivia item about files?

Because publication is a visibility boundary. Lesson 2 required consumers to see a whole snapshot or the previous one. On a single machine, replacing a file on the same filesystem is the mechanism that provides that boundary for the raw batch artifact. Writing in place, or appending, lets a reader observe a partial batch. os.replace does not make a distributed transaction with the warehouse. It does make the local commit honest. The warehouse load must still key off the committed name, not the temp directory.

Q11. How does the GIL change a design review answer about "just parallelize the parser"?

I ask what is CPU-bound in Python and what is waiting. Bytecode parsing of JSON in CPython is largely serialized by the GIL across threads, so eight threads on one file can be slower and can race the index. Processes escape the GIL and do not share the dict, which is what I want if I parallelize at all. The review answer is a measurement and a private index per process, or no parallelism. "Just parallelize" without naming the shared state is how uniqueness becomes occasional.

Q12. What should an orchestrator be allowed to assume about this worker?

A command exists, it exits 0 when the batch is committed or safely skipped, 2 when the file is rejected, and 1 when the lease is busy or the environment is wrong. It will not print payloads. It will not leave a partial published file. It will not bump the metric version. The orchestrator owns schedule, retries of exit 1, and paging. It should not retry exit 2 in a tight loop: the file is wrong until a human or an upstream extract changes it. That split is the lesson 3 page list, expressed as status codes.

Hands-On Exercises

Exercise: An idempotent streaming ingest

Problem statement. Implement the worker in this lesson against the lesson 4 contract. You may simplify the lease, but you must not weaken type checks to make tests pass.

Tasks.

1. Stream NDJSON with a generator of outcomes. Do not build a list of all payloads.
2. Commit with a temp file and os.replace. A rejected file leaves no .accepted output.
3. Make a second run with the same batch id and bytes a skip. Make a different checksum for the same id a rejection.
4. Add a context manager for the lease or for the file, and a test that an exception does not leave the temp file.
5. Log a rejection with code and line number. Include a sentinel email in a bad fixture and assert it is absent from logs.
6. Add the conflict test and the duplicate-delivery test from the walkthrough.
7. Write a short note refusing threads on a shared index, and the measurement that would justify a process pool across files.
8. Pin Python >=3.11 in pyproject.toml and show that importing the package does not read a data file.

Expected deliverables. The package, the tests and their output, one log excerpt, and the concurrency note.

Evaluation rubric.

You pass when a reviewer can run the suite twice and the second run does not change the accepted bytes, and when a conflict fixture leaves the output directory without that batch's published file.

Lesson Summary

The contract from lesson 4 survives contact with a real file only if the process streams lines, releases handles, publishes by replacement, and treats a repeated batch id as a decision rather than as another append. Generators give you the stream. Context managers give you the release. The logging module gives you a severity channel that must not become a second copy of the payload. Packaging and unittest give the next person a command and a regression test instead of a script named final.

Concurrency is a measured exception. One process owns one uniqueness index. Threads that share that index threaten the rule the parser exists to enforce. Processes across independent files are allowed when a profile says the sequential worker misses the bound. Asyncio is the wrong tool for this disk.

The worker's exit codes are the handoff to whatever scheduler you adopt later. Lesson 6 takes rows the worker has accepted and asks a different question, in SQL: once the events are in a table, which joins and aggregates preserve the grain, and which ones quietly multiply it.

Further Reading

• Python documentation: yield expressions and generator types — laziness, close, and the single-pass contract.
• Python documentation: context managers and contextlib — __exit__ on the failure path.
• Python documentation: concurrent.futures — process pools, and why callables must be importable.
• Python documentation: logging — loggers, levels, and why libraries do not call basicConfig.
• Python packaging user guide — pyproject.toml and console scripts.
• PEP 405 if you want the background for virtual environments the worker should be installed into.
• CPython's global interpreter lock is documented in the Python/C API glossary and threading notes. Read the definition before you promise a thread pool will parse faster.
• Primer: /learn/python. Continue this course: /learn/data-engineering/sql-fundamentals-for-data-engineers. Orchestration that may call this worker starts at /learn/data-engineering/data-pipeline-orchestration-fundamentals.