Python 数据处理与评测：从原始记录到可信报告

用途、程度与前置#

AI 应用最常见的离线任务是把日志、人工标签和模型结果整理成一份可靠报告。相比复杂模型，数据关联错误更容易悄悄制造“进步”。本章适用于准备评测集、比较模型版本、检查批处理导出或汇总人工标注。L2 要求是独立做小型数据管道，并能证明样本没有重复、漏算或跨集合泄漏。

前置是上一章的文件、字典、异常与命令行知识，以及 precision、recall 的含义。使用 Python 3.12 以上标准库，不安装 pandas，也不需要联网。这样你能先看清数据操作本身，再判断大规模任务是否需要 DataFrame、数据库或分布式处理。

把管道分成可检查的阶段#

一次可靠处理通常包含读取、解析、校验、规范化、关联、计算和导出。读取成功只表示拿到字节；CSV 语法正确不代表标签有效；关联成功也不代表样本来自同一个实验版本。每个阶段都应有数量和错误证据，例如读取多少行、有效多少行、重复多少、缺失多少。

CSV 是表格交换格式，但引号、逗号、换行可以合法存在于字段内。不要用 line.split(",") 自己解析。Python csv.DictReader 把表头映射为字典键，读取出的值通常仍是字符串，数值和枚举要主动转换验证；文件打开时用 newline=""，交由 csv 模块处理换行。CSV 文档

JSONL 每行是一条 JSON 对象，适合逐行追加和大文件流式处理。空行如何处理、重复 id 是拒绝还是去重、失败结果是否需要 status 字段，必须写入合同。普通 JSON 数组和 JSONL 不同，不能对整个 JSONL 文件只调用一次 json.loads。

规范化也有业务语义。去除首尾空白通常合理，但删除内部空格、标点、大小写或统一繁简可能改变输入含义。保留原始文本和规范化文本，记录规则版本。不要为了消除重复而把本来不同的业务请求合并，更不能从测试集错误里反向构造标签再覆盖原始人工判断。

划分、去重与关联为什么是重点#

数据集切分要避免同一文档、用户或会话的近似内容散落在不同集合。随机按行切分容易让模型“见过几乎相同的题”。按 group_id 进行分组切分，再按类别与时间检查分布，通常更符合真实泛化问题。

“固定随机种子”只能保证同一输入和算法下可复现；数据排序或库版本改变仍可能影响结果。对持续新增数据，可以用稳定业务组标识做哈希分桶，但组大小和类别平衡需要另行检查。冻结版本后输出每条样本所属集合的清单，而不是每次评测重新随机分配。

预测结果可能异步完成，顺序经常变化。必须按 sample_id 关联，禁止 zip 两个列表后当成一一对应。对 test 集的每个 id，明确要求恰好一条结果；缺失、重复、未知 id 均需要处理。若缺失是 API 超时，应该计入运行可靠性报告，而不是删除这题后提高质量分数。

评测报告需要记录数据版本、预测版本、标签集合、指标口径和输入指纹。指纹证明“这次读取的文件字节相同”，不能证明数据正确或没有泄漏。为了审计，还要保留变更说明和来源信息，并限制原始用户数据的访问范围。

完整示例：CSV 标签与 JSONL 预测按 id 关联#

保存为 data_eval.py。为使教程完全离线且可复制，原始小样本嵌在同一文件里；生产时把读取部分替换为实际文件流，后续校验和关联逻辑保持一致。

# data_eval.py
import csv
import hashlib
import io
import json

DATA = """id,group_id,split,label
d1,g1,dev,refund
v1,g2,validation,other
t1,g3,test,refund
t2,g4,test,refund
t3,g5,test,other
t4,g6,test,other
"""
# 故意让输出顺序与标签不同，验证关联不依赖行顺序
PREDICTIONS = """{"id":"t3","label":"refund"}
{"id":"t1","label":"refund"}
{"id":"t4","label":"other"}
{"id":"t2","label":"refund"}
"""

