JSONL 日志逐行读还不够：用 Python 限制超长行、隔离坏记录并生成稳定汇总

日志里混进一条损坏记录，汇总脚本就整批退出；把异常全部吞掉，结果又少了几条却无法解释。改成 for line in file 后，整份文件确实不再一次性进入内存，但一条异常长的物理行仍然可能分配很大的字符串。日志处理的难点，常常藏在这些边界里。

本文做一个小型离线工具：读取 JSONL 请求日志，按服务名统计条数和总耗时。只使用 Python 标准库，全部结论来自本机固定输入的执行与边界检查，不涉及云端服务、吞吐量压测或线上可靠性承诺。

一、先确定哪些数据可以计入结果

每个物理行表示一条记录，例如：

{"service":"api","elapsed_ms":7}

这里采用以下规则。规则本身是应用选择，不应交给 JSON 解析器隐式决定。

service 必须是非空白字符串，长度不超过 32 个字符；原始名称保留，不自动去空格或转小写，因此 api 与 API 是两个服务。elapsed_ms 必须是非负整数，True、浮点数和数字字符串均不接受。Python 中 bool 是 int 的子类，因此这里用 type(elapsed) is int 明确拒绝布尔值。

解析器允许额外字段，但不使用它们。重复键会导致信息被悄悄覆盖，因此在解析阶段拒绝；Python 的 JSON 解析器默认接受的 NaN、Infinity 和 -Infinity 也通过回调拒绝。这样，字段判断看到的是已经通过本例 JSON 规则的对象。

128 字节包含存在的行结束符，CRLF 的两个字节也算在内。最后一行没有换行可以接受。本例不自动移除 UTF-8 BOM，带 BOM 的首行会落入 JSON 错误计数；输入方需要明确编码约定。

二、流式读取需要处理两个增长源

第一处是单行长度。二进制文件的 readline(limit + 1) 最多取出指定数量的字节。读取 129 字节是为了判断是否超过 128 字节：只读 128 字节无法区分“刚好到文件末尾”和“后面还有内容”。

一旦超长，不能直接回到主循环，否则剩余片段会被误当成下一条日志。代码继续以最多 64 字节的片段读到 LF 或 EOF，随后只对这一整个物理行记录一次拒绝。整个过程中不拼接这些片段，也不把原文放进错误列表。

第二处是聚合状态。即使每行都很短，如果每条日志都有一个新的服务名，普通字典仍会持续增长。本例最多保留 8 个服务，并且只保存前 3 个被拒绝的物理行号。超过服务数量上限时只拒绝新名称，已有名称继续累计。

这仍不等于严格的常量内存证明：Python 整数会随着计数和值的大小增加位数，解析器和文件缓冲也有开销。这里验证的是单次取行请求有上限、分组槽位有上限、诊断样本有上限，没有测量 RSS，也没有给出每秒处理量。

三、完整工具：限长、分类、聚合与稳定输出

将下面的完整代码保存为 articles/jsonl-bounded-stream-demo.py，在包含 articles 的目录运行。需要 Python 3.10 或更高版本，无第三方依赖。直接执行使用内置的合成样本；传入路径则以二进制方式读取自己的 UTF-8 JSONL 文件。

"""Bounded JSONL teaching example; Python 3.10+, standard library only."""
from __future__ import annotations

import argparse
import io
import json
import sys
from collections import Counter
from pathlib import Path
from typing import BinaryIO, Iterator

MAX_LINE_BYTES = 128  # Includes LF, when present.
MAX_SERVICES = 8
DRAIN_BYTES = 64


def physical_lines(stream: BinaryIO, limit: int) -> Iterator[tuple[int, bytes | None]]:
    """Yield None once per oversized physical line, draining bounded chunks."""
    if limit < 1:
        raise ValueError("limit must be positive")
    number = 0
    while True:
        raw = stream.readline(limit + 1)
        if not raw:
            return
        number += 1
        if len(raw) <= limit:
            yield number, raw
            continue
        while not raw.endswith(b"\n"):
            raw = stream.readline(DRAIN_BYTES)
            if not raw:
                break
        yield number, None


