Stream CSV to JSONL with Python: Preserve Strings and Publish Complete Output

Your scheduled task receives CSV, but the next tool expects one JSON object per line. Splitting on commas breaks quoted fields, while loading the whole file adds memory pressure. This tutorial uses Python’s standard library to convert records incrementally, preserving Unicode, leading zeros, and embedded line breaks, then creates the final file only after all checks pass.

It is for beginner-to-intermediate readers using Linux and completed local CSV files that remain unchanged during conversion. No third-party packages are needed.

1. Method and data rules

1. Input is comma-separated UTF-8 CSV, optionally with a leading BOM. The first record is the header. Names must be nonblank and unique; preserve spaces and compare names exactly.
2. Keep every value as a string: 001, true, and 12.50 are not converted automatically. An empty field becomes "". Every record must match the header width; blank records fail too.
3. Write one JSON object and one LF per data record. The JSON encoder escapes embedded line breaks so one CSV record cannot become multiple JSONL records.
4. Empty input fails. Header-only CSV succeeds with a zero-byte output representing zero records. A downstream consumer requiring at least one record should reject empty results separately.

The JSON Lines format description requires UTF-8, one valid JSON value per line, and no BOM. This converter chooses objects for batch processing; JSONL also permits other JSON values.

For required business column names and order, add CSV structure validation first. This tool checks general structure without inferring types or business rules.

2. Prepare the environment and demo data

Use an existing Python 3.10 or newer interpreter. These examples were tested on Linux with Python 3.11.4. Choose a new directory and run subsequent commands there:

python3 --version
mkdir csv-jsonl-demo
cd csv-jsonl-demo
cat > samples.csv <<'CSV'
sample_id,note
001,中文结果
002,"alpha, beta"
003,"first line
second line"
004,
CSV

The third data record spans two physical lines but remains one CSV record. Save the next section as csv_to_jsonl.py, or download the complete script.

3. Complete implementation

"""Stream UTF-8 CSV to JSONL without overwriting output. Python 3.10+, Linux."""
import argparse
import csv
import json
import os
from pathlib import Path
import sys
import tempfile


def convert(source, output):
    with source.open(encoding="utf-8-sig", newline="") as incoming:
        reader = csv.reader(incoming, strict=True)
        header = next(reader, None)
        if not header or any(not name.strip() for name in header):
            raise ValueError("missing or empty column name")
        if len(set(header)) != len(header):
            raise ValueError("duplicate column name")
        count = 0
        with tempfile.TemporaryDirectory(prefix=".csv-jsonl-", dir=output.parent) as work:
            staged = Path(work) / "result.jsonl"
            with staged.open("x", encoding="utf-8", newline="\n") as outgoing:
                for number, row in enumerate(reader, 1):
                    if len(row) != len(header):
                        raise ValueError(
                            f"record {number}, ending at CSV line {reader.line_num}: "
                            f"expected {len(header)} fields, got {len(row)}"
                        )
                    record = dict(zip(header, row))
                    outgoing.write(json.dumps(record, ensure_ascii=False) + "\n")
                    count += 1
            os.link(staged, output)
    return count


def main():
    parser = argparse.ArgumentParser(description=__doc__, allow_abbrev=False)
    parser.add_argument("source", type=Path)
    parser.add_argument("output", type=Path)
    args = parser.parse_args()
    try:
        count = convert(args.source, args.output)
    except (OSError, UnicodeError, csv.Error, ValueError) as exc:
        print(f"ERROR: {exc}", file=sys.stderr)
        return 1
    print(f"OK: {count} records -> {args.output}")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

The official Python CSV documentation describes newline="" and the default string values. Using csv.reader lets us check duplicate headers and field counts before building a dictionary, preventing duplicate keys from replacing values or extra fields from disappearing. strict=True catches some parsing errors; structure checks remain necessary.