def main() -> None:
    dataset = {}
    group_splits = {}
    for line, row in enumerate(csv.DictReader(io.StringIO(DATA, newline="")), start=2):
        if set(row) != {"id", "group_id", "split", "label"}:
            raise ValueError(f"CSV 第 {line} 行字段错误")
        if not all(isinstance(v, str) and v.strip() for v in row.values()):
            raise ValueError(f"CSV 第 {line} 行存在空字段")
        if row["id"] in dataset:
            raise ValueError("重复样本：" + row["id"])
        if row["split"] not in {"dev", "validation", "test"}:
            raise ValueError("未知集合")
        if row["label"] not in {"refund", "other"}:
            raise ValueError("未知标签")
        previous = group_splits.setdefault(row["group_id"], row["split"])
        if previous != row["split"]:
            raise ValueError("同一业务组跨集合，存在泄漏风险")
        dataset[row["id"]] = row

    predictions = {}
    for number, line in enumerate(PREDICTIONS.splitlines(), start=1):
        item = json.loads(line)
        if not isinstance(item, dict) or set(item) != {"id", "label"}:
            raise ValueError(f"JSONL 第 {number} 行结构错误")
        if not isinstance(item["id"], str) or item["label"] not in {"refund", "other"}:
            raise ValueError(f"JSONL 第 {number} 行字段无效")
        if item["id"] in predictions:
            raise ValueError("重复预测：" + item["id"])
        predictions[item["id"]] = item["label"]

    test = {key: row for key, row in dataset.items() if row["split"] == "test"}
    if not test:
        raise ValueError("测试集为空")
    missing = set(test) - set(predictions)
    unknown = set(predictions) - set(test)
    if missing or unknown:
        raise ValueError(f"预测集合不完整：missing={sorted(missing)}, unknown={sorted(unknown)}")

    tp = fp = fn = correct = 0
    for key, row in test.items():
        gold = row["label"]
        pred = predictions[key]
        correct += int(pred == gold)
        tp += int(pred == "refund" and gold == "refund")
        fp += int(pred == "refund" and gold != "refund")
        fn += int(pred != "refund" and gold == "refund")
    precision = tp / (tp + fp) if tp + fp else 0.0
    recall = tp / (tp + fn) if tp + fn else 0.0
    print(json.dumps({
        "dataset_sha256": hashlib.sha256(DATA.encode("utf-8")).hexdigest(),
        "test_rows": len(test), "accuracy": correct / len(test),
        "precision": precision, "recall": recall
    }, ensure_ascii=False, indent=2))

if __name__ == "__main__":
    main()

python data_eval.py

预期 test_rows 为 4、accuracy 为 0.75、precision 约为 0.6667、recall 为 1.0。dataset_sha256 为固定的 64 位十六进制字符串，只要 DATA 字节不变就保持一致。这里一条 dev 和一条 validation 仅用于展示集合角色，不足以支撑真实模型调优。

逐段解析：DictReader 读取表头，逐行校验避免脏数据进入下一步；group_splits 记录每组第一次出现的集合，后续冲突就失败；预测字典保证同一 id 只有一次结果；集合差集同时找出缺失和多余结果。最后遍历 test，并用预测字典取标签，所以完成顺序不会影响分数。

本例遇到脏数据直接终止，适合固定评测集。真实导入可采用隔离错误记录的方式，但必须输出拒收数量、原因与原始位置，并在评测发布前确定是否允许缺失。不能在 except 中简单 continue，让数据静悄悄消失。

规模扩大后的选择#

几万条小记录通常可以先用标准库完成；规模取决于每条内容大小而非只有行数。JSONL 可以流式解析，指标计数可增量累加，但去重和关联仍可能占内存。数据更大时，用 SQLite 临时表建立唯一索引，或使用数据库 join；不要只把全部字典换成 pandas 就以为内存问题自动解决。

使用 pandas 的价值是便于分组、透视和统计，不是跳过数据合同。缺失值、字符串类型、时间解析和 join 方式仍需明确。例如 inner join 会丢弃没有匹配项的记录，left join 才更容易保留待检查缺失。即使使用高级库，先写总行数与唯一 id 数断言。

结果要按类别、来源、长度、时间等切片，并报告每片样本量。某类只有两条时达到百分之百不说明稳定优异。重复比较很多方案时容易挑中偶然高分，应该在验证集选型，再用冻结测试集报告，并让人工复查关键错误。

常见错误与排障#

分数每次变化，先看数据指纹、排序、切分清单和预测 id；不要第一反应认为模型随机。CSV 多出 None 键通常表示一行字段数量超过表头，应检查引号和分隔符。中文乱码先确定文件实际编码，UTF-8 BOM 文件可在已知来源下用 utf-8-sig，不能靠忽略错误吞字。

指标突然提升但测试样本减少，检查失败记录、inner join、去重逻辑和标签过滤。样本去重应在切分之前考虑，并保留业务组关系；切分之后发现泄漏，应该重新制作版本并重新跑基线。

把外部数据当作一个没有类型保证的接口#

