streamval

Streaming, Pydantic-backed validation for CSV, JSONL, Parquet, Arrow, and HTTP NDJSON / SSE.

Existing data-validation libraries (Pydantic, Pandera, Great Expectations, Cerberus) all assume the dataset fits in memory. streamval keeps the file (or HTTP response) on disk / on the wire and validates it row by row through a Pydantic schema, so you can validate a multi-gigabyte file with a few tens of megabytes of RAM and start consuming valid rows immediately. The same streaming model handles LLM token streams, log services, and any REST endpoint that emits NDJSON or Server-Sent Events.

Install

pip install streamval
# faster JSON + lazy CSV via polars/orjson:
pip install "streamval[fast]"
# HTTP NDJSON / LLM streaming via httpx:
pip install "streamval[http]"
# everything:
pip install "streamval[fast,http]"

Quickstart

from pydantic import BaseModel
from streamval import stream_csv

class User(BaseModel):
    id: int
    name: str
    score: float
    active: bool

for result in stream_csv("users.csv", User, on_error="collect"):
    if result.valid:
        user = result.data
        # ... do something with the parsed model ...
    else:
        for err in result.errors:
            print(f"row {result.row_index}: {err}")

The generator finishes when the file ends. Stats are available on the underlying validator:

from streamval import StreamValidator
v = StreamValidator(User, on_error="skip", batch_size=2000)
for r in v.stream_csv("users.csv"):
    handle(r.data)
print(v.stats)  # rows_total, rows_valid, throughput_rps, peak_memory_mb, ...

Performance

streamval optimises for bounded memory with strong throughput as a secondary goal. Throughput, 100 000 rows, 4-column schema:

On the naive loop: it stays the fastest option for files that fit comfortably in RAM, because streamval does strictly more work per row — it allocates a ValidationResult carrying the row index, the raw row, and any field errors. That per-row object is the feature. streamval is the right choice when files don't fit in memory, when you want valid rows immediately, or when you need per-row error reporting rather than a single exception.

Numbers from a developer Windows laptop with Python 3.13, pydantic 2.11, polars 1.40. Run STREAMVAL_BENCH=1 pytest tests/benchmarks/ to measure on your own machine.

Performance tuning

• Leave track_memory=False (the default). It gates tracemalloc, which hooks every allocation and costs roughly 4-5× throughput. Turn it on only for profiling runs — it is what populates stats.peak_memory_mb, which reads 0.0 while it is off.

• Install streamval[fast] to unlock the polars path for CSV. Parquet gets the Arrow fast path with no extra dependency. On the sync path polars is roughly on par with the stdlib csv reader for simple schemas — measure before adding the dependency for CSV alone.

• use_arrow=True is the default for CSV and Parquet, and is a clear win for Parquet. For CSV the two modes now measure within a few percent of each other; row mode holds a small edge on peak memory at the same batch_size. Benchmark both on your own data.

• CSV batch mode casts whole columns in Arrow rather than converting each cell in Python. Columns that parse cleanly take the vectorised path; a column containing even one unparseable cell falls back to per-row coercion, so results are identical either way — only the speed differs. Common formatting variants are handled column-wise and stay fast: padded cells (" 42 "), worded booleans (yes/no/y/ n/t/f), and integers written as floats ("42.0"). ISO dates are required for the fast path; other date formats still validate, just per row.

• batch_size is the main throughput / memory dial — larger batches mean fewer Python ↔ Rust crossings but proportionally higher peak memory. Measured peak on 1M rows, Arrow batch mode:

  batch_size=100   → ~0.3 MB live, but see below
  batch_size=1000  → ~1.9 MB peak  (default)
  batch_size=5000  → ~9 MB peak
  batch_size=10000 → ~16 MB peak
  

  From batch_size=1000 upward these reproduce to within 0.01 MB across machines. Below that they do not. Once the live working set drops under ~1 MB, the peak is dominated by however much short-lived garbage happens to be uncollected when the high-water mark is taken, which depends on GC timing rather than on batch_size: at batch_size=100 the same code has measured anywhere from 0.26 MB to 8.2 MB across machines and runs. Prefer batch_size=1000 unless you have measured a smaller value on your hardware and workload.

• workers > 1 enables a thread pool. Pydantic's Rust core is thread-safe; per-row ordering is preserved.

Formats

Why not Pydantic / Pandera / Great Expectations?

How it works

• Each format has a tiny async-generator adapter that yields one row dict at a time without loading the whole file.
• A BatchBuffer chunks the row stream into fixed-size lists so peak memory stays bounded by batch_size.
• Each batch is run through a CompiledValidationPlan (a per-model, cached wrapper around model.model_validate).
• A pluggable error strategy (fail_fast, collect, skip) decides whether each row is emitted, dropped, or terminates the run.
• A StatsAccumulator records per-field error counts, throughput, and peak memory via tracemalloc.

Error strategies

• fail_fast — raise StreamValidationError on the first invalid row.
• collect — emit every row; if max_errors is exceeded, raise on finalize.
• skip — drop invalid rows silently (logged at WARNING level).

Documentation

Online docs: streamval.readthedocs.io

Full guides (also in docs/):

Contributing

git clone https://github.com/AmeerTechsoft/streamval
cd streamval
pip install -e ".[dev]"
pytest

License

MIT — see LICENSE.