Your personal operating system for career transition.

Create a Streaming Document Processor with Generators

Loading 10GB of documents into memory before processing them is the fastest path to an OOM kill. Generator pipelines let you process arbitrarily large corpora in constant memory by pulling one record at a time through each stage.

What to build

Implement a composable generator pipeline for document ingestion:

1. read_jsonl(path: Path) -> Iterator[dict] — yields one parsed dict per line, skipping malformed lines with a warning (do not abort).
2. validate_and_clean(records: Iterator[dict]) -> Iterator[dict] — yields only records with non-empty id and body fields (min 50 chars). Strips whitespace from body.
3. chunk_document(doc: dict, max_chars: int) -> Iterator[dict] — yields chunk dicts {doc_id, chunk_index, text, char_start} by splitting body at max_chars boundaries.
4. stream_chunks(docs: Iterator[dict], max_chars: int) -> Iterator[dict] — composes chunk_document over a stream of docs using yield from.
5. write_jsonl(records: Iterator[dict], path: Path) -> int — writes each record as a JSONL line, returns total count written.
6. run_pipeline(input_path: Path, output_path: Path, chunk_size: int) -> dict — wires all stages together, returns {"chunks_written": int, "parse_errors": int, "docs_skipped": int}.

Why this matters

Generator pipelines are the standard pattern for large-scale document ingestion. Understanding how to compose them lets you build ETL workflows that scale to any corpus size without changing the architecture.

Constraints

• No list() wrapping of intermediate iterators — keep every stage lazy.
• Failures at one stage must not abort downstream stages; capture counts instead.
• Use yield from in stream_chunks for clean composition.