Python JSONL 大文件分批处理：从流式读取到失败样本报告

JSONL 文件很适合日志、商品、用户行为、模型结果等半结构化数据交换：一行就是一条 JSON。问题也很直接：文件一大，不能整份读进内存；字段一乱，后面的入库、统计、导出都会被拖住。本文给出一套 Python 分批处理 JSONL 的完整路线图，从输入边界到失败样本报告，尽量让这件事变成可复用流程。

先说结论：处理 JSONL 大文件时，不要一上来就把所有行读成列表。更稳的方式是逐行读取、单行解析、字段检查、攒够批次再写出，同时把失败样本和错误原因单独保存。这样内存可控，结果可复查，失败数据也不会悄悄丢掉。

目标和边界：这套流程解决什么问题

本文的目标是完成一个可落地的小工具：读取一个很大的 .jsonl 文件，保留字段完整的数据，按固定批次写出新文件，同时记录坏行和汇总报告。它适合几十万到几千万行的离线数据预处理。

边界也要先说清楚：这里不讨论分布式计算，不引入 pandas，不做复杂业务规则引擎，只使用 Python 标准库把基础流程搭稳。如果数据已经大到单机磁盘和 CPU 明显吃不住，再考虑 Spark、Flink 或数据库侧导入工具会更合理。

全流程总览：JSONL 大文件分批处理

先把流程图摆出来。一次稳定的 JSONL 处理，不是“读文件再写文件”这么粗，而是包含输入、解析、检查、分批写出和报告五个环节。

这张图对应到代码里，就是一个主循环：每读一行就尝试解析 JSON；解析通过后再检查字段；字段通过后进入当前批次；批次数量达到阈值就落盘；最后输出一份报告，告诉我们总行数、有效行、失败行和输出文件数量。

阶段一：逐行读取，避免一次性占满内存

这一阶段的目标是控制内存。很多脚本慢，不是 Python 语法慢，而是一开始就把整个文件读进列表。JSONL 的优势正好在这里：天然可以一行一行处理。

from pathlib import Path
import json

source = Path("data.jsonl")

with source.open("r", encoding="utf-8") as f:
    for line_no, line in enumerate(f, start=1):
        text = line.strip()
        if not text:
            continue
        row = json.loads(text)
        # 后续再做字段检查和分批写出

检查点很简单：脚本运行时，内存占用应该主要由“当前批次大小”决定，而不是由文件总大小决定。如果只是清洗字段、拆分批次，这种方式通常足够稳定。

阶段二：字段检查，把脏数据单独留下

这一阶段的目标是让错误可追溯。不要遇到坏行就直接跳过，也不要把坏行和好数据混在同一个输出文件里。我们先定义一组必填字段，再把失败原因记录下来。

REQUIRED_FIELDS = {
    "id": int,
    "name": str,
    "price": (int, float),
}

def check_row(row: dict) -> tuple[bool, str]:
    for field, expected_type in REQUIRED_FIELDS.items():
        if field not in row:
            return False, f"missing:{field}"
        if not isinstance(row[field], expected_type):
            return False, f"type:{field}"
    return True, ""

字段检查不一定要复杂。先把“必须存在”和“基础类型”做好，已经能拦住大量脏数据。更细的规则，比如价格不能小于 0、名称不能为空，可以继续放到这个函数里。

阶段三：分批写出，给后续处理留稳定入口

这一阶段的目标是让结果稳定可用。分批写出有两个好处：一是单个文件不会过大，二是后续入库、上传、压缩时可以按批次重跑。

from pathlib import Path
import json

BATCH_SIZE = 10000
OUT_DIR = Path("out_batches")
OUT_DIR.mkdir(exist_ok=True)

def write_batch(batch: list[dict], index: int) -> Path:
    out_file = OUT_DIR / f"batch_{index:04d}.jsonl"
    with out_file.open("w", encoding="utf-8") as f:
        for row in batch:
            f.write(json.dumps(row, ensure_ascii=False) + "\n")
    return out_file

这里的检查点是批次编号连续、每个输出文件都是合法 JSONL、最后一个批次即使不足 BATCH_SIZE 也要写出。很多数据缺失问题，恰恰来自最后一批没有处理。

我的推荐流程：从样本试跑到全量处理

到这一步不要急着直接跑全量。我的推荐流程是先拿前 1000 行试跑，确认字段规则和输出目录，再跑全量，并保留失败样本。

from pathlib import Path
import json

SOURCE = Path("data.jsonl")
OUT_DIR = Path("out_batches")
FAILED_FILE = Path("failed_samples.jsonl")
REPORT_FILE = Path("report.json")
BATCH_SIZE = 10000

OUT_DIR.mkdir(exist_ok=True)

def write_jsonl(path: Path, rows: list[dict]) -> None:
    with path.open("w", encoding="utf-8") as f:
        for row in rows:
            f.write(json.dumps(row, ensure_ascii=False) + "\n")

def write_batch(batch: list[dict], index: int) -> Path:
    path = OUT_DIR / f"batch_{index:04d}.jsonl"
    write_jsonl(path, batch)
    return path

def check_row(row: dict) -> tuple[bool, str]:
    rules = {"id": int, "name": str, "price": (int, float)}
    for field, expected_type in rules.items():
        if field not in row:
            return False, f"missing:{field}"
        if not isinstance(row[field], expected_type):
            return False, f"type:{field}"
    return True, ""

def main() -> None:
    batch = []
    failed = []
    batch_index = 1
    total = valid = invalid = 0

    with SOURCE.open("r", encoding="utf-8") as f:
        for line_no, line in enumerate(f, start=1):
            text = line.strip()
            if not text:
                continue
            total += 1
            try:
                row = json.loads(text)
            except json.JSONDecodeError as err:
                invalid += 1
                failed.append({"line": line_no, "reason": f"json:{err.msg}", "raw": text})
                continue

            ok, reason = check_row(row)
            if not ok:
                invalid += 1
                failed.append({"line": line_no, "reason": reason, "raw": row})
                continue

            valid += 1
            batch.append(row)
            if len(batch) >= BATCH_SIZE:
                write_batch(batch, batch_index)
                batch_index += 1
                batch = []

    if batch:
        write_batch(batch, batch_index)

    write_jsonl(FAILED_FILE, failed)
    REPORT_FILE.write_text(
        json.dumps({
            "total": total,
            "valid": valid,
            "invalid": invalid,
            "batch_files": batch_index if batch else batch_index - 1,
            "failed_file": str(FAILED_FILE),
        }, ensure_ascii=False, indent=2),
        encoding="utf-8",
    )

if __name__ == "__main__":
    main()

这份脚本的核心不是代码量，而是顺序：读一行、判一行、攒一批、写一批。失败样本单独留下，报告单独生成，后续复查时就能知道问题集中在解析失败、字段缺失还是类型不对。

常见误区与速查表

容易踩坑的地方主要有四个：第一，一次性读取全文件，导致内存随文件大小上涨；第二，只统计成功数据，不保存失败样本；第三，最后一个不足批次的数据忘记写出；第四，输出文件没有固定编号，后续重跑时难以比对。

总结一下：JSONL 大文件处理的关键，是把“读、查、写、记”拆清楚。逐行读取负责稳住内存，字段检查负责稳住质量，分批写出负责稳住后续流程，失败样本和报告负责让结果能复查。这个骨架搭好后，再叠加业务规则、压缩上传或入库，就会顺很多。