Python 批处理一跑就吃满内存把 CSV 汇总改成流式任务财务给你一份 CSV让你按部门汇总金额。小样本跑得很顺换成整月导出后脚本的内存持续上涨甚至把同机服务拖慢。你给机器加了内存下一次文件更大问题又回来。这类任务不一定需要先换框架。先看看数据什么时候进入内存、哪些对象一直活到结束、最终结果何时替换旧文件。三个边界没画清楚换成更复杂的工具也可能只是延后故障。本文用 Python 标准库做一个可以直接运行的 CSV 汇总工具逐条解析记录以整数分累加金额限制分组数量全部成功后再提交输出。它不追求一套代码包办所有报表而是把一项小任务的资源和失败行为说清楚。一、内存问题经常出在这两行下面的写法很常见rowslist(csv.DictReader(file))totalsaggregate(rows)DictReader原本可以迭代读取外面套上list所有记录就留在内存里了。原始文件的字节数也不等于对象占用每行变成字典、键值字符串容器本身还有管理开销。不要凭文件大小猜内存够不够也不要直接宣布“换成生成器就解决了”。生成器只约束读取端。如果汇总字典的键是每条记录都不同的订单号结果端仍会随记录数增长。可以用两个变量描述这件事输入有 N 条记录最终有 K 个分组。全量加载会保留 N 条记录流式汇总只保留当前记录和 K 个累计值。它削掉的是整份输入的常驻副本不是承诺内存恒定。分组标签长度、单个字段大小和整数位数也会影响实际占用。对于按部门、门店或日期汇总K 通常更容易设置业务上限。对于按用户或订单明细保留全部结果K 可能接近 N应考虑磁盘数据库、分区文件或分批落盘而不是无限放大字典。二、先约定输入不要让脚本猜业务这个示例只接收两列顺序固定为department,amount。金额单位是元允许负数用于冲销最多两位小数不接收科学计数法、千位分隔符或空值。这里只做加总不决定哪些业务单据该冲销。边界本文示例的处理文件编码UTF-8可带 BOM部门名称去掉首尾空白非空且不超过 80 个字符金额ASCII 十进制文本长度不超过 20最多两位小数坏记录立即失败保留已有输出分组数量由参数设置上限超过即中止CSV 方言逗号分隔使用标准双引号规则金额不直接使用float累加。十进制小数转换为二进制浮点时可能不能精确表示这里先用Decimal解析已校验的短文本再转换为整数分。加总过程中使用整数不进行隐含四舍五入。还要区分“一行文本”和“一条 CSV 记录”。带引号的字段可以包含换行因此不应该用line.split(,)代替 CSV 解析器。打开文件时使用newline让csv处理换行细节错误提示中的reader.line_num是已读取的物理行位置不一定等于业务记录编号。三、完整示例读取、校验、汇总、提交适用 Python 3.10只使用标准库。保存为csv_totals.py。这是本地、单进程示例不包含并发写入协调输出文件所在目录需要事先存在。importargparseimportcsvimportosimportreimporttempfilefromdecimalimportDecimalfrompathlibimportPath MONEYre.compile(r[-]?[0-9](?:\.[0-9]{1,2})?)defamount_to_cents(raw):iflen(raw)20ornotMONEY.fullmatch(raw):raiseValueError(金额应为最多两位小数的十进制文本)returnint(Decimal(raw)*100)defsummarize(source,target,max_groups):source,targetPath(source),Path(target)ifmax_groups0:raiseValueError(max_groups 必须为正数)ifsource.resolve()target.resolve():raiseValueError(输入和输出不能是同一文件)csv.field_size_limit(64*1024)totals{}records0withsource.open(encodingutf-8-sig,newline)asfile:readercsv.DictReader(file,strictTrue)ifreader.fieldnames![department,amount]:raiseValueError(表头必须为 department,amount)forrowinreader:try:ifNoneinroworany(visNoneforvinrow.values()):raiseValueError(列数不匹配)departmentrow[department].strip()ifnotdepartmentorlen(department)80:raiseValueError(部门名称为空或超过80个字符)centsamount_to_cents(row[amount].strip())ifdepartmentnotintotalsandlen(totals)max_groups:raiseValueError(分组数量超过上限)totals[department]totals.get(department,0)cents records1exceptValueErrorasexc:raiseValueError(f读取到物理行{reader.line_num}:{exc})fromexc temporaryNonetry:withtempfile.NamedTemporaryFile(modew,encodingutf-8,newline,dirtarget.parent,prefix.csv-totals-,suffix.tmp,deleteFalse,)asfile:temporaryPath(file.name)writercsv.writer(file)writer.writerow([department,amount_cents])fordepartment,centsintotals.items():writer.writerow([department,cents])os.replace(temporary,target)finally:iftemporaryisnotNoneandtemporary.exists():temporary.unlink()returnrecords,len(totals)if__name____main__:parserargparse.ArgumentParser()parser.add_argument(source)parser.add_argument(target)parser.add_argument(--max-groups,typeint,default10000)argsparser.parse_args()records,groupssummarize(args.source,args.target,args.max_groups)print(f处理完成{records}条记录{groups}个分组)输入示例保存为expenses.csvdepartment,amount 研发,0.10 研发,0.20 研发,-0.05 运维,12.30执行python csv_totals.py expenses.csv totals.csv --max-groups 100输出字段特意叫amount_cents而不是含糊的amount。研发部门的累计值为25运维为1230都是分下游需要元时再按业务规则格式化。验证覆盖了这组精确累加以及后文列出的异常输入不包含大文件性能基准。四、为什么输出也要有边界如果读取一条就直接写最终文件中途遇到错误目录里可能留下看起来正常、实际只算了一半的报表。用户只看到文件存在后续任务也可能误读它。示例先完成输入校验与汇总再在目标目录写临时文件关闭文件后调用os.replace替换目标。这让“新报表已提交”有明确的操作边界也避免因跨文件系统移动造成另一种失败路径。这里的承诺要克制在通常支持原子替换的本地文件系统上替换成功后读者得到完整的新文件它不等于断电持久性保证也不是分布式事务。示例没有执行文件与目录的同步落盘不能据此宣称突然断电也绝不丢结果。多进程同时生成同一个目标文件还可能出现后完成的任务覆盖先完成的任务。生产环境应使用任务锁、批次命名或版本化输出网络共享盘的语义也要单独验证。临时文件解决的是部分写入暴露不自动解决所有并发问题。五、不要只测“正常跑完”这个工具至少需要检验以下路径0.10 0.20 - 0.05得到 25 分避免金额单位和符号出错。UTF-8 BOM、带逗号或换行的部门字段能按 CSV 规则处理。三位小数、空金额、非数字和列数异常会失败。新分组超过上限时退出已有结果文件保持原样。输出路径与输入路径相同时拒绝执行避免覆盖输入。这些路径已在临时目录中运行校验。它们说明示例的功能与失败行为符合约定不说明它在你的机器上能处理多少数据、节省多少内存。真实容量需要用代表性文件测量记录耗时、峰值内存、实际分组数和最长字段最好在独立进程里测试。还有一个容易遗漏的点CSV 被后续表格软件打开时文本字段可能触发表格公式解释。csv.writer的正确转义不等于公式安全。本文输出适用于程序消费若直接分发给人工打开应按目标软件规则处理以、、-、等字符开头的文本字段或者使用具有明确文本单元格类型的输出格式。六、今天就能做的四步改造先在现有脚本里找list(reader)、readlines()和全量中间结果画出对象的存活范围。再确认分组键是不是受控集合如果它来自任意用户输入上限必须显式配置。然后补齐两类测试一类核对金额和编码另一类故意制造坏记录检查旧输出是否仍然完整。最后才用真实文件比较资源占用不要把“小样本正确”与“大规模可靠”当作同一件事。把这些边界写出来之后工具选择反而更容易。分组少、处理线性、单机离线用标准库可以先把任务做清楚需要复杂联表、多维分析或高基数状态就评估更合适的数据引擎。Python 效率工具的价值不只是省掉重复操作更是让输入、资源和失败都可预测。