影刀RPA进阶教程_CSV大文件处理方案十万行数据的高效读写与内存管理

影刀RPA进阶教程：CSV大文件处理方案十万行数据的高效读写与内存管理

你做电商数据采集，每天跑出来一个CSV文件。

一开始几千行，Excel打开秒开。

上了几个月后，文件长到了十万行。Excel打开要等30秒，保存要一分钟。

再往后，文件超过一百万行——Excel直接告诉你"文件太大，无法完全加载"。

这时候你需要换一种思路来处理数据。

Excel的极限在哪里

Excel（xlsx格式）的单sheet上限是 1,048,576行 × 16,384列。

但实际使用中，超过10万行后操作就明显变慢，超过50万行基本上"动一下卡一下"。

而CSV（纯文本格式）没有这个限制——它就是一个文本文件，文件有多大取决于你的硬盘有多大。

关键认知：CSV文件超过10万行后，不要再用Excel打开处理，改用Python/pandas直接读写。

为什么用pandas处理大CSV

pandas处理CSV有三个优势：

一个100MB的CSV，pandas可能只用30MB内存就读完了。而Excel打开同一个文件可能消耗1GB+。

实战一：分块读取并处理

店群矩阵自动化突破运营极限！

假设你有一个200万行的订单CSV，需要筛选出状态为"已发货"的记录。

一次性读完内存可能不够，但分块读毫无压力：

import pandas as pd

# 分块大小：每次读5万行
chunk_size = 50000
output_file = '已发货订单.csv'
first_chunk = True

# 分块读取
for chunk in pd.read_csv('orders.csv', chunksize=chunk_size):
# 筛选当前块
filtered = chunk[chunk['状态'] == '已发货']

# 写入（第一个块写表头，后续追加）
if first_chunk:
filtered.to_csv(output_file, index=False, mode='w')
first_chunk = False
else:
filtered.to_csv(output_file, index=False, mode='a', header=False)

print(f'已处理 {len(chunk)} 行，筛选出 {len(filtered)} 行')

print('处理完成！')

chunksize 是核心参数。50,000是一个比较安全的数字（大多数机器的内存扛得住），可以根据你的机器内存调整。

实战二：只读取需要的列

你的CSV有50个列，但你只需要其中3个：

# 错误做法：全部加载（浪费内存）
df = pd.read_csv('orders.csv') # 500MB
df = df[['订单号', '金额', '状态']]

# 正确做法：只加载需要的列
df = pd.read_csv('orders.csv', usecols=['订单号', '金额', '状态'])
# 可能只需要30MB

usecols 参数在文件头部解析阶段就跳过了不需要的列，内存节省是数量级的。

实战三：优化数据类型减少内存

pandas默认给每列一个"宽松"的数据类型，但这很浪费内存。

# 先看看自动分配的类型有多浪费
df = pd.read_csv('orders.csv', nrows=10000)
print(df.dtypes)
print(f'内存占用: {df.memory_usage(deep=True).sum() / 1024 / 1024:.2f} MB')

# 手动指定类型
dtype_dict = {
'订单号': 'string', # 不用int64（可能溢出）
'金额': 'float32', # 不用float64（精度够用）
'数量': 'int16', # 不用int64（数量不会超过32767）
'状态': 'category', # 枚举值用category（大幅节省）
'创建时间': 'string', # 后面再转datetime
'备注': 'string'
}

df_optimized = pd.read_csv('orders.csv', dtype=dtype_dict)
print(f'优化后内存: {df_optimized.memory_usage(deep=True).sum() / 1024 / 1024:.2f} MB')

一个常见的实际案例：400MB的CSV，优化后降到60MB。

特别是 'status' 这种枚举列（只有几个取值），用 category 类型能节省90%+的内存。

实战四：CSV到Excel分批转换

有时候你需要把大CSV转成xlsx给别人——毕竟不是所有人都会操作CSV。

但百万行直接转xlsx不行（Excel上限104万行）。

方案：按指定行数分sheet

import pandas as pd

max_rows_per_sheet = 500000 # 每个sheet最多50万行

chunk_size = 50000
sheet_index = 0
total_rows = 0
buffer = []

