Operational Context

A Python automation engineer schedules a nightly job to synchronize a legacy Collection Management System export into a normalized database, and watches the process die halfway through with a MemoryError or a silent OOM kill on a constrained CI runner. The export is a single 1–5 GB CSV carrying millions of object records — accession numbers, controlled-vocabulary terms, multi-line provenance narratives, OCR-extracted metadata, and IIIF manifest URLs. Resident memory climbs linearly with rows read, garbage collection begins thrashing, and the job stalls before a single batch commits. This page resolves that exact failure: converting an eager, load-everything import into a strictly streaming pipeline whose peak memory is bounded by batch size rather than file size. It is the high-volume variant of the async-first patterns in the parent CSV to database sync strategies, and it feeds the same validated-record contract the rest of the automated record ingestion pipeline depends on.

Root Cause Analysis

The dominant cause is not exotic: holding every parsed row in one in-memory list keeps all of those string objects alive for the entire run, so the heap grows in proportion to file size. Python frees an object only when its reference count reaches zero, and appending each row to a long-lived list guarantees that never happens until the list itself is dropped. Layer the cyclic garbage collector’s repeated sweeps over an ever-larger object graph on top, and the result is the heap growth and lengthening GC pauses that exhaust RAM.

Three secondary vectors amplify it. First, eager evaluation through a dataframe library reads the whole CSV into contiguous column arrays and infers dense float64/object dtypes, allocating far more than the sparse museum data warrants — which is why this pipeline avoids those libraries at the ingestion boundary. Second, row-by-row INSERT statements accumulate cursor state, per-statement type coercion, index updates, and Write-Ahead Log flushes, so the database side leaks state in lockstep with the reader. Third, schema drift — malformed dates, mixed-language encodings, embedded base64 thumbnails — forces repeated coercion attempts that fragment the heap. The fix targets the primary cause directly: never let more than one batch of rows be reachable at once, and enforce that boundary the same way schema validation with Pydantic enforces record contracts elsewhere in the pipeline.

Canonical Solution

Resolving the leak means restructuring the import around three invariants: lazy iteration so rows are pulled one at a time, a fixed-size batch that is validated and then handed off and dereferenced, and bulk insertion so the database sees one parameterized statement per batch instead of one per row. The diagram below shows how those pieces compose into a single constant-memory cycle.

Constant-memory streaming ingestion cycle Large CSV feeds DictReader (lazy, one row), which feeds a Validate decision on LIDORecord. The invalid branch drops down to a Quarantine log; the valid branch flows right into a fixed-size batch, then a bulk execute_values insert, then the Database. A return path runs from the bulk insert back to DictReader labelled "pull next batch — previous batch freed", closing the loop so only one batch is live at a time. valid invalid pull next batch — previous batch dereferenced & freed Large CSV 1–5 GB, streamed DictReader lazy · one row Validate LIDORecord Fixed-size batch ~2,000 rows only one live Bulk insert execute_values Database Quarantine log malformed rows

Step 1: Stream Rows and Yield Fixed-Size Batches

Use the stdlib csv.DictReader for lazy iteration — it pulls one row at a time off the file handle instead of materializing the file. Open the file inside a with block so the descriptor is always released, and decode as utf-8-sig to absorb the UTF-8 BOM that legacy museum exporters routinely emit. Validate each row at the boundary through a Pydantic v2 model, keep only the normalized dict, and drop the raw row immediately. When the batch fills, yield it and rebind batch to a fresh list — clearing the existing list in place would mutate the object already handed to the consumer.

python
import csv
import logging
from pathlib import Path
from typing import Iterator
from pydantic import BaseModel, Field, ValidationError

logger = logging.getLogger(__name__)

class LIDORecord(BaseModel):
    accession: str = Field(pattern=r"^[A-Z]{2,4}\.\d{4}\.\d{1,6}$")
    title: str
    iiif_manifest: str = Field(pattern=r"^https?://.*manifest\.json$")
    lido_category: str

