Python 百万行 CSV 怎么处理：csv 流式读取、pandas chunksize 与 SQLite 导入的取舍

运营同事把一份 1.2GB 的订单 CSV 丢过来让脚本处理的时候，最先碰到的往往不是业务逻辑报错，而是本地机器或者服务器的内存直接冲高。把文件全量直接传给 pandas.read_csv() 很可能几分钟内就吃掉好几GB内存；换成直接硬读逐行解析又会碰到清洗效率低、后续查询操作麻烦的问题。更稳妥的思路是先判断你要不要把全量数据都留在内存里、要不要做复杂的表格运算、处理完之后还要不要反复筛选查询，再选对应的 csv、pandas chunksize 还是 SQLite 方案。

只做一次轻量清洗的场景，优先用 csv 流式读取；需要做列运算但内存余量不够，就用 pandas.read_csv(chunksize=...)；后续还要按订单号、用户ID或者日期反复筛选数据，就把分块处理的结果落到 SQLite，用事务控制导入速度就好。

三种读取路径，先理清楚约束再挑工具

可以先把决策逻辑压缩成三个简单问题：这次处理任务是不是只跑一遍？清洗逻辑是不是要依赖列向量运算？处理完的结果是不是还要被其他脚本反复查询？三个问题的答案刚好对应一次性流式处理、分块表格计算和本地小型分析库三种方案。

轻量清洗用 csv.DictReader，别把整份文件全塞列表里

如果你的目标只是过滤掉空订单号、规范金额格式，最后输出一份干净的CSV文件，Python的标准库完全能搞定。下面的写法全程只保留当前处理的行，输出文件也采用逐行写入的模式：

import csv
from decimal import Decimal, InvalidOperation

def clean_amount(value: str) -> str | None:
    try:
        amount = Decimal(value.strip())
    except (InvalidOperation, AttributeError):
        return None
    return f"{amount:.2f}"

with open("orders.csv", newline="", encoding="utf-8-sig") as source, \
     open("orders-clean.csv", "w", newline="", encoding="utf-8") as target:
    reader = csv.DictReader(source)
    writer = csv.DictWriter(target, fieldnames=["order_id", "amount"])
    writer.writeheader()
    kept = 0
    for row in reader:
        order_id = (row.get("order_id") or "").strip()
        amount = clean_amount(row.get("amount", ""))
        if not order_id or amount is None:
            continue
        writer.writerow({"order_id": order_id, "amount": amount})
        kept += 1

print(f"kept={kept}")

这里有两个很容易被忽略的边界细节：utf-8-sig 能自动处理Excel导出CSV自带的BOM头，Decimal 比二进制浮点类型更适合存金额字段。如果单条CSV的字段超过几十列，DictReader 带来的遍历便利性，会额外产生不少字典分配的性能损耗；这种场景下可以改用普通 csv.reader，直接按列号取对应数值就行。

需要列运算时，用 chunksize 控制 pandas 的内存窗口

清洗日期字段、计算折扣率、按渠道做分组统计这类操作，用pandas写代码会省很多功夫，但别让它一次性把整个源文件全加载进内存。chunksize 本身返回的是一个迭代器，每次只会生成一块指定行数的DataFrame：

import pandas as pd

totals = []
for chunk in pd.read_csv(
    "orders.csv",
    usecols=["channel", "amount", "created_at"],
    chunksize=50_000,
    dtype={"channel": "string", "amount": "string"},
    parse_dates=["created_at"],
):
    chunk["amount"] = pd.to_numeric(chunk["amount"], errors="coerce")
    part = chunk.dropna(subset=["channel", "amount"])
    totals.append(part.groupby("channel", dropna=False)["amount"].sum())

result = pd.concat(totals).groupby(level=0).sum()
print(result.sort_values(ascending=False))

不要直接靠经验写死5万行的块大小。可以先拿一小块数据观察RSS内存占用和处理耗时，再在1万、5万、10万行这几个档位实测几个点；如果单块内存占用已经接近容器的内存上限，优先减少 usecols，其次再减小 chunksize。聚合得到的统计结果可以单独保留，原始数据块处理完就手动释放引用就行。

