如何用Python和Pandas创建大于内存容量的CSV文件？

嘿，这个需求我在实际工作中经常碰到——要生成/处理比内存大得多的CSV，核心就是绝对不能把所有数据一次性加载到内存，而是用「分批次生成/处理 + 逐批追加写入」的思路。下面我分纯Python基础实现和Pandas优化方案来详细说：

一、纯Python基础实现（无需依赖Pandas） #

如果只是生成结构化数据或者处理简单文本行，纯Python的文件操作就足够高效，完全不用引入额外库。关键是用文件的a（追加）模式，每次只写入一小批数据。

代码示例：生成超大规模CSV #

import random
import numpy as np  # 可选，用numpy生成数据更快

total_rows = 100_000_000  # 要生成1亿行数据，远超普通内存容量
batch_size = 100_000       # 每批写入10万行，可根据内存调整

# 第一步：先写入CSV表头
with open('huge_data.csv', 'w', encoding='utf-8') as f:
    f.write('id,value_int,value_float\n')

# 第二步：循环生成批次并追加写入
for batch_idx in range(total_rows // batch_size):
    # 生成当前批次的起始ID
    start_id = batch_idx * batch_size
    # 高效生成批次数据（用numpy比列表推导快很多）
    ids = np.arange(start_id, start_id + batch_size)
    int_vals = np.random.randint(0, 101, size=batch_size)
    float_vals = np.random.uniform(0, 100, size=batch_size)
    
    # 把数据转成CSV行格式
    batch_lines = [f"{id},{int_val},{float_val:.4f}\n" 
                   for id, int_val, float_val in zip(ids, int_vals, float_vals)]
    
    # 追加写入文件
    with open('huge_data.csv', 'a', encoding='utf-8') as f:
        f.writelines(batch_lines)
    
    # 打印进度（可选）
    if (batch_idx + 1) % 10 == 0:
        print(f"已完成 {batch_idx+1} 批，累计写入 {(batch_idx+1)*batch_size} 行")

二、Pandas方案（适合结构化数据处理） #

如果需要对数据做复杂的清洗、转换，Pandas的分块处理+追加写入会更方便。核心是利用to_csv()的mode='a'参数，同时注意只有第一次写入时保留表头。

场景1：分批次生成DataFrame并追加 #

import pandas as pd
import numpy as np

total_rows = 100_000_000
batch_size = 100_000

# 处理第一个批次：写入表头
first_batch = pd.DataFrame({
    'id': np.arange(batch_size),
    'value_int': np.random.randint(0, 101, size=batch_size),
    'value_float': np.random.uniform(0, 100, size=batch_size)
})
# 第一次写入带表头，不要索引列
first_batch.to_csv('huge_data_pandas.csv', index=False, encoding='utf-8')

# 后续批次：追加写入，不写表头
for batch_idx in range(1, total_rows // batch_size):
    start_id = batch_idx * batch_size
    batch_df = pd.DataFrame({
        'id': np.arange(start_id, start_id + batch_size),
        'value_int': np.random.randint(0, 101, size=batch_size),
        'value_float': np.random.uniform(0, 100, size=batch_size)
    })
    # 关键参数：mode='a'（追加），header=False（不写表头）
    batch_df.to_csv('huge_data_pandas.csv', mode='a', index=False, header=False, encoding='utf-8')
    
    if (batch_idx + 1) % 10 == 0:
        print(f"已完成 {batch_idx+1} 批，累计写入 {(batch_idx+1)*batch_size} 行")

场景2：处理现有大CSV并输出新的超大规模CSV #

如果你需要读取一个已经存在的大CSV（比如从数据库导出的100G文件），处理后再写入新的CSV，同样可以用Pandas的分块读取+追加写入：

import pandas as pd

chunk_size = 50_000  # 每次读取5万行，根据内存调整
source_path = 'source_huge.csv'
output_path = 'processed_huge.csv'

# 分块读取源文件
for chunk_idx, chunk in enumerate(pd.read_csv(source_path, chunksize=chunk_size)):
    # 在这里做你的数据处理：比如新增列、过滤行、计算指标等
    chunk['value_double'] = chunk['value_int'] * 2
    chunk['value_filtered'] = chunk['value_float'].where(chunk['value_float'] > 50, 0)
    
    # 写入逻辑：第一次写表头，后续追加
    if chunk_idx == 0:
        chunk.to_csv(output_path, index=False, encoding='utf-8')
    else:
        chunk.to_csv(output_path, mode='a', index=False, header=False, encoding='utf-8')
    
    print(f"已处理第 {chunk_idx+1} 块，累计写入 {(chunk_idx+1)*chunk_size} 行")

三、关键注意事项 #

• 批次大小调整：根据你的可用内存灵活调整batch_size/chunk_size，内存小就设小一点（比如1万行），避免单批数据占满内存。
• 编码一致性：全程用统一的编码（比如utf-8），避免出现乱码问题。
• 避免冗余索引：Pandas写入时一定要加index=False，不然会把DataFrame的索引列写入CSV，造成多余数据。
• 异常处理：可以给写入逻辑加try-except块，避免中途出错导致数据中断，比如写入失败时记录日志并重试。
• 性能优化：生成数据时优先用numpy替代纯Python列表，速度更快且内存效率更高。

内容的提问来源于stack exchange，提问作者mihagazvoda