def stream_batches(csv_path: Path, batch_size: int = 2000) -> Iterator[list[dict]]:
    with csv_path.open("r", encoding="utf-8-sig") as fh:
        reader = csv.DictReader(fh)          # lazy: one row pulled per iteration
        batch: list[dict] = []
        for row in reader:
            try:
                # Validate at the boundary; retain only the normalized dict so
                # the raw row (and its strings) become collectable immediately.
                batch.append(LIDORecord(**row).model_dump())
            except ValidationError as exc:
                logger.warning("Quarantined malformed row: %s", exc)
                continue
            if len(batch) >= batch_size:
                yield batch
                batch = []                   # rebind, do NOT clear the yielded list
        if batch:
            yield batch                      # flush the final partial batch

A batch size of 1,000–5,000 rows balances throughput against peak RAM; 2,000 is a safe default. Because the previous batch reference drops out of scope on each yield, the collector reclaims it before the next batch fills, and resident memory plateaus regardless of whether the file is 200 MB or 20 GB.

Step 2: Commit Each Batch in Bulk

Send each validated batch to the database as a single statement with psycopg2.extras.execute_values (or SQLAlchemy insert().values()), not one INSERT per row. This bypasses per-row cursor allocation, collapses network round-trips, and lets the database perform type coercion natively. Wrap each batch in an explicit transaction, commit on success, and roll back and quarantine the batch on a constraint violation so a single bad row never poisons the run. Committing per batch is what keeps the WAL from accumulating and holds the connection pool stateless between commits — the durable-queue extension of this is documented for Celery-based museum data sync.

python
import psycopg2
from psycopg2.extras import execute_values

INSERT_SQL = """
    INSERT INTO objects (accession, title, iiif_manifest, lido_category)
    VALUES %s
    ON CONFLICT (accession) DO UPDATE
        SET title = EXCLUDED.title,
            iiif_manifest = EXCLUDED.iiif_manifest,
            lido_category = EXCLUDED.lido_category
"""

def commit_batch(conn, batch: list[dict]) -> None:
    rows = [
        (r["accession"], r["title"], r["iiif_manifest"], r["lido_category"])
        for r in batch
    ]
    with conn:                               # transaction: commit on exit, rollback on error
        with conn.cursor() as cur:
            # page_size caps how many rows go in one round-trip VALUES list.
            execute_values(cur, INSERT_SQL, rows, page_size=len(rows))

def ingest(csv_path: Path, dsn: str, batch_size: int = 2000) -> None:
    conn = psycopg2.connect(dsn)
    try:
        for batch in stream_batches(csv_path, batch_size):
            commit_batch(conn, batch)        # one batch reachable at a time
    finally:
        conn.close()

The ON CONFLICT clause makes the import idempotent, so a resumed or re-run job upserts rather than duplicating — the same guarantee the broader automated record ingestion pipeline relies on for safe reruns.

Edge Cases and Variants

  • XML or API input instead of CSV. Swap csv.DictReader for xml.etree.ElementTree.iterparse (calling elem.clear() after each record to release parsed nodes) or for a paged generator over an API cursor. The batch/validate/bulk-commit tail is unchanged — only the row source differs. The API variant is worked through in polling museum APIs with Python Requests.
  • Single large JSON array. Never json.load a multi-gigabyte array; use an incremental parser such as ijson.items(fh, "item") so memory stays flat.
  • Strict vs. lenient validation. In a backfill, catch ValidationError, attach the raw row, and route it to a durable quarantine table; in an interactive re-sync, surface the error to the operator instead of silently skipping.
  • Embedded base64 thumbnails. Strip binary payloads out of the row before validation and write them to object storage, keying by accession — buffering base64 blobs in the batch defeats the memory budget the batch size was chosen for.
  • Horizontal scaling. Partition the export by accession range or collection and hand each partition to a separate multiprocessing.Pool worker with its own connection; isolated address spaces sidestep GIL contention and let database writes proceed in parallel. Match the connection-pool ceiling to worker count and tune work_mem / maintenance_work_mem for the bulk index updates.
  • When gc.collect() actually helps. Scope exit reclaims batches on its own; only force a collection (e.g. every tenth batch) if tracemalloc shows fragmentation creeping up, and never reach for manual del on the batch list as a substitute for rebinding it.

Validation

Prove the two invariants that keep the import bounded — malformed rows are quarantined rather than committed, and peak memory is independent of file size — with an assert-based test and a tracemalloc probe that need no live database:

