用 Python 标准库读取 JSONL：别让重复记录和坏行把统计带偏

JSONL 每行保存一个 JSON 对象，适合持续追加日志、任务结果和数据导出。真正开始统计时，却常常遇到两类问题：同一个任务写了多次，文件末尾还有一条没有写完的记录。直接数行会高估任务数量；一遇到解析错误就退出，又会让整份报告无法生成。

先明确统计对象：如果关心“发生过几次事件”，每行都应该保留；如果关心“每个任务的最新状态”，则需要按稳定标识去重。下面采用第二种口径，用 id 作为标识，让文件中最后一条有效记录覆盖前面的状态。

一个可以直接运行的最小实现

import json
from io import StringIO


def snapshot(lines):
    latest, skipped = {}, []
    for number, line in enumerate(lines, 1):
        if not line.strip():
            continue
        try:
            row = json.loads(line)
        except json.JSONDecodeError:
            skipped.append(number)
            continue
        if not isinstance(row, dict) or not isinstance(row.get('id'), str) or not row['id']:
            skipped.append(number)
            continue
        latest[row['id']] = row
    return latest, skipped


# 明确标注的测试数据，不是生产日志
sample = StringIO(
    '{"id":"A","status":"pending"}\n'
    '{"id":"B","status":"done"}\n'
    '{"id":"A","status":"done"}\n'
    '{"id":\n'
    '[]\n'
)
rows, skipped = snapshot(sample)
assert set(rows) == {'A', 'B'}
assert rows['A']['status'] == 'done'
assert skipped == [4, 5]
print(len(rows), skipped)

示例输出为 2 [4, 5]：三个有效对象对应两个任务，第四行是残缺 JSON，第五行虽然是合法 JSON，却不是约定的对象结构。解析成功与数据有效，是两件不同的事。读取真实文件时，改成 with open('records.jsonl', encoding='utf-8') as f: rows, skipped = snapshot(f) 即可。

为什么要留下坏行行号

跳过坏行不等于把问题藏起来。报告应同时列出有效任务数和异常行数，必要时回看对应原文。持续写入的文件可能恰好被读到半行，下次重新读取时它可能恢复完整；如果中间位置反复出现坏行，则应检查写入方，而不是长期忽略。

只捕获 JSONDecodeError 也很重要。文件打不开、编码不匹配或程序自身出错，不应该被一个宽泛的 except Exception 伪装成“数据为空”。错误是否可忽略，必须由明确的数据约定决定。

最后一条，并不总是最新时间

这个实现相信文件追加顺序。如果把多台机器的日志拼在一起，最后出现的记录未必时间最新，此时要按时间戳或递增版本选择。还应先约定时间戳单位：秒和毫秒不能混用，按自然日汇总则应显式选择时区，避免午夜附近的数据被分错日期。

使用边界

逐行遍历避免把整个文件一次性读入内存，但 latest 仍保存每个唯一任务，空间开销随任务数增长。普通日志先用标准库就够；当唯一标识达到无法放进内存的规模，再迁移到 SQLite，利用主键和更新语句维护快照。

本地验证覆盖了重复标识、残缺 JSON、非对象结构、缺失标识、空行和空输入。统计脚本不需要复杂框架，但至少应留下一个能运行的固定样例：先把口径验证正确，再讨论速度。