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

作者：林焱

什么情况用什么

日常工作中总有人用Excel打开一个50MB的CSV，然后电脑卡死。更惨的是，上司让你"把A表和B表按客户ID关联，挑出没有匹配的记录"——A表200万行，B表150万行，Excel直接报"文件过大无法打开"。

这时候不要硬扛Excel。影刀+Python的组合才是正确姿势：用影刀做文件采集和流程调度，用Python的pandas做数据处理。CSV处理在RPA场景中的典型需求：

• 多CSV合并（每天生成的日志文件合并成月报）

• CSV大文件筛选（从几百万行里抽出特定条件的数据）

• CSV格式转换（分号分隔转逗号，GBK转UTF-8）

• CSV与Excel互转

怎么做

第一步：分块读取大文件

对于几百MB甚至几GB的CSV，一次性读完内存直接炸。

importpandasaspd chunk_size=100000# 每次读10万行chunks=[]fori,chunkinenumerate(pd.read_csv(r'D:\big_data.csv',chunksize=chunk_size,encoding='utf-8')):# 只保留需要的列chunk=chunk[['用户ID','订单金额','下单时间']]# 只保留特定条件chunk=chunk[chunk['订单金额']>100]chunks.append(chunk)print(f'已处理第{i+1}批，共{len(chunk)}行')# 合并所有分块result=pd.concat(chunks,ignore_index=True)result.to_csv(r'D:\filtered_data.csv',index=False)

第二步：多CSV合并

影刀从FTP或邮件里拉了一堆CSV文件下来，需要合并。

影刀操作步骤：

1. 【遍历文件夹】遍历 D:\csv_files\ 下所有.csv文件 2. 对每个文件，【执行Python】调用合并脚本

importosimportpandasaspd folder=r'D:\csv_files'all_data=[]forfinos.listdir(folder):iff.endswith('.csv'):df=pd.read_csv(os.path.join(folder,f))df['来源文件']=f# 标记数据来源all_data.append(df)print(f'已加载:{f},{len(df)}行')merged=pd.concat(all_data,ignore_index=True)merged.to_csv(r'D:\merged_output.csv',index=False)print(f'合并完成，共{len(merged)}行')

踩过的坑：不同CSV文件的列可能不一样多。比如A文件有8列，B文件有9列。pd.concat默认对齐列名，多出来的列填NaN。如果字段顺序不一致但要按位置合并，用pd.concat(all_data, ignore_index=True, axis=0)是不够的——需要先统一列名。

第三步：编码与分隔符处理

中文CSV最常见的坑是编码问题和分隔符不一致。

importchardet# 自动检测编码withopen(r'D:\unknown.csv','rb')asf:raw=f.read(10000)result=chardet.detect(raw)encoding=result['encoding']# 如 'GB2312' 或 'utf-8'# 用检测到的编码读df=pd.read_csv(r'D:\unknown.csv',encoding=encoding)# 自动检测分隔符importcsvwithopen(r'D:\unknown.csv','r',encoding=encoding)asf:dialect=csv.Sniffer().sniff(f.read(5000))delimiter=dialect.delimiter# 如 ',' 或 ';' 或 '\t'df=pd.read_csv(r'D:\unknown.csv',encoding=encoding,sep=delimiter)

第四步：CSV与Excel互转

CSV转Excel（带格式）：

影刀操作步骤：

1. 【执行Python】读取CSV并写出xlsx 2. 【打开Excel】打开生成的xlsx文件 3. 【自动调整列宽】 4. 【设置表头加粗】 5. 【保存】

importpandasaspdfromopenpyxlimportload_workbookfromopenpyxl.stylesimportFont,Alignment,PatternFill# CSV读入df=pd.read_csv(r'D:\data.csv')# 写出xlsxdf.to_excel(r'D:\data.xlsx',index=False,sheet_name='数据')# 美化wb=load_workbook(r'D:\data.xlsx')ws=wb.active# 表头加粗+背景色forcellinws[1]:cell.font=Font(bold=True)cell.fill=PatternFill(start_color='4472C4',end_color='4472C4',fill_type='solid')cell.font=Font(bold=True,color='FFFFFF')# 自动列宽forcolinws.columns:max_len=max(len(str(cell.valueor''))forcellincol)ws.column_dimensions[col[0].column_letter].width=min(max_len+2,50)wb.save(r'D:\data.xlsx')

有什么坑

坑1：chunksize太小导致处理极慢

chunksize设成1000行，一个2000万行的文件要循环20000次，每次pandas都要重新初始化各种内部结构。推荐设成50000-200000行，在内存和速度之间取平衡。

坑2：dtype推断错误

pandas自动推断列类型时，可能把"00123"推断为整数123，丢失前导零。solution：读的时候指定dtype：

df=pd.read_csv(file,dtype={'工号':str,'手机号':str})

坑3：特殊字符导致解析失败

CSV中某行包含未转义的引号或换行符，会导致解析错位。用error_bad_lines=False跳过问题行（但可能丢数据），更好的是加quoting=csv.QUOTE_ALL规范导出。

坑4：内存溢出

即使分块处理，如果每个chunk都在accumulate（累积），总内存占用还是会飙升。对于不需要全量结果的场景（如只统计不保留细节），在chunk内部做完聚合就扔掉：

total=0forchunkinpd.read_csv(file,chunksize=100000):total+=chunk['金额'].sum()# chunk在这之后被GC回收print(f'总金额:{total}')

总结：大CSV处理的核心原则只有一条：不要试图一次性把所有数据装进内存。分块读、流式处理、及时释放，这三个做到就不会翻车。