如何优化Python百万行CSV网络组件正则匹配替换的性能？

处理超100万行的CSV文件，需求是：将每行gs_upstream_aa_guid中包含的GUID，与gs_attached_assembly_guid列的GUID匹配，把对应行的Field1_num值替换到当前行的Field1中。现有Python代码功能正常，但写入速度仅约15kb/s，因内存限制无法使用Pandas，Dask也因环境权限问题无法部署，需优化现有代码性能。

现有代码的性能瓶颈分析 #

• 循环内重复定义函数：search_and_replace在每一行循环里重新定义，带来不必要的函数定义开销。
• 正则匹配的冗余开销：用正则匹配逗号分隔的GUID完全没必要，直接拆分字符串后查字典的效率远高于正则。
• DictWriter的字段映射开销：csv.DictWriter需要每次将字典映射到字段顺序，比直接使用csv.writer处理列表行更耗时。
• 字符串拼接的低效写法：循环append后join，不如用列表推导式一次性生成结果。

优化方案与代码实现 #

核心优化点 #

1. 将替换函数移出循环，避免重复定义开销
2. 移除正则匹配，改用字符串拆分+字典查询的方式，大幅降低CPU消耗
3. 改用csv.writer替代DictWriter，减少字段映射开销
4. 用列表推导式优化字符串生成逻辑
5. 可选：批量写入减少磁盘IO次数

优化后的完整代码 #

import csv

# 文件路径（用原始字符串避免转义）
input_file = r'L:\Dev_h\Device Heirarchy\fulljoin_device_flow2.csv'
output_file = r'L:\Dev_h\Device Heirarchy\output2.csv'

# 输出字段顺序
output_fields = [
    'gs_attached_assembly_guid', 'gs_upstream_aa_guid', 'Field1_num',
    'Dev_no', 'gs_guid', 'gs_display_feature_guid', 'field2',
    'gs_network_feature_name', 'gs_assembly_guid', 'gs_display_feature_name',
    'Field1', 'gs_network_feature_guid', 'OID_'
]

# 提前定义替换逻辑，避免循环内重复定义
def generate_field1(upstream_guids, guid_map):
    return ','.join([
        guid_map.get(guid.strip(), '')
        for guid in upstream_guids.split(',')
        if guid.strip()
    ])

def process_csv():
    # 第一步：构建GUID到Field1_num的映射字典（仅保留有效非空值）
    guid_map = {}
    with open(input_file, 'r', newline='', encoding='utf-8') as in_file:
        reader = csv.DictReader(in_file)
        for row in reader:
            assembly_guid = row['gs_attached_assembly_guid'].strip()
            field1_value = row['Field1_num'].strip()
            if assembly_guid and field1_value:
                guid_map[assembly_guid] = field1_value

    # 第二步：遍历处理每行数据并写入输出文件
    with open(input_file, 'r', newline='', encoding='utf-8') as in_file, \
         open(output_file, 'w', newline='', encoding='utf-8') as out_file:
        
        reader = csv.DictReader(in_file)
        writer = csv.writer(out_file)
        writer.writerow(output_fields)  # 写入表头

        # 可选：批量写入，减少IO次数（根据内存调整batch_size）
        batch_size = 1000
        batch = []
        
        for row in reader:
            upstream_value = row['gs_upstream_aa_guid']
            row['Field1'] = generate_field1(upstream_value, guid_map) if upstream_value.strip() else ''
            
            # 按输出字段顺序整理成列表
            output_row = [row[field] for field in output_fields]
            batch.append(output_row)
            
            # 达到批量大小则写入
            if len(batch) >= batch_size:
                writer.writerows(batch)
                batch = []
        
        # 写入剩余的行
        if batch:
            writer.writerows(batch)

if __name__ == '__main__':
    process_csv()
    print("处理完成")

额外优化建议 #

• 指定文件编码：明确指定encoding='utf-8'，避免系统默认编码的兼容性问题，同时提升读写效率。
• 过滤空值：构建映射字典时过滤空的GUID和Field1值，减少字典大小，提升查询速度。
• 调整批量大小：根据可用内存调整batch_size（比如2000或5000），平衡内存占用和IO效率。
• 路径优化：使用原始字符串r'path'替代转义的\\\\，代码更清晰且不易出错。

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