Processing Large Files and NDJSON Streams in Python CLIs

Event exports, audit logs, API dumps and log archives increasingly arrive as NDJSON — newline-delimited JSON, one object per line, often gzipped and often far larger than the laptop it is being inspected on. A CLI that handles them with json.load(f) or f.read().splitlines() works on the sample file and then dies on the real one, taking several times the file's size in memory before it prints anything. The alternative costs nothing in code: read one line, handle it, write the result, repeat. Memory stays flat whatever the input size, the first results appear immediately, and the command composes with head, grep and jq in a pipeline. This guide builds a small filter command that streams NDJSON from files, .gz archives or stdin, reports malformed lines precisely, offers an explicit lenient mode, and proves with a test that its memory use does not grow with the input. It belongs to the working with stdin, stdout and pipes topic.

Prerequisites

• Python 3.10+ and a Typer or Click CLI.
• Familiarity with reading piped input in Python CLIs.

Stream, do not load

A Python file object is an iterator over lines, reading from disk in buffered chunks. for line in fh therefore holds one line in memory at a time, however big the file. Parse it with json.loads, decide whether it matches, write the result straight to stdout, and let the object go. Every stage of the command is a generator or a loop, and nothing ever builds a list of all records.

The difference is not a percentage but orders of magnitude. Loading a file as text costs its size again as a Python string; parsing everything into dictionaries costs several times more, because every key, value and dictionary carries object overhead. Streaming costs the size of the longest line plus buffers. The test at the end of this guide measures it: a peak of about 25 KB while filtering an 18 MB file.

The recipe

The streaming logic lives in a module that knows nothing about the CLI framework. open_input handles the three kinds of input, records parses lines lazily and deals with bad ones, and filter_records connects them to an output stream:

# src/mytool/stream.py
from __future__ import annotations

import gzip
import io
import json
import sys
from collections.abc import Iterator
from contextlib import contextmanager
from dataclasses import dataclass, field
from pathlib import Path
from typing import IO, Any


class BadRecord(Exception):
    def __init__(self, source: str, line_no: int, reason: str) -> None:
        super().__init__(f"{source}:{line_no}: {reason}")
        self.source, self.line_no, self.reason = source, line_no, reason


@dataclass
class Stats:
    read: int = 0
    written: int = 0
    skipped: int = 0
    first_errors: list[str] = field(default_factory=list)


@contextmanager
def open_input(name: str) -> Iterator[IO[str]]:
    """'-' is stdin; .gz files are decompressed on the fly. Text, UTF-8, read lazily."""
    if name == "-":
        fh = io.TextIOWrapper(sys.stdin.buffer, encoding="utf-8")
        try:
            yield fh
        finally:
            fh.detach()                     # leave the real stdin open
    elif name.endswith(".gz"):
        with gzip.open(name, "rt", encoding="utf-8") as fh:
            yield fh
    else:
        with Path(name).open(encoding="utf-8") as fh:
            yield fh


def records(fh: IO[str], source: str, *, skip_invalid: bool, stats: Stats) -> Iterator[dict[str, Any]]:
    for line_no, line in enumerate(fh, start=1):
        if not line.strip():
            continue
        stats.read += 1
        try:
            obj = json.loads(line)
            if not isinstance(obj, dict):
                raise ValueError("expected a JSON object")
        except ValueError as exc:          # json.JSONDecodeError is a ValueError
            if not skip_invalid:
                raise BadRecord(source, line_no, str(exc)) from None
            stats.skipped += 1
            if len(stats.first_errors) < 5:
                stats.first_errors.append(f"{source}:{line_no}: {exc}")
            continue
        yield obj


def filter_records(inputs: list[str], out: IO[str], *, where: dict[str, Any],
                   fields: list[str] | None, skip_invalid: bool) -> Stats:
    stats = Stats()
    for name in inputs:
        with open_input(name) as fh:
            for obj in records(fh, name, skip_invalid=skip_invalid, stats=stats):
                if all(obj.get(k) == v for k, v in where.items()):
                    if fields:
                        obj = {k: obj.get(k) for k in fields}
                    out.write(json.dumps(obj, separators=(",", ":")) + "\n")
                    stats.written += 1
    return stats

