影刀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有三个优势：

1. 分块读取：不用一次性把整个文件加载到内存，每次读5万行处理完再读下5万行
2. 列筛选：可以只读取需要的列，不用加载全部
3. 数据类型优化：可以指定每列的数据类型，节省内存

一个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', 
            ![在这里插入图片描述](https://i-blog.csdnimg.cn/direct/9677436cc56d4221988289f4e6889fba.png#pic_center)

                               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 (xlsx)
文件大小上限	无限制	实际50MB以内流畅
行数上限	无限制	约104万行
多Sheet	不支持	支持
格式（颜色/合并）	不支持	支持
打开速度（大文件）	快	慢
适合用于	数据存储、中转、API输出	最终报表、人工查看
编码问题	可能有乱码	无

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

---

在影刀中集成

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

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

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

---

避坑指南

坑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进阶教程系列——从数据处理到工程化实践