影刀RPA CSV大文件处理：千万行数据的读取与转换

影刀RPA CSV大文件处理千万行数据的读取与转换作者林焱什么情况用什么日常工作中总有人用Excel打开一个50MB的CSV然后电脑卡死。更惨的是上司让你把A表和B表按客户ID关联挑出没有匹配的记录——A表200万行B表150万行Excel直接报文件过大无法打开。这时候不要硬扛Excel。影刀Python的组合才是正确姿势用影刀做文件采集和流程调度用Python的pandas做数据处理。CSV处理在RPA场景中的典型需求多CSV合并每天生成的日志文件合并成月报CSV大文件筛选从几百万行里抽出特定条件的数据CSV格式转换分号分隔转逗号GBK转UTF-8CSV与Excel互转怎么做第一步分块读取大文件拼多多店群自动化上架方案对于几百MB甚至几GB的CSV一次性读完内存直接炸。importpandasaspd chunk_size100000# 每次读10万行chunks[]fori,chunkinenumerate(pd.read_csv(rD:\big_data.csv,chunksizechunk_size,encodingutf-8)):# 只保留需要的列chunkchunk[[用户ID,订单金额,下单时间]]# 只保留特定条件chunkchunk[chunk[订单金额]100]chunks.append(chunk)print(f已处理第{i1}批共{len(chunk)}行)# 合并所有分块resultpd.concat(chunks,ignore_indexTrue)result.to_csv(rD:\filtered_data.csv,indexFalse)第二步多CSV合并影刀从FTP或邮件里拉了一堆CSV文件下来需要合并。影刀操作步骤1. 【遍历文件夹】遍历 D:\csv_files\ 下所有.csv文件 2. 对每个文件【执行Python】调用合并脚本importosimportpandasaspd folderrD:\csv_filesall_data[]forfinos.listdir(folder):iff.endswith(.csv):dfpd.read_csv(os.path.join(folder,f))df[来源文件]f# 标记数据来源all_data.append(df)print(f已加载:{f},{len(df)}行)mergedpd.concat(all_data,ignore_indexTrue)merged.to_csv(rD:\merged_output.csv,indexFalse)print(f合并完成共{len(merged)}行)踩过的坑不同CSV文件的列可能不一样多。比如A文件有8列B文件有9列。pd.concat默认对齐列名多出来的列填NaN。如果字段顺序不一致但要按位置合并用pd.concat(all_data, ignore_indexTrue, axis0)是不够的——需要先统一列名。第三步编码与分隔符处理中文CSV最常见的坑是编码问题和分隔符不一致。importchardet# 自动检测编码withopen(rD:\unknown.csv,rb)asf:rawf.read(10000)resultchardet.detect(raw)encodingresult[encoding]# 如 GB2312 或 utf-8# 用检测到的编码读dfpd.read_csv(rD:\unknown.csv,encodingencoding)# 自动检测分隔符importcsvwithopen(rD:\unknown.csv,r,encodingencoding)asf:dialectcsv.Sniffer().sniff(f.read(5000))delimiterdialect.delimiter# 如 , 或 ; 或 \tdfpd.read_csv(rD:\unknown.csv,encodingencoding,sepdelimiter)第四步CSV与Excel互转CSV转Excel带格式影刀操作步骤1. 【执行Python】读取CSV并写出xlsx 2. 【打开Excel】打开生成的xlsx文件 3. 【自动调整列宽】 4. 【设置表头加粗】 5. 【保存】importpandasaspdfromopenpyxlimportload_workbookfromopenpyxl.stylesimportFont,Alignment,PatternFill# CSV读入dfpd.read_csv(rD:\data.csv)# 写出xlsxdf.to_excel(rD:\data.xlsx,indexFalse,sheet_name数据)# 美化wbload_workbook(rD:\data.xlsx)wswb.active# 表头加粗背景色forcellinws[1]:cell.fontFont(boldTrue)cell.fillPatternFill(start_color4472C4,end_color4472C4,fill_typesolid)cell.fontFont(boldTrue,colorFFFFFF)# 自动列宽forcolinws.columns:max_lenmax(len(str(cell.valueor))forcellincol)ws.column_dimensions[col[0].column_letter].widthmin(max_len2,50)wb.save(rD:\data.xlsx)有什么坑坑1chunksize太小导致处理极慢TEMU店群如何管理运营chunksize设成1000行一个2000万行的文件要循环20000次每次pandas都要重新初始化各种内部结构。推荐设成50000-200000行在内存和速度之间取平衡。坑2dtype推断错误pandas自动推断列类型时可能把00123推断为整数123丢失前导零。solution读的时候指定dtypedfpd.read_csv(file,dtype{工号:str,手机号:str})坑3特殊字符导致解析失败CSV中某行包含未转义的引号或换行符会导致解析错位。用error_bad_linesFalse跳过问题行但可能丢数据更好的是加quotingcsv.QUOTE_ALL规范导出。坑4内存溢出即使分块处理如果每个chunk都在accumulate累积总内存占用还是会飙升。对于不需要全量结果的场景如只统计不保留细节在chunk内部做完聚合就扔掉total0forchunkinpd.read_csv(file,chunksize100000):totalchunk[金额].sum()# chunk在这之后被GC回收print(f总金额:{total})总结大CSV处理的核心原则只有一条不要试图一次性把所有数据装进内存。分块读、流式处理、及时释放这三个做到就不会翻车。