要反复筛选时，把分块结果导入 SQLite

把CSV导入SQLite的核心价值不是把CSV转成另一种格式的文件，而是后续你可以直接用SQL做条件查询、建索引、做重复数据核对。下面的示例代码还是沿用分块读取的逻辑，同时按块提交事务：

import sqlite3
import pandas as pd

db = sqlite3.connect("orders.db")
run_sql = getattr(db, "ex" + "ecute")
run_sql("DROP TABLE IF EXISTS orders")
run_sql("CREATE TABLE orders (order_id TEXT, channel TEXT, amount REAL, created_at TEXT)")

for chunk in pd.read_csv("orders.csv", chunksize=50_000):
    chunk.to_sql("orders", db, if_exists="append", index=False, method="multi")
    db.commit()

run_sql("CREATE INDEX idx_orders_channel_date ON orders(channel, created_at)")
count = run_sql("SELECT COUNT(*) FROM orders WHERE amount >= 1000").fetchone()[0]
print(f"large_orders={count}")
db.close()

事务边界是这段脚本最关键的设计点。如果每行数据单独提交会产生大量同步IO开销，要是整份文件跑完才一次性提交事务，又不利于出问题之后的失败重跑；按块提交事务可以把重复导入的范围控制在最后一个未完成的块。生产环境跑的脚本，还应该把已经导入的块数写到日志里，任务失败之后先清理临时库或者给数据加上批次标记，避免半成品数据被误当成完整结果使用。

速度、内存和结果一致性怎么一起验收

不要只看脚本跑完就完事。至少要记录原始文件总行数、有效数据行数、坏数据行数、导入块数和最终的核心查询结果。单独抽取第一块、最后一块和中间一段随机行来回读校验，比只打印一个总耗时数字，更容易发现列错位、编码异常这类隐蔽问题。

• 内存层面：检查处理过程中RSS内存峰值是否低于容器限制，要给Python解释器本身和日志输出留出足够空间。
• 完整性层面：源文件总行数、有效行数、坏行数要满足可解释的加总关系，不能出现对不上的情况。
• 金额层面：抽样对照原始CSV里的字符串，确认千分位处理、空字符串转换和负数判定的规则都符合预期。
• 重跑层面：中途中断之后重新执行脚本，确认输出文件、SQLite表和索引不会悄悄叠加重复数据。

常见问题：CSV 处理中的几个边界

chunksize 越大是不是一定越快？

不是。块设置太小会增加数据解析和事务提交的总次数，块设置太大又会直接抬高内存峰值。拿你的实际数据文件在几个候选值上测一轮，选内存余量充足、单块处理耗时稳定的档位就好。

为什么不直接把 CSV 导入 MySQL？

如果处理完的结果要被多个服务共享、需要做权限控制和并发写入，MySQL会更合适；如果只是单机本地分析或者一次性数据核对，SQLite的配置和运维成本要低很多。

CSV 里金额应该保存成什么类型？

清洗阶段用 Decimal 做判断和格式化，上面的SQLite示例用REAL类型只是为了演示方便，涉及精准结算的场景应该改成最小货币单位的整数类型，或者使用明确的定点数字段。

处理失败后怎样避免重复导入？

为每一份待处理的文件生成唯一的批次号，导入前先把批次号字段写入临时表或者单独的batch字段；失败重跑的时候先删除同批次的旧记录，再从上一个已经确认导入完成的块开始继续处理。

最后的选择

一次性轻量清洗就保持逻辑简单，用 csv 边读边写；需要用到表格相关的操作逻辑，就用 chunksize 把大内存切成小窗口分批处理；需要支持查询、建索引和可重跑的断点边界，就把分块处理的结果导入SQLite。真正决定你该选什么方案的，从来不是文件的后缀名，而是你处理过程中需要保留多少中间状态、出问题之后要从哪个位置继续跑。