批量CSV按ID映射合并：替代Pandas map/apply的高效方案

嘿，针对你要处理数千个CSV文件对的场景，用Pandas的map/apply确实可能在内存和效率上碰到瓶颈——毕竟每次把大文件全加载到DataFrame里，处理几千次的话内存压力会很大。这里有两个更高效的方案，既能避免生成中间的两个CSV文件，又能快速完成映射合并：

方案1：用Python内置csv模块逐行处理（内存友好，速度快） #

这个方案的核心是先把小映射文件（File2）的内容加载成一个ID为键，(Char_Name, Cosmic Awareness)为值的字典，然后不管是从非结构化数据直接生成最终文件，还是处理已有的大文件，都可以逐行处理、实时完成映射，完全不需要生成中间的两个CSV文件。

代码示例：直接从非结构化数据生成最终文件 #

如果你的脚本原本是先生成File1和File2，现在可以改成直接输出最终文件：

import csv

# 第一步：从非结构化数据提取映射数据，构建字典（替代生成File2）
# 这里模拟从非结构化数据获取File2内容，实际替换成你的提取逻辑
mapping_data = [
    {"ID": "123", "Char_Name": "Uatu", "Cosmic Awareness": "3.4"},
    {"ID": "121", "Char_Name": "Galan", "Cosmic Awareness": "4.5"},
    {"ID": "122", "Char_Name": "Norrin Radd", "Cosmic Awareness": "1.6"},
    {"ID": "124", "Char_Name": "Shalla-bal", "Cosmic Awareness": "0.3"},
    {"ID": "125", "Char_Name": "Nova", "Cosmic Awareness": "1.2"},
]
id_mapping = {item["ID"]: (item["Char_Name"], item["Cosmic Awareness"]) for item in mapping_data}

# 第二步：从非结构化数据提取大文件数据，逐行映射并写入最终文件（替代生成File1）
# 模拟从非结构化数据获取File1行数据，实际替换成你的提取逻辑
large_file_rows = [
    {"feats": "AA", "ID": "123", "A": "3343", "B": "234", "C": "2342", "E": "112"},
    {"feats": "BB", "ID": "121", "A": "3342", "B": "237", "C": "2642", "E": "213"},
    {"feats": "CC", "ID": "122", "A": "3341", "B": "232", "C": "2352", "E": "912"},
    {"feats": "DD", "ID": "123", "A": "3343", "B": "233", "C": "5342", "E": "12"},
    {"feats": "EE", "ID": "121", "A": "3345", "B": "235", "C": "2442", "E": "2112"},
]

# 写入最终文件
with open("final_file.csv", "w", newline="", encoding="utf-8") as f_out:
    fieldnames = ["feats", "ID", "A", "B", "C", "E", "Char_Name", "Cosmic Awareness"]
    writer = csv.DictWriter(f_out, fieldnames=fieldnames)
    writer.writeheader()
    
    for row in large_file_rows:
        # 根据ID获取映射值，无匹配时设默认空值，可按需调整
        char_name, cosmic_awareness = id_mapping.get(row["ID"], ("", ""))
        row["Char_Name"] = char_name
        row["Cosmic Awareness"] = cosmic_awareness
        writer.writerow(row)

代码示例：合并已生成的File1和File2 #

如果已经有了两个中间文件，用以下代码快速合并：

import csv

# 读取小映射文件构建字典
id_mapping = {}
with open("file2.csv", "r", encoding="utf-8") as f_in:
    reader = csv.DictReader(f_in)
    for row in reader:
        id_mapping[row["ID"]] = (row["Char_Name"], row["Cosmic Awareness"])

# 读取大文件，逐行映射并写入最终文件
with open("file1.csv", "r", encoding="utf-8") as f_in, open("final_file.csv", "w", newline="", encoding="utf-8") as f_out:
    reader = csv.DictReader(f_in)
    # 构造最终文件表头
    final_fieldnames = reader.fieldnames + ["Char_Name", "Cosmic Awareness"]
    writer = csv.DictWriter(f_out, fieldnames=final_fieldnames)
    writer.writeheader()
    
    for row in reader:
        char_name, cosmic_awareness = id_mapping.get(row["ID"], ("", ""))
        row["Char_Name"] = char_name
        row["Cosmic Awareness"] = cosmic_awareness
        writer.writerow(row)

这个方案的核心优势：

• 内存占用极低：大文件逐行读取处理，不会把上万行（甚至更大）的数据全加载到内存
• 速度快：字典查询是O(1)操作，比Pandas的map/apply在大规模循环下效率高很多
• 灵活可控：可以直接跳过中间文件生成步骤，从非结构化数据直接输出最终结果

方案2：用Dask处理超大规模文件（并行化，适合TB级数据） #

如果你的文件规模特别大（比如每个大文件是百万/千万行级别），或者需要并行处理数千个文件对，可以用Dask。它会自动分块处理数据、利用多核CPU并行运算，而且不需要把整个数据集加载到内存。

代码示例 #

import dask.dataframe as dd

# 读取两个文件
df_large = dd.read_csv("file1.csv")
df_mapping = dd.read_csv("file2.csv")

# 按ID左连接合并，只保留需要的映射列
df_final = df_large.merge(
    df_mapping[["ID", "Char_Name", "Cosmic Awareness"]],
    on="ID",
    how="left"  # 保留大文件所有行，无匹配ID会显示NaN
)

# 写入最终文件（single_file=True确保输出单个CSV）
df_final.to_csv("final_file.csv", single_file=True, index=False)

Dask的优势：

• 自动并行处理，大幅提升处理速度
• 分块加载数据，内存友好，适合超大规模数据集
• 语法和Pandas几乎一致，学习成本低

注意事项 #

• 针对ID不匹配的情况，方案1用get方法设置了默认空值，方案2用how="left"会得到NaN，你可以根据需求调整默认值
• 处理数千个文件时，可以写循环批量遍历文件对，比如用os.listdir获取文件列表，逐个用方案1的逻辑处理，效率会很高

内容的提问来源于stack exchange，提问作者Mr. Confused