使用Pandas写入CSV遇内存不足/运行缓慢，如何实现分块写入？

你的问题根源在于一次性将所有生成的数据存入内存，再转成DataFrame写入CSV。你生成的总数据量非常大（超过8000万行），完全加载到内存必然导致内存耗尽或系统卡顿。分块写入是完美的解决方案，下面是修改后的代码：

import itertools, hashlib, pandas as pd, time

chars = ['0','1','2','3','4','5','6','7','8','9','a','b','c','d','e','f']
numbers_list = list(range(0,25))
chunksize = 1_000_000  # 每100万行写入一次
rows = []
output_file = 'data.csv'
header_written = False  # 标记表头是否已写入

t0 = time.time()

for combination in itertools.combinations_with_replacement(chars, 10):
    for A in numbers_list:
        pure = str(A) + ':' + str(combination) 
        B = pure.replace(")", "").replace("(", "").replace("'", "").replace(",", "").replace(" ", "") 
        C = hashlib.sha256(B.encode('utf-8')).hexdigest()
        rows.append([A, B, C])
        
        # 当积累到chunksize行时，写入CSV
        if len(rows) >= chunksize:
            df = pd.DataFrame(rows, columns=['A', 'B', 'C'])
            # 第一次写入写表头，之后追加不写表头
            df.to_csv(output_file, index=False, mode='a', header=not header_written)
            if not header_written:
                header_written = True
            # 清空rows列表释放内存
            rows = []

# 处理剩余不足chunksize的行
if rows:
    df = pd.DataFrame(rows, columns=['A', 'B', 'C'])
    df.to_csv(output_file, index=False, mode='a', header=not header_written)

tdelta = time.time() - t0
print(f"总耗时: {tdelta:.2f} 秒")

关键优化点： #

• 分块积累+即时写入：每生成100万行就写入磁盘，避免所有数据占用内存
• 内存释放：写入后立即清空rows列表，让垃圾回收机制释放内存
• 表头控制：只在第一次写入时写入表头，后续追加时跳过表头

额外优化建议（可选）： #

如果想进一步提升效率，可以直接使用Python内置的csv模块，避免Pandas的额外开销，内存占用更低：

import itertools, hashlib, csv, time

chars = ['0','1','2','3','4','5','6','7','8','9','a','b','c','d','e','f']
numbers_list = list(range(0,25))
chunksize = 1_000_000
output_file = 'data.csv'

t0 = time.time()

with open(output_file, 'w', newline='', encoding='utf-8') as f:
    writer = csv.writer(f)
    # 写入表头
    writer.writerow(['A', 'B', 'C'])
    row_count = 0
    
    for combination in itertools.combinations_with_replacement(chars, 10):
        for A in numbers_list:
            pure = str(A) + ':' + str(combination) 
            B = pure.replace(")", "").replace("(", "").replace("'", "").replace(",", "").replace(" ", "") 
            C = hashlib.sha256(B.encode('utf-8')).hexdigest()
            writer.writerow([A, B, C])
            row_count +=1
            
            # 每chunksize行刷新一次缓冲区，避免磁盘缓存积压
            if row_count % chunksize == 0:
                f.flush()

tdelta = time.time() - t0
print(f"总耗时: {tdelta:.2f} 秒")

内容的提问来源于stack exchange，提问作者Juan Soto Valdez