def reject_constant(token: str) -> None:
    raise ValueError("non-finite constant: " + token)


def unique_object(pairs: list[tuple[str, object]]) -> dict[str, object]:
    result: dict[str, object] = {}
    for key, value in pairs:
        if key in result:
            raise ValueError("duplicate key")
        result[key] = value
    return result


def summarize(stream: BinaryIO, limit: int = MAX_LINE_BYTES,
              max_services: int = MAX_SERVICES) -> dict[str, object]:
    if max_services < 1:
        raise ValueError("max_services must be positive")
    counts: Counter[str] = Counter()
    groups: dict[str, list[int]] = {}
    # Bound diagnostics too: only the first three rejected physical line numbers.
    rejected_sample: list[int] = []

    def reject(reason: str, number: int) -> None:
        counts[reason] += 1
        if len(rejected_sample) < 3:
            rejected_sample.append(number)

    for number, raw in physical_lines(stream, limit):
        counts["physical_lines"] += 1
        if raw is None:
            reject("oversized", number)
            continue
        try:
            text = raw.decode("utf-8")
        except UnicodeDecodeError:
            reject("invalid_utf8", number)
            continue
        if not text.strip(" \t\r\n"):
            counts["blank"] += 1
            continue
        try:
            row = json.loads(text, parse_constant=reject_constant,
                             object_pairs_hook=unique_object)
        except (ValueError, RecursionError):
            reject("invalid_json", number)
            continue
        if not isinstance(row, dict):
            reject("invalid_record", number)
            continue
        service, elapsed = row.get("service"), row.get("elapsed_ms")
        if (not isinstance(service, str) or not service.strip()
                or len(service) > 32 or type(elapsed) is not int or elapsed < 0):
            reject("invalid_record", number)
            continue
        if service not in groups:
            if len(groups) >= max_services:
                reject("service_limit", number)
                continue
            groups[service] = [0, 0]
        groups[service][0] += 1
        groups[service][1] += elapsed
        counts["accepted"] += 1

    names = ("physical_lines", "accepted", "blank", "oversized", "invalid_utf8",
             "invalid_json", "invalid_record", "service_limit")
    return {
        "counts": {name: counts[name] for name in names},
        "rejected_line_sample": rejected_sample,
        "services": [
            {"service": name, "count": groups[name][0],
             "total_ms": groups[name][1]}
            for name in sorted(groups)
        ],
    }


def demo_bytes() -> bytes:
    return b"".join([
        b'{"service":"pay","elapsed_ms":12}\n',
        b'{"service":"api","elapsed_ms":7}\n',
        b' \t\r\n',
        b'{broken}\n',
        b'{"service":"api","elapsed_ms":true}\n',
        b'[1,2]\n',
        b'\xff\n',
        b'x' * 300 + b'\n',
        b'{"service":"pay","elapsed_ms":8}\n',
        b'{"service":"api","elapsed_ms":0}\n',
        b'{"service":"api","elapsed_ms":NaN}\n',
        b'{"service":"api","service":"pay","elapsed_ms":9}\n',
        b'{"service":"batch","elapsed_ms":5}',  # No final LF.
    ])


def main() -> None:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("path", nargs="?", type=Path,
                        help="UTF-8 JSONL input; omit for embedded fixture")
    args = parser.parse_args()
    if args.path is None:
        result = summarize(io.BytesIO(demo_bytes()))
    else:
        with args.path.open("rb") as stream:
            result = summarize(stream)
    sys.stdout.write(json.dumps(result, ensure_ascii=False, sort_keys=True,
                               indent=2) + "\n")


if __name__ == "__main__":
    main()

读取、解析、业务校验和聚合分别占据独立阶段。拒绝一条记录后立即 continue，因此一条物理行只属于一个分类。判断顺序也属于规则：超长行先归为超长，即使内部还有坏 UTF-8，也不继续解析以寻找更多错误。

文件打开失败、读取时的 I/O 错误不会被这段代码吞掉，会让程序以错误结束。这样可以区分“输入里存在坏记录”和“根本没能完成文件读取”。如果需要严格模式，可以在调用方根据拒绝计数决定是否接收结果。