前端请求 API 时，你不会因为写了 TypeScript 类型就相信服务器一定返回正确字段。处理 CSV、JSONL 和标注文件也是同样的边界，只是请求变成文件，响应变成成千上万行。解析器通常只能证明语法符合格式，不能证明时间、金额、标签、用户归属和主键都有业务意义。

先写出输入合同，再写转换代码。每个字段应有名称、允许类型、是否必填、空值语义、取值集合、单位和来源。例如 latency_ms 是非负整数毫秒，缺失表示未采集，零表示观察到零或计时精度不足；两者不应混为一谈。标签为空可能是尚未标注，不应自动归为负类。只有把这些含义固定，后续分组统计才有解释基础。

CSV 的全部字段通常从字符串开始，因此 "0012" 是否转换为整数取决于业务。订单号里的前导零可能有意义，转换后就失去了原标识；金额字符串则可能需要转换，但要明确是元还是分。对标识符更适合保留字符串，对测量值进行严格解析。为了“看起来统一”而对所有字段调用 int 或 strip，常常造成无声的数据损失。

空字符串、只有空格、JSON null、缺失键和字符串 "null" 是五种可能不同的状态。你可以在合同中选择合并某些状态，但必须能解释为什么。数据从多个团队导出时，应该在来源适配层分别规范化，之后再进入统一内部结构；不要在每一个指标函数里散落各式兼容判断。

错误信息应能定位到源文件和行号。记录拒收原因时，尽量包含字段名、错误类别和非敏感标识，而不是把整行用户消息复制进公共日志。保留原始文件的受控副本便于追溯，处理结果则带来源指纹和清洗版本。这样发现错误后可以修规则重跑，而不是手工修改最后一份报告。

编码、Unicode 与日期中的隐蔽差异#

UTF-8 是常用交换编码，但某些表格导出会带字节顺序标记。确定来源确实如此时，utf-8-sig 可以在读取时处理开头标记；如果直接把带标记的 id 当表头，第一列名称可能看起来正确却无法匹配。编码错误不应该通过忽略无法解码字节来“修复”，因为一个被丢弃的字就可能改变政策条件或否定含义。

Unicode 中视觉相同的字符串可能有不同代码点组合。NFC 规范化可将某些等价组合表示统一，适合文本比对，但不是万能清洗。全角半角、繁简、大小写和空白合并有不同语义，不能一股脑归入一个函数。尤其代码、密码、表格对齐和多语言文本中的空格未必可删除。Unicode 规范化接口

文本的原始内容与用于去重的标准化内容最好分开保存。去重指纹可以基于规范化副本，但模型输入是否也使用该副本必须明确。若去重时把标点删除，两个问题“可以退款吗”和“不可以退款吗”仍然不同；更激进的相似匹配需要人工抽样核查，不能用哈希替代语义判断。

时间字段还涉及时区。没有时区的本地时间无法在全球事件中唯一定位；同一时间字符串可能代表不同真实时刻。对日志汇总，尽量接受带偏移的 ISO 时间并转换为 UTC 保存，同时保留必要业务时区。按深圳自然日统计和按 UTC 日统计的边界不同，报告必须写清，否则同一批请求可能被分到不同日期。

不要在处理时使用“今天”来隐式补齐缺失时间，因为重跑会改变结果。需要默认值时把默认策略、基准时间和版本作为显式输入。稳定的数据管道应尽量做到相同输入与配置得到相同输出，除了单独记录的运行时间，不应因机器所在时区、语言或当前日期不同而悄悄变化。

第二个完整例子：清洗导出并稳定划分业务组#

下面程序读取一个明确的 CSV 合同，保留原文与规范化文本，以业务组稳定划分集合，并一次性写出 JSONL。保存为 normalize_export.py。它不会调用模型，唯一副作用是写入你明确指定的输出文件；默认拒绝覆盖已有文件。

# normalize_export.py
import argparse
import csv
import hashlib
import json
import os
import tempfile
import unicodedata
from pathlib import Path

def assign_split(group_id):
    # 固定算法让同一业务组在重复运行中落入同一集合
    bucket = int(hashlib.sha256(group_id.encode("utf-8")).hexdigest()[:8], 16) % 10
    return "test" if bucket == 0 else "validation" if bucket == 1 else "dev"

