Generators & Lazy Evaluation for Massive Datasets

Using generator functions (yield) and itertools to stream, filter, and batch massive token datasets line-by-line in constant O(1) memory, rather than loading multi-gigabyte JSONL files into RAM.

Interactive Systems Visualizer

Vectorized SIMD Kernel vs Interpreted Bytecode Loop

Array Size:100,000 elements

Pure Python for-loop--

for x in array:
total += x * 2.0

Per-element PyObject type checks, pointer dereferencing, and evaluation loop overhead.

NumPy Vectorized Kernel--

total = array * 2.0 # AVX-512 SIMD

Single C kernel call executing 8-16 parallel floating-point operations per CPU clock cycle.

Key Systems Engineering Principles

Generators produce one item on demand via next() and preserve state between iterations.
Generator pipelines can be composed: parse_jsonl(stream) -> tokenize(stream) -> create_batches(stream).
Keeps working memory footprint under 50MB even when processing 500GB training corpora.

Mental Model Analogy

“A generator is an assembly line conveyor belt delivering one part at a time, rather than a dump truck unloading 100,000 parts onto your living room floor all at once.”

Production Pitfall to Avoid

Calling list(generator) on a 10-million row pretraining dataset, immediately triggering an Out-Of-Memory (OOM) kernel kill.

production_pattern.py

from typing import Iterator

def stream_jsonl(filepath: str) -> Iterator[str]:
    with open(filepath, 'r', encoding='utf-8') as f:
        for line in f:
            yield line.strip() # O(1) memory footprint

def batch_stream(stream: Iterator[str], batch_size=32) -> Iterator[list[str]]:
    batch = []
    for item in stream:
        batch.append(item)
        if len(batch) == batch_size:
            yield batch
            batch = []
    if batch: yield batch

Previous Subtopic#20 Self-Correcting LLM Extraction Loops Next Subtopic #22 Context Managers for Latency & GPU Telemetry

Loading Python for AI Engineering...

from typing import Iterator def stream_jsonl(filepath: str) -> Iterator[str]: with open(filepath, 'r', encoding='utf-8') as f: for line in f: yield line.strip() # O(1) memory footprint def batch_stream(stream: Iterator[str], batch_size=32) -> Iterator[list[str]]: batch = [] for item in stream: batch.append(item) if len(batch) == batch_size: yield batch batch = [] if batch: yield batch