四、复现实验：13 个物理行，5 条合法记录

内置样本不是随机生成的，包含以下有意排列的输入：

在项目根目录执行：

python -B articles/jsonl-bounded-stream-demo.py

标准输出如下，最后带一个换行：

{
  "counts": {
    "accepted": 5,
    "blank": 1,
    "invalid_json": 3,
    "invalid_record": 2,
    "invalid_utf8": 1,
    "oversized": 1,
    "physical_lines": 13,
    "service_limit": 0
  },
  "rejected_line_sample": [
    4,
    5,
    6
  ],
  "services": [
    {
      "count": 2,
      "service": "api",
      "total_ms": 7
    },
    {
      "count": 1,
      "service": "batch",
      "total_ms": 5
    },
    {
      "count": 2,
      "service": "pay",
      "total_ms": 20
    }
  ]
}

结果可以直接手算：api 两条，总耗时 7 + 0 = 7；pay 两条，总耗时 12 + 8 = 20；batch 一条，总耗时 5。物理行总数满足 13 = 5 + 1 + 1 + 1 + 3 + 2 + 0，分别对应接受、空白、超长、编码错误、JSON 错误、字段错误和分组上限拒绝。

坏记录样本只有 [4, 5, 6]，因为样本上限是 3。其余坏记录仍然计数，没有全部保留行号，更没有打印原始日志内容。生产环境若要定位每一条问题记录，应另行设计隔离文件及访问控制，不能把无限错误列表留在内存中。

读取自己的文件时执行：

python -B articles/jsonl-bounded-stream-demo.py sample.jsonl

这里的 sample.jsonl 是调用方提供的路径。不要把格式化为多行的一个大 JSON 对象当作 JSONL 输入；本工具按物理行切分，不会跨行拼接 JSON。

五、稳定输出依赖规则，而不只依赖排序参数

sort_keys=True 只排序 JSON 对象的键，不会排序数组。因此 services 在构造时单独使用 sorted(groups)，按 Python 字符串顺序输出服务名。示例输入先出现 pay，结果仍依次为 api、batch、pay。

汇总只保留整数条数和整数总耗时，避免为了显示平均值引入浮点舍入。需要平均值时，可以保留 total_ms / count 这个分数，或者另外明确显示精度和舍入规则。

确定性在这里指：同一份输入、相同规则与运行环境，输出字节可以重复比较。不要据此推断所有输入重排后都相同：拒绝行号会改变，而且触发 8 个服务上限时，最先出现的名称决定哪些服务被保留。这里选择的是先到先收策略；如果业务要求全文件中的服务按名称取前 8 个，需要另一套筛选规则。

六、本地核验与适用边界

配套验证器位于 articles/jsonl-bounded-stream-validate.py，与工具和两份 Markdown 放在同一目录。它运行工具两次并逐字节比较 stdout，以独立字面量核对预期汇总；另外检查空输入、含 LF 的临界行、没有末尾 LF 的临界行、超长无换行行、长行排空恢复和服务数量上限。受监测的输入对象会拒绝没有上限或超过 129 字节的 readline 请求，这验证了指定测试路径的读取方式，不是进程内存测量。

python -B articles/jsonl-bounded-stream-validate.py

验证器也检查文章中的完整 Python 代码和 JSON 输出与实际文件一致，再保存逐字节 stdout 与验证 JSON。实际执行环境为 Windows、Python 3.12.14；本次验证通过 14 项检查，输出 PASS: 14 checks; demo exits 0 twice; stdout bytes identical。它不需要网络、数据库或云账号。

这个工具适合说明如何把日志处理的边界写清楚。128 字节和 8 个服务都是教学参数，真实业务需要根据日志契约调整。它没有实现时间窗口、跨文件去重、断点恢复、压缩输入、隔离文件落盘和输出文件的原子替换；也没有证明对恶意输入的全面防护。若输入仍在追加，读取结果还受读取时刻影响，本文只使用固定不变的样本。

逐行处理解决了整文件加载的问题；单行限制、错误分类、聚合上限和稳定序列化，才让这个小工具的行为能够被准确解释和重复核对。