def load_rows(path):
    result, seen = [], set()
    with path.open(encoding="utf-8-sig", newline="") as source:
        reader = csv.DictReader(source)
        required = {"id", "group_id", "text", "label"}
        if set(reader.fieldnames or []) != required:
            raise ValueError("表头必须是 id、group_id、text、label")
        for number, row in enumerate(reader, start=2):
            if None in row or any(value is None for value in row.values()):
                raise ValueError(f"第 {number} 行列数不符")
            if not row["id"] or not row["group_id"]:
                raise ValueError(f"第 {number} 行主键或业务组为空")
            if row["id"] in seen:
                raise ValueError(f"第 {number} 行主键重复")
            if row["label"] not in {"refund", "other"}:
                raise ValueError(f"第 {number} 行标签无效")
            original = row["text"]
            normalized = unicodedata.normalize("NFC", original).strip()
            if not normalized:
                raise ValueError(f"第 {number} 行正文为空")
            seen.add(row["id"])
            result.append({
                **row, "original_text": original, "text": normalized,
                "split": assign_split(row["group_id"]),
                "text_sha256": hashlib.sha256(normalized.encode("utf-8")).hexdigest(),
            })
    if not result:
        raise ValueError("输入没有记录")
    return result

def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("input", type=Path)
    parser.add_argument("output", type=Path)
    args = parser.parse_args()
    if args.output.exists():
        raise FileExistsError("拒绝覆盖现有输出")
    rows = load_rows(args.input)  # 全部验证通过后才开始写
    args.output.parent.mkdir(parents=True, exist_ok=True)
    temporary = None
    try:
        with tempfile.NamedTemporaryFile(
            mode="w", encoding="utf-8", newline="\n",
            dir=args.output.parent, delete=False
        ) as target:
            temporary = Path(target.name)
            for row in rows:
                target.write(json.dumps(row, ensure_ascii=False) + "\n")
            target.flush()
            os.fsync(target.fileno())
        # 本练习假设单写者；并发发布需要额外的原子竞争控制
        os.replace(temporary, args.output)
        temporary = None
    finally:
        if temporary is not None:
            temporary.unlink(missing_ok=True)
    print(json.dumps({"records": len(rows), "output": str(args.output)}, ensure_ascii=False))

if __name__ == "__main__":
    main()

保存 source.csv 为以下内容；双引号是 CSV 语法的一部分，用于保留正文的首尾空格。

id,group_id,text,label
a,conversation-1," 我要退款 ",refund
b,conversation-1,"退款需要哪些资料",refund
c,conversation-2,"物流到哪里了",other

执行 python normalize_export.py source.csv cleaned.jsonl，预期 records 为三，输出文件有三行 JSON。a 的 original_text 保留空格，text 去掉首尾空格；a 与 b 的 split 必须相同，因为它们属于同一会话。再次向同一输出路径运行会拒绝覆盖，要比较结果应选择另一个文件名。

assign_split 是一个确定的分桶规则，不会因 Python 进程随机哈希变化而漂移。它也不保证只有三条样本时每个集合都有数据，因此必须另外检查分布。真正的版本发布可以先用此规则生成候选清单，再按组做受控调整并冻结清单；不应为了凑比例拆开同一会话。

load_rows 在写入前收集并验证全部数据，适合教学规模。输出先写同目录临时文件，再替换到最终路径，避免消费者读到半份结果。flush 把语言层缓冲交给操作系统，fsync 请求文件数据同步；这不等于跨所有文件系统和灾难情形的持久化保证。目录同步、并发写者和网络文件系统语义需要按目标环境设计。

第三个例子：错误数据不应产生“部分成功”#

把第二行的标签改为 unknown，选择一个新的输出路径执行程序。预期在校验阶段失败，最终输出文件不存在。再将 c 的 id 改为 a，观察重复主键错误。这个实验验证了一个重要属性：下游不会拿到一份缺了一行却没有错误标志的“成功数据集”。

若业务确实允许部分导入，输出合同必须分为成功记录与隔离记录，并写出 accepted、rejected、reason_counts 与总量守恒关系。比如读到一百条、接受九十七条、拒收三条，三个数字必须对得上。后续评测还要决定那三条是运行失败、待修数据还是明确排除，不能靠导入器偷偷决定分母。

对正在追加的日志文件，读到半行 JSON 可能表示写入尚未完成，而不一定永久损坏。离线评测通常应读取已经封存的导出文件；实时消费则需要带偏移、检查点和完整记录边界的设计。不要把一个临时日志文件直接当成稳定数据集反复统计。

从内存字典迁移到表格和数据库时保持语义#

DataFrame 的便利在于表达分组和关联，但自动类型推断可能改变主键、空值和时间。迁移时先比较输入记录数、唯一主键数、缺失值数和几个边界字段，再比较最终分数。只有最终平均数相同，不足以证明实现等价，因为不同错误可能互相抵消。

