用 Python 标准库逐行读取 JSON Lines：定位坏行与控制内存

日志和数据导出常用 JSON Lines（也叫 NDJSON）：文件的每一行都是一个独立的 JSON 值。例如两条事件可以写成两行，而不需要用一个大数组包住。这样的文件可以边读边处理，遇到格式错误时也能指出具体是哪一行。

下面只用 Python 标准库写一个读取器。示例约定：空白行跳过，非空行必须是 JSON 对象；如果你的数据允许数字或数组，可以移除对象类型检查。

可运行示例

将下面的代码保存为 read_jsonl.py，用 Python 3.8 或更新版本运行：

import json
import tempfile
from pathlib import Path
from typing import Iterator


class JsonLinesError(ValueError):
    pass


def iter_records(path: Path) -> Iterator[dict]:
    try:
        with path.open("r", encoding="utf-8") as source:
            for line_number, line in enumerate(source, 1):
                if not line.strip():
                    continue
                try:
                    record = json.loads(line)
                except json.JSONDecodeError as error:
                    raise JsonLinesError(
                        f"{path.name}:{line_number}:{error.colno}: "
                        f"JSON 格式错误：{error.msg}"
                    ) from error
                if not isinstance(record, dict):
                    raise JsonLinesError(
                        f"{path.name}:{line_number}: 每行必须是 JSON 对象"
                    )
                yield record
    except UnicodeDecodeError as error:
        raise JsonLinesError(f"{path.name}: 文件不是有效的 UTF-8") from error


if __name__ == "__main__":
    with tempfile.TemporaryDirectory() as directory:
        path = Path(directory) / "events.jsonl"
        path.write_text('{"name":"Ada"}\n\n{"name":"Lin"}\n', encoding="utf-8")
        print([record["name"] for record in iter_records(path)])

        path.write_text('{"name":"Ada"}\n{"name": }\n', encoding="utf-8")
        try:
            list(iter_records(path))
        except JsonLinesError as error:
            print(error)

        path.write_text('{"name":"Ada"}\n42\n', encoding="utf-8")
        try:
            list(iter_records(path))
        except JsonLinesError as error:
            print(error)

本地运行的输出是：

['Ada', 'Lin']
events.jsonl:2:10: JSON 格式错误：Expecting value
events.jsonl:2: 每行必须是 JSON 对象

为什么逐行读取

json.load() 期望整个文件是一个 JSON 文档，不能直接用于多行独立 JSON 的格式。这里通过文件迭代器一次取一行，解析成功后立即 yield，调用方可以一条一条处理，例如：

for record in iter_records(Path("events.jsonl")):
    print(record.get("name"))

这样不需要把所有记录先放进列表。示例里的 list(iter_records(path)) 只用于一次性演示错误；实际处理大文件时应保持逐条消费。内存仍会容纳当前的一整行，所以若输入可能含极长行，还要按业务增加行长或文件大小限制。

错误位置与格式约定

enumerate(source, 1) 给出文件中的实际行号，空白行虽然被跳过，仍计入行号。JSONDecodeError.colno 给出坏行中的列号；这比只提示“JSON 格式错误”更容易定位多余逗号、缺失引号等问题。

这个读取器没有吞掉 FileNotFoundError：路径不存在时，调用方会直接看到标准文件异常。它会把 UTF-8 解码失败转为较容易理解的提示。如果文件来自不可信来源，还应考虑限制单行长度、记录数量，并校验每个对象的字段结构；“是 JSON 对象”只是一层最基本的检查。