The command is a thin wrapper that parses options, calls the streaming function with sys.stdout, and turns problems into messages and exit codes:

# src/mytool/cli.py
from __future__ import annotations

import sys
from typing import Annotated

import typer

from mytool.stream import BadRecord, filter_records

app = typer.Typer()


@app.callback()
def main() -> None:
    """Work with NDJSON event files."""


@app.command("filter")
def filter_cmd(
    inputs: Annotated[list[str], typer.Argument(help="NDJSON files (.gz ok); '-' for stdin.")] = None,
    where: Annotated[list[str], typer.Option("--where", help="KEY=VALUE to match.")] = None,
    fields: Annotated[str, typer.Option("--fields", help="Comma-separated keys to keep.")] = "",
    skip_invalid: Annotated[bool, typer.Option("--skip-invalid", help="Skip malformed lines.")] = False,
) -> None:
    """Stream matching records to stdout, one JSON object per line."""
    conditions = dict(w.split("=", 1) for w in (where or []))
    try:
        stats = filter_records(inputs or ["-"], sys.stdout, where=conditions,
                               fields=[f for f in fields.split(",") if f] or None,
                               skip_invalid=skip_invalid)
    except BadRecord as exc:
        typer.echo(f"error: {exc} (use --skip-invalid to skip bad lines)", err=True)
        raise typer.Exit(65) from None          # EX_DATAERR
    if stats.skipped:
        typer.echo(f"skipped {stats.skipped} of {stats.read} lines:", err=True)
        for line in stats.first_errors:
            typer.echo(f"  {line}", err=True)

In use, it behaves like any other filter in a pipeline:

mytool filter events.ndjson.gz --where level=error --fields ts,user | head -20
kubectl logs deploy/api | mytool filter --where level=error
mytool filter a.ndjson b.ndjson.gz - < c.ndjson > errors.ndjson

Why it is built this way

Generators all the way down. records is a generator, so parsing happens only as fast as the consumer asks for records. When head -20 has what it needs and closes the pipe, the command stops reading — a 20 GB file costs the same as a 20 KB one. Handling the resulting broken pipe cleanly is covered in handling broken pipe and SIGPIPE.

Explicit UTF-8, and stdin in binary. NDJSON is UTF-8 by definition, so the files are opened with encoding="utf-8" rather than the platform default, which differs on Windows. Stdin is wrapped from its binary buffer for the same reason, and detached afterwards so the real stdin is not closed.

Compression is transparent. gzip.open(..., "rt") decompresses as it reads, so archives never need to be unpacked to disk first. The same pattern extends to .bz2, .xz and, with a third-party package, .zst.

Compact output. separators=(",", ":") writes each object without spaces, which matters when the output is gigabytes, and keeps one object per line so the output is NDJSON too — ready for the next command in the pipeline.

Blank lines are ignored. Many producers end files with an empty line or separate batches with one; treating those as errors would make the tool fussy for no benefit.

Bad records: stop or skip?

A malformed line in the middle of a large file is common: a truncated write, a log line that is not JSON, a stray array. The default here is to stop with the file name and line number and exit 65 (EX_DATAERR), because silently dropping records is how data goes missing without anybody noticing. When the user knows the input is messy and wants the good records anyway, --skip-invalid skips bad lines, counts them, and reports the count and the first few locations on stderr so the output stays clean. Leniency is opt-in and visible — never the default.

UX considerations

• Default to stdin. With no file arguments the command reads stdin, and - means stdin among other files, following the conventions in following POSIX and GNU argument conventions.
• Results on stdout, everything else on stderr. The skip summary and errors go to stderr, so > out.ndjson captures only records. Output rules are covered in emitting JSON output for scripting.
• Progress only on a terminal. For long runs over files, a byte-based progress bar on stderr helps — but only when stderr is a TTY; see detecting TTY and adapting output. Stdin has no known length, so show a record count instead.
• Faster parsing is a drop-in. If profiling shows json.loads dominating, orjson.loads is several times faster and accepts the same lines; keep the streaming structure and swap only the parser.
• Parallelism rarely pays. Reading and parsing a stream is usually I/O-bound or bottlenecked on one core's JSON parsing; splitting a file across processes adds complexity and reorders output. Measure first, as in multiprocessing for CPU-bound CLI tasks.

