Python JSON Lines 实战：逐行处理、定位坏数据与控制内存

为什么日志导出适合 JSON Lines

导出事件日志或离线处理接口记录时，常见做法是把所有对象放进一个 JSON 数组，再用 json.load 一次读入。数据量变大后，整个列表会同时驻留内存；其中一个位置格式错误，也不容易直接定位到原始文件行。

JSON Lines 的约定更简单：每个物理行保存一个完整 JSON 值。本文把范围进一步收窄为“每行一个对象”，用 Python 标准库实现逐行读取、错误定位和增量写出。示例仅处理本地模拟数据，不涉及云服务调用，也不声称做过大文件性能测试。

1. 先分清物理换行与字符串中的换行

下面是两条记录，而不是一个 JSON 数组：

{"id": 1, "message": "hello"}
{"id": 2, "message": "first\nsecond"}

第二条 message 里的反斜杠 n 是 JSON 转义序列，解析后才成为字符串中的换行。不要对整个文件使用 json.load，也不要让单条记录使用 indent 参数输出成多行，否则会破坏逐行读取的边界。

JSON Lines 通常使用 UTF-8，且不带 BOM。对来源不确定的导入文件，可以有意识地选择 utf-8-sig 兼容文件开头的 BOM；对新生成的文件，建议明确使用 utf-8。

2. 一个可复用的逐行解析器

import json
from collections.abc import Iterator
from typing import TextIO


def iter_jsonl(stream: TextIO) -> Iterator[tuple[int, dict]]:
    for line_number, line in enumerate(stream, start=1):
        if not line.strip():
            continue
        try:
            record = json.loads(line)
        except json.JSONDecodeError as exc:
            raise ValueError(
                f"line {line_number}, column {exc.colno}: {exc.msg}"
            ) from exc
        if not isinstance(record, dict):
            raise ValueError(f"line {line_number}: expected a JSON object")
        yield line_number, record

这里有三个明确的设计选择：

• 返回物理行号，便于把后续字段校验错误关联回原文件；即使跳过空白行，行号也不会重新计数。
• 本示例容忍空白行，属于导入策略，并不是说严格的 JSON Lines 格式允许空白记录。需要严格检查时，应把 continue 改成报错。
• 语法错误或非对象值立即终止，不默默吞掉坏数据。若业务允许隔离错误记录，应在另一版导入流程中显式记录错误数量及行号，并避免把完整敏感日志写进错误信息。

JSON 对象解析成功只代表语法与顶层类型符合要求，不代表字段满足业务约束。必填字段、时间格式、金额单位等仍需单独校验。

3. 增量写出，不收集整个结果列表

def rewrite_jsonl(source, target):
    with open(source, encoding="utf-8-sig") as src:
        with open(target, "w", encoding="utf-8", newline="\n") as dst:
            for line_number, record in iter_jsonl(src):
                if "id" not in record:
                    raise ValueError(f"line {line_number}: missing id")
                encoded = json.dumps(record, ensure_ascii=False, allow_nan=False)
                dst.write(encoded + "\n")

调用 rewrite_jsonl(“events.jsonl”, “normalized.jsonl”) 即可把输入规范化为每行一个对象。source 和 target 必须是不同文件，不能边读取原文件边以 w 模式覆盖它。

ensure_ascii=False 保留可读的中文；allow_nan=False 拒绝输出 NaN、Infinity 等非标准 JSON 数值。如果希望输入端也严格拒绝这些常量，可以为 json.loads 配置一个会抛出异常的 parse_constant 函数。

逐行读取避免保留整个文件，但内存仍受单行长度、解析后对象大小以及下游处理逻辑影响。若后续把迭代器包成 list，仍然会一次累积全部记录。对于不可信的大文件，还应按业务限制单条记录大小。

此外，这个简化写法不是原子替换：中途报错时，target 可能已经含有前几条结果。正式导入可先写同目录临时文件，全部校验成功后再替换目标；不要把半成品当成完整导出。

4. 最小验证与观察结果

使用 io.StringIO 构造两条对象记录，中间插入一条空白行，解析返回的物理行号应为 1 和 3；第二条记录中的转义换行应恢复为真正的字符串换行。

本地验证还覆盖了三种情况：第二行 JSON 语法错误能报告 line 2；顶层数组被拒绝；两条记录写出后仍然只有两个物理行，中文与字符串换行往返保持一致。缺少 id 时能够报告原始行号。这五组检查均通过。

这些检查验证的是示例行为，不是吞吐量、峰值内存或生产环境可靠性指标。上线前应使用真实字段约束和代表性文件补充验证。

小结

JSON Lines 的价值不在于更复杂的解析器，而在于清晰的记录边界。用迭代器逐行读取、保留物理行号、明确空白行和坏数据策略，再把结果逐条写出，就能得到一个易理解、易排错的本地处理流程。格式校验、业务校验和文件交付完整性是不同问题，应分别处理。