python
# test_stream_batches.py  ->  run with:  pytest -q test_stream_batches.py
import csv, tracemalloc
from pathlib import Path

def _write_csv(path: Path, n: int) -> None:
    with path.open("w", encoding="utf-8") as fh:
        w = csv.writer(fh)
        w.writerow(["accession", "title", "iiif_manifest", "lido_category"])
        for i in range(n):
            w.writerow([f"AB.2024.{i}", f"Object {i}",
                        "https://example.org/manifest.json", "painting"])
        w.writerow(["not-an-accession", "Bad row", "nope", "painting"])  # invalid

def test_invalid_rows_are_dropped(tmp_path):
    csv_path = tmp_path / "export.csv"
    _write_csv(csv_path, 5)
    kept = [r for batch in stream_batches(csv_path, batch_size=2) for r in batch]
    assert len(kept) == 5                         # the malformed row never made it through
    assert all(r["accession"].startswith("AB.2024.") for r in kept)

def test_peak_memory_is_bounded_by_batch(tmp_path):
    small, large = tmp_path / "s.csv", tmp_path / "l.csv"
    _write_csv(small, 1_000)
    _write_csv(large, 200_000)
    def peak(path):
        tracemalloc.start()
        for _ in stream_batches(path, batch_size=2000):
            pass
        p = tracemalloc.get_traced_memory()[1]
        tracemalloc.stop()
        return p
    # 200x the rows must NOT cost anything close to 200x the peak memory.
    assert peak(large) < peak(small) * 5

A green run demonstrates that the invalid row is quarantined instead of committed and that a 200x larger file stays within a small constant multiple of the smaller file’s peak — the signature of a constant-footprint stream rather than an eager load.

Peak memory: eager load versus streaming pipeline Two curves over the same horizontal axis of rows processed. The eager curve rises linearly from the origin to the top-right and terminates at a MemoryError burst. The streaming curve holds a flat low sawtooth near the baseline across the whole run, illustrating that peak memory is bounded by batch size rather than file size. rows processed → peak resident memory → MemoryError eager load — heap grows with file size streaming — peak bounded by batch size

Standards Alignment

The streaming boundary is also where records are shaped for downstream conformance. The LIDORecord fields map onto a LIDO object record — accession to the persistent identifier, title to lido:titleSet, and lido_category onto the object classification that reconciling against Getty AAT and TGN upgrades to stable authority URIs. Each iiif_manifest must resolve to a manifest conforming to the IIIF Presentation API 3.0, whose id is a dereferenceable HTTP(S) URI, which is exactly what the manifest\.json pattern guards at ingestion time. Persisting source-file hashes, validation timestamps, and commit offsets alongside the upsert preserves the audit trail accreditation reviews expect, and the normalized dicts drop cleanly into the LIDO-to-database mapping layer without post-processing.

Frequently Asked Questions

Why does appending rows to a list cause the leak rather than the CSV reader itself?

csv.DictReader is already lazy — it holds only the current row. The leak comes from the consumer: every row you append to a long-lived list stays referenced, so its reference count never hits zero and the collector cannot reclaim it. Bounding the list to one batch and rebinding it on each yield is what caps the live object set and flattens the memory curve.

What batch size should I use?

Start at 2,000 rows and adjust from measurement, not intuition. Smaller batches lower peak RAM but add per-batch commit overhead and round-trips; larger batches improve throughput until the batch itself dominates memory or the VALUES list strains the driver. Profile with tracemalloc and pick the largest batch whose peak stays within your runner’s budget.

Can I just use a dataframe library with a chunk size instead?

Chunked dataframe reads do bound memory, but at the ingestion boundary they still infer dense dtypes and allocate column arrays the sparse museum data does not need, and they obscure per-row validation. Streaming with DictReader plus a Pydantic model keeps the footprint minimal and puts a strict, auditable validation gate exactly where malformed rows must be caught.

How do I resume a run that died partway through?

Make the insert idempotent with ON CONFLICT ... DO UPDATE and persist the last committed row offset (or accession) after each batch. On restart, skip to that offset; because upserts replace rather than duplicate, re-running an already-committed batch is harmless.