Python 读取大 CSV 怎么避免内存峰值：分块迭代、类型推断与失败行处理

CSV 文件一旦从几百 MB 涨到几 GB，你就不能再像处理小文件那样随手写一行 pandas 代码直接全读，很容易触发内存溢出，原本简单的需求直接变成了需要仔细设计的小型数据管道问题：要不要一次性把全量文件加载进内存、字符串列要不要手动指定类型、异常行怎么留存留证，每一步选择都会直接影响内存峰值。比较稳的做法是先验证表头合法性，再按固定块大小迭代读取，提前给关键字段显式指定类型；每个数据块完成校验处理后立刻释放相关临时对象，处理失败的记录单独写入隔离文件留存。

先判断：内存峰值是文件大小，还是列类型造成的

同一个 2 GB CSV，全部列都作为字符串读入，和只保留必要列、为数值列指定类型，占用的内存峰值可能完全不同。CSV 本身没有内置列类型元数据，读取器通常要先扫描部分样本再猜测字段类型；含有空值、同时混合数字和文本的列尤其容易退化成占用空间更高的对象列。

这也是为什么只把 chunksize 从 100,000 调到 10,000，结果仍可能不理想：每一块里的对象列仍然会占用大量 Python 对象空间。这里先别急着改块大小，先把字段合同写出来。

从表头和样本开始做输入校验

生产环境做数据导入不要直接把整个大文件丢给全量读取逻辑，先检查编码、必需字段是否存在，再抽取少量样本预览，能在文件结构不符合要求的时候第一时间终止流程，避免不必要的资源浪费。

from pathlib import Path
import pandas as pd

path = Path("incoming/orders.csv")
required = {"order_id", "created_at", "amount", "postal_code"}

sample = pd.read_csv(path, nrows=2000, encoding="utf-8-sig")
missing = required - set(sample.columns)
if missing:
    raise ValueError(f"missing columns: {sorted(missing)}")

print(sample.dtypes)
print(sample[["order_id", "postal_code"]].head(3))

nrows 只用于探测，不代表全文件通过。样本中的日期格式和金额格式还需要在真正的分块循环里继续核对，因为坏数据往往出现在导出文件的尾部。

分块读取时显式约束关键列类型

下面这段代码只保留导入所需列，并把不能丢前导零的标识列设为 string。金额列先读成可转换的数值，转换失败时保留为缺失值，之后由业务规则决定是否拒绝。

import pandas as pd

usecols = ["order_id", "created_at", "amount", "postal_code"]
dtype = {
    "order_id": "string",
    "postal_code": "string",
}

accepted = []
rejected = []

for chunk_no, chunk in enumerate(
    pd.read_csv(
        "incoming/orders.csv",
        usecols=usecols,
        dtype=dtype,
        chunksize=50_000,
        encoding="utf-8-sig",
    ),
    start=1,
):
    chunk["amount_num"] = pd.to_numeric(chunk["amount"], errors="coerce")
    bad = chunk["order_id"].isna() | chunk["amount_num"].isna()

    rejected.append(chunk.loc[bad, ["order_id", "amount"]].assign(chunk_no=chunk_no))
    accepted.append(chunk.loc[~bad].assign(chunk_no=chunk_no))

result = pd.concat(accepted, ignore_index=True)
errors = pd.concat(rejected, ignore_index=True)
result.to_parquet("out/orders.parquet", index=False)
errors.to_csv("out/orders-rejected.csv", index=False)

这里的 accepted 和 rejected 仍会把所有块保存在列表中，只适合演示对账逻辑；如果文件很大，应改成每块追加到 Parquet 或数据库，再只累计行数。分块读取不等于自动低内存，是否保留块才是后半段的关键。

真正低峰值的写法：每块处理、每块落盘

当目标是把多个 CSV 合并成一个结果文件，推荐让每块处理完尽快从内存释放。以 Parquet 格式为例，可以使用支持追加或分区写入的实现；如果项目暂时只用到 pandas，也可以先写临时 CSV，最后再合并，核心是不要把所有处理过的块对象全部塞进列表常驻内存。

from pathlib import Path
import gc
import pandas as pd

out_dir = Path("out/orders-parts")
out_dir.mkdir(parents=True, exist_ok=True)
accepted_rows = rejected_rows = 0

for chunk_no, chunk in enumerate(pd.read_csv(
    "incoming/orders.csv",
    usecols=usecols,
    dtype=dtype,
    chunksize=20_000,
    encoding="utf-8-sig",
), start=1):
    chunk["amount_num"] = pd.to_numeric(chunk["amount"], errors="coerce")
    bad = chunk["order_id"].isna() | chunk["amount_num"].isna()

    chunk.loc[~bad].to_csv(out_dir / f"accepted-{chunk_no:05d}.csv", index=False)
    chunk.loc[bad].to_csv(out_dir / f"rejected-{chunk_no:05d}.csv", index=False)
    accepted_rows += int((~bad).sum())
    rejected_rows += int(bad.sum())

    del chunk, bad
    gc.collect()

print({"accepted": accepted_rows, "rejected": rejected_rows})

gc.collect() 不是万能的降内存按钮；它只是让临时对象更快进入可回收路径。先删除引用、避免全局列表，再用进程级峰值监控确认效果，比盲目频繁调用垃圾回收更重要。

坏行怎么留证：原始内容、行号和原因要能对上

如果只使用 on_bad_lines="skip"，导入完成后很难回答“少了哪些订单”。对账文件至少需要批次号、业务主键、原因和原始行的可追踪信息。对格式损坏的 CSV 行，可以先用 Python 的 csv 模块按物理行扫描，把行号写入隔离文件，再交给 pandas 处理结构合格的输入。

业务字段校验逻辑可以直接放在块循环中执行：订单号为空、金额无法正常转换、日期超出允许范围，都应该标注成明确的拒绝原因。拒绝不等于直接丢弃数据，隔离文件应和本次导入的批次 ID 一起存储，直到全链路对账完成之后再做后续处理。

用三组数字验收一轮导入

验收时至少要使用一份包含空值、超长文本、错误金额、带前导零编码和引号换行的测试文件。只拿“每行都很规整”的样本测出来的低峰值，不能代表真实业务场景下导出的复杂文件。

常见问题

chunksize 越小，内存就一定越低吗？

不一定。块设置得越小反而会增加额外的读写次数；如果代码里把所有处理完的块都留在列表里，最终内存还是会持续累积。需要同时检查列类型定义、临时对象释放和落盘策略是否都符合预期。

为什么 postal_code 要用 string？

邮编、订单号这类标识字段本身不需要做数学运算，不该读成数值类型。读成整数可能直接丢掉前导零，后续出现空值的时候还会被隐式转换成浮点表示引发精度问题。

可以直接跳过坏行吗？

只有在业务明确允许丢弃异常数据、且后续可以从原始源文件完整重放时才可以这么做。更稳妥的做法是把坏行、行号和报错原因一起写入隔离文件，完成全量对账后再决定是修复重录还是直接跳过。

把导入结果留成可重放的批次

大 CSV 导入的核心不是找到一个适配所有场景的“神奇块大小”，而是把输入读取、类型约束、分块处理、失败隔离和资源峰值都变成可核对的可追溯记录。批次号、源文件哈希、输入行数、接收行数、拒绝行数和峰值内存数据都齐全之后，下一次重跑任务时你才能快速定位哪里发生了变化。