Streaming JSONL Parser Without Loading the File

from __future__ import annotations
import io
import json
from typing import Iterable, Iterator

# JSONL = one JSON object per line. The whole point of the format is that you
# never need the full file in memory. So: open, iterate, json.loads each line.

def iter_jsonl(stream: Iterable[str]) -> Iterator[dict]:
    for line in stream:
        line = line.strip()
        if not line:
            continue
        yield json.loads(line)

# Demo with an in-memory file. In real code: open('events.jsonl', 'r').
sample = '\n'.join([
    '{"id": 1, "event": "login",  "user": "alice"}',
    '{"id": 2, "event": "click",  "user": "alice", "path": "/home"}',
    '',  # blank line, valid in JSONL streams
    '{"id": 3, "event": "logout", "user": "alice"}',
])

for row in iter_jsonl(io.StringIO(sample)):
    print(row)

from __future__ import annotations
import io
import json
from typing import Iterable, Iterator, Literal, Tuple

# In production the file is never clean. A truncated upload, a control
# character mid-line, a half-flushed log shipper, all produce malformed JSON.
# Decide the policy explicitly instead of crashing the pipeline.

def iter_jsonl_safe(
    stream: Iterable[str],
    on_error: Literal['skip', 'fail', 'count'] = 'skip',
) -> Iterator[Tuple[int, dict]]:
    bad = 0
    for lineno, raw in enumerate(stream, start=1):
        line = raw.strip()
        if not line:
            continue
        try:
            obj = json.loads(line)
        except json.JSONDecodeError:
            if on_error == 'fail':
                raise
            bad += 1
            continue
        yield lineno, obj
    if on_error == 'count' and bad:
        # Reporting policy: log instead of raise so the pipeline finishes.
        print(f'jsonl: skipped {bad} malformed lines')

sample = '\n'.join([
    '{"id": 1, "ok": true}',
    '{"id": 2, "ok":',         # truncated, JSONDecodeError
    'not-json-at-all',           # garbage line
    '{"id": 3, "ok": true}',
])

for lineno, row in iter_jsonl_safe(io.StringIO(sample), on_error='count'):
    print(lineno, row)

from __future__ import annotations
import io
import json
from typing import Iterable, Iterator, List

# Most downstream sinks (a database insert, a Kafka producer, an HTTP API)
# want batches, not single rows. Keep memory bounded by yielding fixed-size
# chunks. The generator lets the consumer pull at its own pace, which gives
# you free backpressure: if the sink is slow, we do not pre-read the file.

def batched_jsonl(
    stream: Iterable[str],
    batch_size: int = 500,
) -> Iterator[List[dict]]:
    buf: List[dict] = []
    for raw in stream:
        line = raw.strip()
        if not line:
            continue
        try:
            buf.append(json.loads(line))
        except json.JSONDecodeError:
            continue
        if len(buf) >= batch_size:
            yield buf
            buf = []
    if buf:
        yield buf

sample = '\n'.join(json.dumps({'id': i, 'tag': 'x' if i % 2 else 'y'}) for i in range(11))

for i, batch in enumerate(batched_jsonl(io.StringIO(sample), batch_size=4)):
    print(f'batch {i}: {len(batch)} rows  first={batch[0]}')