连接操作尤其需要谨慎。inner join 只保留两边都匹配的记录，如果预测缺失就会直接消失；left join 保留标签侧全部样本，更容易检查缺失。多对多连接会放大行数，一个重复预测可能使某道题被重复计分。连接之前建立唯一性约束，连接之后验证行数，比事后解释奇怪指标更可靠。

当数据不能全部放进内存时，SQLite 可以成为中间存储：为样本 id 建主键，为业务组和集合建索引，把预测写入另一张唯一主键表，再通过 SQL 检查缺失和多余记录。事务按合理批量提交，避免一行一个事务的开销，也避免一个超大事务占用过多资源。不要为了速度关闭所有持久性设置，再把生成结果当作已可靠保存。

增量处理还需要水位和版本。仅按“上次最后一条时间”继续可能漏掉迟到数据；仅按文件名也无法发现旧文件内容被修改。可以使用稳定事件 id 去重，记录读取范围和源指纹，并对迟到窗口重新计算。统计报告的“截止时间”与数据实际完整到的时间应分别描述。

保存足够的中间证据来解释一次分数变化#

一次评测至少关联三类文件：冻结标签、模型运行结果和判分结果。模型运行结果记录实际输出与调用状态，判分结果记录使用了哪个判分器以及为什么通过。它们的 id 应能互相连接，但不能用最后的通过标记覆盖模型原文。否则判分规则升级后只能重新调用模型，既增加费用，也无法公平比较同一批旧输出。

处理运行失败时，最好同时报告“系统完成率”和“已完成请求的质量”。例如一百条里八十条完成，完成的七十二条正确，条件正确率是九成，端到端正确完成率只有七成二。只报九成会掩盖二十条失败；把两者都列出，才能知道该优化网络可靠性还是模型质量。

数据里还可能包含重试产生的多个结果。必须先定义选择规则，例如以同一业务操作最终成功结果评质量，同时把所有尝试计入成本；或者固定只比较第一次尝试。不能每条挑最好的输出，因为线上用户未必得到那一个。选择策略应在运行前确定，并在报告中可追溯到实际请求记录。

综合练习：审计输出数据集而不是只检查能否解析#

练习要求读取 cleaned.jsonl，验证 id 唯一、同组不跨集合，并按 split 输出组数和记录数。不能只看每个集合有多少行，因为一个巨大业务组可能让分布看起来足够大却缺少独立性。

# audit_dataset.py
import json
import sys
from collections import defaultdict
from pathlib import Path

path = Path(sys.argv[1])
seen, group_split = set(), {}
counts, groups = defaultdict(int), defaultdict(set)
with path.open(encoding="utf-8") as source:
    for number, line in enumerate(source, start=1):
        row = json.loads(line)
        key, group, split = row["id"], row["group_id"], row["split"]
        if key in seen:
            raise ValueError(f"第 {number} 行主键重复")
        if split not in {"dev", "validation", "test"}:
            raise ValueError(f"第 {number} 行集合无效")
        if group in group_split and group_split[group] != split:
            raise ValueError("同一业务组跨集合")
        seen.add(key)
        group_split[group] = split
        counts[split] += 1
        groups[split].add(group)
report = {
    split: {"records": counts[split], "groups": len(groups[split])}
    for split in ("dev", "validation", "test")
}
print(json.dumps(report, ensure_ascii=False, indent=2))

数据管道的最终价值是让结果可追溯、可重跑、可解释。每次调整清洗规则，应在固定小样本上检查变化，再运行全量数据并比较拒收数量、集合分布和指标。把清洗规则也作为代码版本管理，而不是把修过的最终 CSV 当成唯一事实。这样模型效果发生变化时，才能区分模型改进与输入数据被悄悄改变。

练习、提示与参考解答#

练习：删除 t2 预测，加入重复 t1，再让 g3 同时出现在 dev 与 test。每个错误单独运行，记录程序在哪一步拒绝。

提示：不要一次制造多个错误，否则第一个异常会遮住后面的验证路径。

可验证验收与自测#

验收顺序打乱后分数不变、缺失预测失败、重复预测失败、跨集合业务组失败、指纹可复现。报告中同时写样本数量与指标，不把无效样本悄悄排除。

1. 为什么不能按行号关联结果？ 异步输出、失败和重试会改变顺序与数量，应使用稳定 id。
2. 哈希相同证明数据安全吗？ 只证明对应字节相同，不证明内容正确、合法或没有泄漏。
3. 清洗时发现异常就丢弃可以吗？ 只有合同明确允许并报告拒收范围时才可以；评测默认应暴露异常，防止选择性统计。