Testing the behaviour

The tests cover filtering and projection, stdin and gzip input together, both error modes, laziness and — the property that matters most — flat memory, measured with tracemalloc on an 18 MB file:

# tests/test_stream.py
import gzip
import io
import json
import tracemalloc

import pytest
from typer.testing import CliRunner

from mytool.cli import app
from mytool.stream import BadRecord, Stats, filter_records, records

runner = CliRunner()
EVENTS = [{"level": "info", "user": "ana"}, {"level": "error", "user": "bo"}, {"level": "error", "user": "cy"}]
NDJSON = "".join(json.dumps(e) + "\n" for e in EVENTS)


def test_filters_and_projects(tmp_path):
    path = tmp_path / "events.ndjson"
    path.write_text(NDJSON)
    result = runner.invoke(app, ["filter", str(path), "--where", "level=error", "--fields", "user"])
    assert result.exit_code == 0
    assert result.stdout.splitlines() == ['{"user":"bo"}', '{"user":"cy"}']


def test_reads_stdin_and_gzip(tmp_path):
    gz = tmp_path / "events.ndjson.gz"
    with gzip.open(gz, "wt", encoding="utf-8") as fh:
        fh.write(NDJSON)
    result = runner.invoke(app, ["filter", "-", str(gz), "--where", "user=ana"], input=NDJSON)
    assert result.stdout.count('"ana"') == 2


def test_bad_line_stops_with_its_line_number(tmp_path):
    path = tmp_path / "bad.ndjson"
    path.write_text(NDJSON + "{not json\n")
    result = runner.invoke(app, ["filter", str(path)])
    assert result.exit_code == 65
    assert f"{path}:4:" in result.stderr


def test_skip_invalid_counts_and_reports(tmp_path):
    path = tmp_path / "bad.ndjson"
    path.write_text("[1, 2]\n" + NDJSON + "{oops\n")
    result = runner.invoke(app, ["filter", str(path), "--skip-invalid"])
    assert result.exit_code == 0 and len(result.stdout.splitlines()) == 3
    assert "skipped 2 of 5 lines" in result.stderr


def test_records_is_lazy():
    stats = Stats()
    gen = records(io.StringIO('{"a": 1}\n{broken\n'), "x", skip_invalid=False, stats=stats)
    assert next(gen) == {"a": 1}            # the bad second line has not been read yet
    with pytest.raises(BadRecord):
        next(gen)


def test_memory_stays_flat(tmp_path):
    path = tmp_path / "big.ndjson"
    with path.open("w") as fh:
        for i in range(200_000):
            fh.write(json.dumps({"i": i, "level": "info", "msg": "x" * 50}) + "\n")
    tracemalloc.start()
    filter_records([str(path)], io.StringIO(), where={"level": "none"}, fields=None, skip_invalid=False)
    _, peak = tracemalloc.get_traced_memory()
    tracemalloc.stop()
    assert path.stat().st_size > 15_000_000
    assert peak < 1_000_000                 # measured: about 25 KB for an 18 MB file

test_records_is_lazy proves the generator yields the first record before it has read the broken second line; a version that parsed everything up front would raise immediately. test_memory_stays_flat is the regression test for "someone added list(...) for convenience": a change that loads the file would push the peak into tens of megabytes and fail. It writes 200,000 records, so it takes about a second — mark it as slow if your suite is strict about speed.

Conclusion

Handling large inputs in a CLI is a matter of never holding more than one record: iterate over lines, parse each with json.loads, write results immediately, and let generators connect the stages so the command stops as soon as its consumer does. Open files with explicit UTF-8, decompress .gz on the fly, treat - and no arguments as stdin, stop on malformed lines with a file and line number by default, and make skipping explicit with a counted summary. A tracemalloc test keeps memory flat through future changes, and the command stays usable on files far larger than the machine's memory.

Frequently asked questions

Related

• Up: Working with stdin, stdout and pipes
• Reading piped input in Python CLIs
• Handling broken pipe and SIGPIPE
• Emitting JSON output for scripting
• Downloading files with progress in Python