The json.dumps() documentation describes ensure_ascii. Here it keeps Chinese readable while retaining JSON escaping for control characters. Do not add indent, which would spread an object across multiple lines.

Output has two stages: write complete JSONL inside the destination directory, then create the final name with os.link(). On a local Linux filesystem, an existing destination makes that step fail, preserving the old file. Staging and output share the same filesystem. After linking, removing the staging name does not remove the final file’s data.

The TemporaryDirectory context manager handles cleanup on normal exit and exception unwinding. The destination directory must already exist, be writable, and support hard links. This example makes no network-filesystem or power-loss durability guarantee and does not overwrite with os.replace(). Forced termination can leave a .csv-jsonl-* staging directory; confirm no process still uses it before cleanup.

4. Run and inspect the result

python3 csv_to_jsonl.py --help
python3 csv_to_jsonl.py samples.csv samples.jsonl
cat samples.jsonl

The success message is OK: 4 records -> samples.jsonl, followed by four lines:

{"sample_id": "001", "note": "中文结果"}
{"sample_id": "002", "note": "alpha, beta"}
{"sample_id": "003", "note": "first line\nsecond line"}
{"sample_id": "004", "note": ""}

first line\nsecond line occupies one JSON line in the file; decoding restores the embedded line break. Do not call json.load() once on the whole JSONL file. Call json.loads() on each line; see the JSONL validation tutorial for further checks.

Memory mainly depends on the header and largest individual record rather than the total record count. The CSV parser still imposes a field-length limit. Disk must accommodate the full new output, and downstream processing should start only after success. For the basics of incremental processing, see streaming large CSV files.

5. Verify content and failure behavior

1. Check the actual file, beyond its success message:

python3 - <<'PY_CHECK'
import json
from pathlib import Path
payload = Path("samples.jsonl").read_bytes()
assert not payload.startswith(b"\xef\xbb\xbf")
assert payload.endswith(b"\n")
lines = payload.decode("utf-8").splitlines()
assert len(lines) == 4
rows = [json.loads(line) for line in lines]
assert rows[0] == {"sample_id": "001", "note": "中文结果"}
assert rows[1]["note"] == "alpha, beta"
assert rows[2]["note"] == "first line\nsecond line"
assert rows[3]["note"] == ""
print("OK: four objects, strings, UTF-8, and multiline field")
PY_CHECK

2. Continue in your fresh experiment directory with these checks. The first data record in bad.csv is valid; the second deliberately has an extra field:

python3 - <<'PY_TEST'
from pathlib import Path
import subprocess
import sys
original = Path("samples.jsonl").read_bytes()
command = [sys.executable, "csv_to_jsonl.py"]
again = subprocess.run(command + ["samples.csv", "samples.jsonl"], capture_output=True)
assert again.returncode == 1
assert Path("samples.jsonl").read_bytes() == original
Path("bad.csv").write_text("a,b\n1,2\n3,4,5\n", encoding="utf-8")
bad = subprocess.run(command + ["bad.csv", "bad.jsonl"], capture_output=True, text=True)
assert bad.returncode == 1
assert not Path("bad.jsonl").exists()
assert not list(Path(".").glob(".csv-jsonl-*"))
print(bad.stderr.strip())
print("OK: existing output unchanged; late failure leaves no output")
PY_TEST

Expect record 2, ending at CSV line 3: expected 2 fields, got 3 and the successful check message. Conversion returns 0 on success, 1 for input, output, or validation failures, and 2 for usage errors such as missing arguments. Testing also covered BOM, CRLF, empty input, header-only input, malformed quotes, invalid UTF-8, duplicate headers, an existing symlink destination, and 100,000 records.

6. Troubleshooting

Summary

Read CSV records with a CSV parser, then let a JSON encoder write one line per record to preserve strings and embedded line breaks. Stage the output until conversion succeeds before creating its final name, so downstream tasks can use the exit status and a clear file boundary to decide when to proceed.