for chunk in pd.read_csv('large.csv', chunksize=chunk_size):
buffer.append(chunk)
total_rows += len(chunk)

if total_rows >= max_rows_per_sheet:
# 合并buffer，写入一个sheet
sheet_data = pd.concat(buffer)
sheet_name = f'Sheet{sheet_index + 1}'

if sheet_index == 0:
sheet_data.to_excel('output.xlsx',
sheet_name=sheet_name, index=False)
else:
with pd.ExcelWriter('output.xlsx', mode='a',

engine='openpyxl') as writer:
sheet_data.to_excel(writer,
sheet_name=sheet_name, index=False)

# 重置
buffer = []
total_rows = 0
sheet_index += 1
print(f'已写入 {sheet_name}')

# 处理剩余数据
if buffer:
sheet_data = pd.concat(buffer)
with pd.ExcelWriter('output.xlsx', mode='a',
engine='openpyxl') as writer:
sheet_data.to_excel(writer,
sheet_name=f'Sheet{sheet_index + 1}',
index=False)
print(f'已写入最后一个sheet，共 {sheet_index + 1} 个sheet')

print('全部完成！')

实战五：CSV去重（内存友好版）

百万行数据去重，直接drop_duplicates()内存不够怎么办？

方案：分块+排序去重

import pandas as pd
import hashlib

def deduplicate_large_csv(input_file, output_file, key_columns, chunksize=50000):
seen = set()
first_chunk = True

for chunk in pd.read_csv(input_file, chunksize=chunksize):
# 生成唯一标识（基于去重列）
chunk['_hash'] = chunk[key_columns].astype(str).agg('|'.join, axis=1)
chunk['_hash'] = chunk['_hash'].apply(
lambda x: hashlib.md5(x.encode()).hexdigest()
)

# 过滤掉已见过的
mask = ~chunk['_hash'].isin(seen)
new_rows = chunk[mask]

# 更新已见集合
seen.update(chunk['_hash'].values)

# 写入（去掉辅助列）
new_rows = new_rows.drop(columns=['_hash'])
if first_chunk:
new_rows.to_csv(output_file, index=False, mode='w')
first_chunk = False
else:
new_rows.to_csv(output_file, index=False, mode='a', header=False)

print(f'本块 {len(chunk)} 行，去重后 {len(new_rows)} 行')

deduplicate_large_csv('input.csv', 'deduped.csv', ['订单号', '商品ID'])

这里用了hash而不是直接存整行数据到set里，内存占用小很多。

CSV和Excel的选型对照

选型口诀：数据中转用CSV，最终交付用Excel，超过50MB用CSV+pandas处理。

在影刀中集成

影刀流程调用Python处理大CSV：

1. HTTP请求获取数据 → 写入CSV
2. Python代码块：分块读取CSV → 清洗 → 输出
3. （可选）Python代码块：CSV分sheet转Excel
4. 发送Excel或CSV给业务方

第2步和第3步在同一个Python代码块里完成，减少中间文件。

temu店群自动化报活动案例

避坑指南

坑1：编码乱码

CSV用UTF-8保存但在Excel中打开中文乱码。

解决：用UTF-8 BOM格式保存

df.to_csv('output.csv', index=False, encoding='utf-8-sig')

utf-8-sig 会在文件头加BOM标记，Excel能正确识别。

坑2：CSV的逗号冲突

数据里如果包含逗号，pandas会自动加引号，没问题。但如果你的字段值里包含双引号，需要用两个双引号转义。

pandas自动处理了这些，但如果你手动拼接CSV字符串就要注意。

坑3：pandas自动推断类型错误

比如ID列 00123 被pandas读成数字 123：

df = pd.read_csv('data.csv', dtype={'ID': 'str'})

总结

处理大CSV的核心思路：不要一次性全加载。

分块读、筛选列、优化类型——三个技巧解决90%的大文件问题。

10万行以下用Excel直接看没问题，50万行以上必须切到pandas+分块模式。

CSV作为数据管道中的"中转格式"，轻量且无上限，值得你熟练掌握。

内容标签：#影刀RPA #CSV #pandas #大数据 #数据处理 作者：林焱 系列：影刀RPA进阶教程系列——从数据处理到工程化实践