07 文件 IO 与数据序列化

一句话：文件读写真正要解决的是三件事：编码一致、失败不留半截文件、大数据不整块进内存；pathlib + 显式 UTF-8 + 原子替换 + JSONL/流式解析分别对应这三点。

适用：CPython 3.11+。Path.unlink(missing_ok=True) 需要 3.8+。

为什么需要

数据管道每天都在「读文件 → 解析 → 转换 → 写回」。最常见的故障不是语法，而是：同一份文件在 Windows 按 GBK 读、在 Linux 按 UTF-8 读，直接 UnicodeDecodeError；用 "w" 覆盖配置，写到一半进程崩溃，留下半截 JSON；用 json.load 读几 GB 数组，内存被打满；用 pickle 加载网络来源的模型，执行了任意代码。代价依次是「读不了」「数据损坏」「服务 OOM」「安全事件」。

核心机制

路径与文本 IO

from pathlib import Path

p = Path("data") / "raw" / "corpus.jsonl"
print(p.name, p.stem, p.suffix, p.parent)
for f in Path("data").rglob("*.jsonl"):
    print(f)
Path("outputs/run1").mkdir(parents=True, exist_ok=True)

with open("note.txt", "w", encoding="utf-8") as f:
    f.write("第一行\n第二行\n")
with open("note.txt", encoding="utf-8") as f:
    for line in f:                          # 文件对象是惰性迭代器
        print(line.rstrip("\n"))

• 永远显式写 encoding="utf-8"，不要依赖系统默认值。
• "w" 清空文件，"a" 追加，"x" 在文件已存在时报错；写 CSV 还要加 newline=""。
• 大文件用 for line in f，不要 f.read() 或 readlines() 一次装进内存。
• errors="replace" 只适合显示/容错清洗，会不可逆地丢字符；数据入库保持 strict。

utf-8-sig 与 BOM

带 BOM 的 UTF-8 文件开头是 EF BB BF。用 encoding="utf-8" 读取时，BOM 会变成第一个字符 ，表现为「第一个键名对不上」；用 utf-8-sig 读取会自动去掉 BOM（Path("from_excel.csv").read_text(encoding="utf-8-sig")）。反过来，用 utf-8-sig 写入会主动加 BOM，方便 Windows Excel 识别 UTF-8。规则：读别人给的文件用 utf-8-sig 兼容，自己内部流转统一 utf-8，不要混用。

JSON、JSONL 与流式解析

import json

text = json.dumps({"model": "demo"}, ensure_ascii=False, indent=2)
data = json.loads(text)

JSON 只支持对象、数组、字符串、数字、布尔、null；set、datetime、Decimal、NumPy 数组都要先转换。json.load 会把整个文档读进内存。

JSONL（每行一个 JSON）是流式场景的默认格式：可追加、可逐行解析、一行坏了不影响其他行。

with open("qa.jsonl", "w", encoding="utf-8") as f:
    for q, a in [("1+1?", "2"), ("Python?", "语言")]:
        f.write(json.dumps({"q": q, "a": a}, ensure_ascii=False) + "\n")

with open("qa.jsonl", encoding="utf-8") as f:
    for lineno, line in enumerate(f, 1):
        try:
            record = json.loads(line)
        except json.JSONDecodeError as e:
            print(f"第 {lineno} 行损坏: {e}")

单个大 JSON 数组不能用逐行解析：用第三方 ijson（pip install ijson）增量读取 for item in ijson.items(f, "item"): ...，或用 json.JSONDecoder.raw_decode 在缓冲区里逐个解码。能改格式就改 JSONL，不要为了「纯标准库」把几 GB 文件读进内存。

CSV、pickle 与原子写

import csv

with open("scores.csv", "w", newline="", encoding="utf-8") as f:
    w = csv.DictWriter(f, fieldnames=["model", "acc"])
    w.writeheader()
    w.writerow({"model": "a", "acc": 0.9})
with open("scores.csv", newline="", encoding="utf-8") as f:
    print([row["model"] for row in csv.DictReader(f)])

