如何加速Python中CSV文件的写入？现有脚本优化求助

针对你的代码，有多个明确的优化方向，能显著提升运行速度：

1. 移除冗余操作 #

• 删掉重复创建的csv.writer实例（原代码连续写了两次writer = csv.writer(f)）
• 去掉多余的time.time()初始化（第一行的t0 = time.time()完全没用）

2. 重构低效的字符串处理 #

原代码中把元组转成字符串后反复用replace清理格式，这是极大的性能浪费：

• 直接用''.join(combination)生成无格式的字符组合字符串，避免后续的括号、引号替换
• 跳过"列表转字符串再多次替换"的弯路，直接生成目标格式的B3和C3字符串

3. 批量写入减少IO开销 #

writerow()逐行写入会频繁触发磁盘IO，改用writerows()一次性写入所有行（或分批次写入），能大幅降低IO耗时。

4. 减少重复计算 #

把内层循环中可以复用的计算提到外层，比如每个combination的字符串只拼接一次，不用在每个A的循环里重复处理。

优化后的完整代码 #

import time
import itertools
import hashlib
import csv

def get_num_str(char):
    lower_char = char.lower()
    val = ord(lower_char) - 96
    # 直接去掉负号（对应数字字符的计算结果）
    return str(val).replace('-', '')

start_time = time.time()
chars = ['0','1','2','3','4','5','6','7','8','9','a','b','c','d','e','f']
numbers_list = range(0,25)  # 不用转成list，range本身可迭代
header = ['A', 'B', 'C']

output_rows = []
for combo in itertools.combinations_with_replacement(chars, 10):
    combo_str = ''.join(combo)
    for a_val in numbers_list:
        # 生成B的原始内容
        b_raw = f"{a_val}:{combo_str}"
        # 直接构建B3
        b3 = ''.join([get_num_str(c) for c in b_raw])[:10]
        # 计算哈希并构建C3
        hash_result = hashlib.sha256(b_raw.encode('utf-8')).hexdigest()
        c3 = ''.join([get_num_str(c) for c in hash_result])[:10]
        output_rows.append([a_val, b3, c3])

# 批量写入文件
with open('test.csv', 'w', newline='') as f:
    writer = csv.writer(f)
    writer.writerow(header)  # 原代码未写入表头，若不需要可删除此行
    writer.writerows(output_rows)

elapsed = time.time() - start_time
print(f"耗时: {elapsed:.2f}秒")

额外优化建议 #

如果数据量极大（比如组合数过多导致内存放不下所有行），可以改成分批次写入：

# 替换批量写入的部分
batch_size = 10000
with open('test.csv', 'w', newline='') as f:
    writer = csv.writer(f)
    writer.writerow(header)
    current_batch = []
    for combo in itertools.combinations_with_replacement(chars, 10):
        combo_str = ''.join(combo)
        for a_val in numbers_list:
            # 生成行内容的逻辑和之前一样
            b_raw = f"{a_val}:{combo_str}"
            b3 = ''.join([get_num_str(c) for c in b_raw])[:10]
            hash_result = hashlib.sha256(b_raw.encode('utf-8')).hexdigest()
            c3 = ''.join([get_num_str(c) for c in hash_result])[:10]
            current_batch.append([a_val, b3, c3])
            # 达到批次大小就写入并清空
            if len(current_batch) >= batch_size:
                writer.writerows(current_batch)
                current_batch = []
    # 写入剩余的行
    if current_batch:
        writer.writerows(current_batch)

如果你的机器有多核CPU，还可以用multiprocessing并行计算行内容，不过需要注意控制进程数和IO的同步，适合CPU密集的场景。

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