内存高效的同键值聚合方法：CSV去重合并同键对应值

针对你提到的CSV重复键去重并合并对应值的需求，内存高效的核心思路是逐行处理文件，不一次性加载整个文件到内存——这对超大CSV文件特别友好，不会因为内存不足导致崩溃。下面我用Python给出两种不同场景的实现，同时拆解关键要点：

通用场景方案（适用于无序CSV） #

如果你的CSV文件按键是无序的，我们需要用一个字典来缓存每个键对应的所有值，但只会保留必要的数据，不会加载整个文件：

import csv

def merge_duplicate_keys(input_path, output_path, delimiter=',', value_separator=' '):
    key_values = {}
    
    # 逐行读取，每次仅在内存中保留当前处理的一行
    with open(input_path, 'r', newline='', encoding='utf-8') as infile:
        reader = csv.reader(infile, delimiter=delimiter)
        for row in reader:
            if len(row) < 2:
                continue  # 跳过无效的短行
            key = row[0].strip()
            value = row[1].strip()
            
            if key in key_values:
                key_values[key].append(value)
            else:
                key_values[key] = [value]
    
    # 将合并后的结果写入输出文件
    with open(output_path, 'w', newline='', encoding='utf-8') as outfile:
        writer = csv.writer(outfile, delimiter=delimiter)
        for key, values in key_values.items():
            merged_value = value_separator.join(values)
            writer.writerow([key, merged_value])

# 替换成你的实际文件路径即可使用
merge_duplicate_keys('input.csv', 'output.csv')

为什么这个方案内存高效？ #

• 逐行迭代读取：通过csv.reader的迭代特性，每次只处理一行，处理完就释放该行的内存，不会把几十GB的大文件全部塞进内存。
• 最小化缓存数据：字典仅存储每个键对应的所有值的列表，相比存储所有原始行，内存占用大幅降低——如果键的数量远小于总行数，效果会特别明显。

极致内存高效方案（适用于有序CSV） #

如果你的CSV文件已经按键排序（或者可以先预处理排序），我们可以做到内存占用恒定，完全不需要缓存所有键，只保留当前正在处理的键的值列表：

import csv

def merge_duplicate_keys_sorted(input_path, output_path, delimiter=',', value_separator=' '):
    with open(input_path, 'r', newline='', encoding='utf-8') as infile, \
         open(output_path, 'w', newline='', encoding='utf-8') as outfile:
        reader = csv.reader(infile, delimiter=delimiter)
        writer = csv.writer(outfile, delimiter=delimiter)
        
        current_key = None
        current_values = []
        
        for row in reader:
            if len(row) < 2:
                continue
            key = row[0].strip()
            value = row[1].strip()
            
            if key == current_key:
                current_values.append(value)
            else:
                # 输出上一个键的合并结果（跳过首次初始化的空值）
                if current_key is not None:
                    merged_value = value_separator.join(current_values)
                    writer.writerow([current_key, merged_value])
                # 更新当前处理的键和值列表
                current_key = key
                current_values = [value]
        
        # 处理文件最后一个键的数据
        if current_key is not None:
            merged_value = value_separator.join(current_values)
            writer.writerow([current_key, merged_value])

# 仅适用于按键有序的CSV文件
merge_duplicate_keys_sorted('sorted_input.csv', 'output.csv')

额外优化：预处理排序 #

如果你的原始CSV是无序的，可以先用系统级的sort命令预处理（这个命令本身就是内存高效的，会自动处理大文件）：

# Linux/macOS下：按第一列（逗号分隔）排序，输出到新文件
sort -t, -k1,1 input.csv > sorted_input.csv

补充说明 #

• 如果你的CSV没有复杂格式（比如值里不含逗号、引号），也可以不用csv模块，直接用split(',')分割行，性能会稍高一点，但csv模块更健壮，能处理各种边缘情况。
• 编码可以根据你的实际文件调整，比如gbk或者其他编码。

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