CSV 的值读出来都是字符串；类型转换、空值、引号由 csv 模块处理。Windows 上写 CSV 必须 newline=""，否则会多空行。pickle 能序列化任意 Python 对象，但 pickle.load 会执行对象还原代码——永远不要加载不可信来源。模型权重用 torch.load(..., weights_only=True)；跨语言交换用 JSON/Protobuf。

import os, tempfile

def atomic_write_json(path: Path, data) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    fd, tmp = tempfile.mkstemp(dir=path.parent, prefix=path.name, suffix=".tmp")
    try:
        with os.fdopen(fd, "w", encoding="utf-8") as f:
            json.dump(data, f, ensure_ascii=False)
            f.flush(); os.fsync(f.fileno())
        os.replace(tmp, path)          # 同一文件系统内原子替换
    except BaseException:
        Path(tmp).unlink(missing_ok=True)
        raise

临时文件必须和目标文件在同一目录/同一文件系统，os.replace 才能原子替换；先 flush + fsync 再替换。Windows 上目标文件被占用时替换可能失败；目录级 fsync 不可用，具体语义以官方文档为准。

最小上手

保存为 io_demo.py 后运行，预期输出 [0, 1, 2] 和 {'lr': 0.0}：

import json, os, tempfile
from pathlib import Path

work = Path("io_demo"); work.mkdir(exist_ok=True)
with (work / "rows.jsonl").open("w", encoding="utf-8") as f:
    for i in range(3):
        f.write(json.dumps({"id": i}) + "\n")
with (work / "rows.jsonl").open(encoding="utf-8") as f:
    print([json.loads(line)["id"] for line in f])

def atomic_write_json(path, data):
    fd, tmp = tempfile.mkstemp(dir=path.parent, prefix=path.name, suffix=".tmp")
    try:
        with os.fdopen(fd, "w", encoding="utf-8") as f:
            json.dump(data, f, ensure_ascii=False)
            f.flush(); os.fsync(f.fileno())
        os.replace(tmp, path)
    except BaseException:
        Path(tmp).unlink(missing_ok=True); raise

atomic_write_json(work / "config.json", {"lr": 0.0})
print(json.loads((work / "config.json").read_text(encoding="utf-8")))

生产动作

• 编码与写入：内部统一 UTF-8，读外部文件先尝试 utf-8-sig，错误信息带上路径和编码；配置、检查点、索引文件用「临时文件 + fsync + os.replace」，日志用追加模式，允许最后一行不完整。
• 大文件：JSONL 逐行；单数组用 ijson；CSV 流式；不要把 read() 放进循环。
• 安全：不加载不可信 pickle；模型文件优先 weights_only=True；JSON 解析失败保留行号。
• 排障三段：
  • 现象：UnicodeDecodeError → 命令：Path(p).read_bytes()[:3] 看 BOM，尝试 utf-8-sig/gbk → 判断：编码是否与写入方一致。
  • 现象：首行键名多了  → 命令：print(repr(line[:10])) → 判断：是否用 utf-8 读了带 BOM 的文件。
  • 现象：JSON 解析到一半失败 → 命令：看 JSONDecodeError.lineno/colno，逐行验证 JSONL → 判断：是整文件截断还是单行损坏。

边界

• JSON 不适合超大单文档、二进制、循环引用；大数据考虑 Parquet/数据库。JSONL 不是合法 JSON 文档，不能直接 json.load。
• CSV 没有类型和嵌套结构；复杂表格用 Parquet 或数据库。
• pickle 是 Python 专属且不安全；跨语言、跨版本、对外接口都不要用。

讲清楚

文件 IO 的核心不是 open 的语法，而是编码、原子性、内存三个约束。编码上永远显式 UTF-8，读外部文件用 utf-8-sig 兼容 BOM，但内部流转不要混用；写入配置和检查点时用同目录临时文件 + fsync + os.replace，避免崩溃留下半截文件；大 JSON 数组不要 json.load，能改 JSONL 就逐行读，不能改就用 ijson 增量解析。CSV 记得 newline=""，所有值都是字符串；pickle 只用于可信数据，模型权重优先 weights_only=True。

适用 CPython 3.11+ | 初稿 2026-08，2026-10 压缩并补原子写、BOM 与流式解析