Streaming Large CSVs with pandas: A Chunked Aggregation Architecture

When read_csv Runs Out of Memory

When a CSV file grows larger than available RAM, pd.read_csv() fails with a MemoryError before any analysis begins. The useful takeaway is that pandas can process these files without loading them whole: stream the file in chunks, validate each chunk against a strict schema, aggregate incrementally, and checkpoint results to Parquet. This architecture keeps memory bounded and makes failures auditable.

Requirements for a Single-Node pandas Pipeline

Before choosing a design, define the constraints:

• Memory ceiling: Peak process memory must stay below a fixed limit, regardless of file size.
• Schema safety: New or missing columns in the input must stop the job, not silently corrupt aggregates.
• Auditability: Start time, end time, chunk count, and exceptions must be logged.
• Correctness: Incremental results must match a one-shot aggregation on a representative subset.

The Smallest Suitable Design

The minimal pipeline has four stages: chunked ingestion, a validation gate, incremental aggregation, and a Parquet checkpoint. The implementation below relies on long-standing pandas behavior available in 1.x and 2.x: read_csv with chunksize, groupby().agg(), and to_parquet with the PyArrow engine.

import pandas as pd
import logging
import time
from contextlib import contextmanager

logger = logging.getLogger(__name__)

REQUIRED_COLUMNS = {"user_id", "event_type", "amount", "timestamp"}
EXPECTED_DTYPES = {
    "user_id": "int64",
    "event_type": "category",
    "amount": "float64",
    "timestamp": "datetime64[ns]",
}

def validate_chunk(df: pd.DataFrame, chunk_idx: int) -> pd.DataFrame:
    """Strict schema gate: raises on drift before any state mutation."""
    missing = REQUIRED_COLUMNS - set(df.columns)
    extra = set(df.columns) - REQUIRED_COLUMNS
    if missing or extra:
        raise ValueError(
            f"Chunk {chunk_idx}: schema drift (missing={missing}, extra={extra})"
        )
    return df.astype(EXPECTED_DTYPES, errors="raise")

@contextmanager
def audit_logger(job_name: str):
    start = time.time()
    logger.info("Starting %s", job_name)
    try:
        yield
    except Exception:
        logger.exception("Job %s failed", job_name)
        raise
    finally:
        logger.info("Finished %s in %.2fs", job_name, time.time() - start)

def run_pipeline(csv_path: str, out_parquet: str, chunk_size: int = 500_000):
    partials = []
    with audit_logger("csv_aggregation"):
        with pd.read_csv(
            csv_path,
            chunksize=chunk_size,
            dtype={"user_id": "int64", "amount": "float64"},
            parse_dates=["timestamp"],
        ) as reader:
            for i, chunk in enumerate(reader):
                clean = validate_chunk(chunk, i)
                partial = (
                    clean.groupby("event_type", observed=True)["amount"]
                    .agg(["sum", "count"])
                )
                partials.append(partial)
                # Checkpoint every 10 chunks so a crash does not lose all work
                if i % 10 == 0:
                    pd.concat(partials).groupby(level=0).sum().to_parquet(
                        f"{out_parquet}.partial.parquet", engine="pyarrow"
                    )
        result = pd.concat(partials).groupby(level=0).sum()
        result.to_parquet(out_parquet, engine="pyarrow")
    return result

Run this from a shell or scheduler with read access to <csv_path> and write access to the directory containing <out_parquet>. The chunk_size placeholder controls the trade-off between I/O overhead and memory pressure; 500,000 rows is a common starting point for wide tables, but tune it against your row width.

Trust and Data Boundaries

Treat the input CSV as untrusted. Its column set, ordering, and dtypes may change without notice. The validation function is the only component allowed to touch raw chunks; it enforces an allowlist of columns and casts to explicit dtypes with errors="raise". Once a chunk passes, it is trusted in memory. The final Parquet file is the trusted output: it carries explicit types and can be consumed downstream without re-validating the original CSV.

Operational Checks

A practical way to verify the memory bound is to profile the job. On a machine with memory_profiler installed, run:

python -m memory_profiler pipeline.py

Compare peak memory against the chunk size. If peak memory scales with file size rather than chunk size, the aggregation state—not the reader—is the bottleneck.

To verify correctness, run the pipeline on a small subset (for example, the first 100,000 rows) and compare the result against a one-shot pd.read_csv() followed by the same groupby().agg(). The two outputs should match exactly.

To verify schema safety, inject a CSV with an extra column and confirm the job raises ValueError before any Parquet file is written. To verify corruption handling, include a malformed line; read_csv will raise by default, and the context manager should log the exception and abort.

Failure Modes and Conditions That Change the Design

• High-cardinality groupby keys: If event_type had millions of unique values, the partials list would grow until it exhausted memory. Pre-filter to relevant keys or move the aggregation into a database.
• Chunk-size miscalculation: A chunk size that is too large reproduces the original MemoryError. One that is too small makes I/O overhead dominate. Measure, do not guess.
• Thread safety: pandas DataFrames are not thread-safe for concurrent writes. Do not share a DataFrame across threads; use separate DataFrames per process or rely on the single-threaded chunk loop.
• Schema drift without validation: Skipping the gate lets new columns silently alter aggregation semantics.

When the dataset grows beyond a single node, or when partial aggregations no longer fit in memory, switch to Dask DataFrame. It mirrors the pandas API (read_csv, groupby, to_parquet) but partitions work across cores or a cluster. If multiple writers or complex transactional joins are required, a relational database is the safer boundary.

Approach	Best for	Memory model	Operational cost
pandas chunked	Single node, files slightly larger than RAM	Bounded by chunk size	Low
Dask DataFrame	Multi-core or cluster, same pandas API	Distributed partitions	Medium
Database	Shared state, concurrent access, complex joins	Server-managed	High

Limitations

This design assumes a single input file and a single aggregation pass. It does not handle streaming updates, multi-file transactions, or schemas that require row-level parsing beyond pandas' native CSV engine. Feather and Parquet preserve dtypes, but they do not preserve arbitrary external metadata; store that separately if downstream consumers depend on it.