如何用Python提取三个大型CSV文件中的相似行并生成新文件？

嗨，这个问题其实挺常见的，用Python就能轻松搞定！核心思路是先锁定三个文件里都存在的唯一标识（看你的示例，ID字段最适合当这个标识），然后再从每个文件里把这些标识对应的行捞出来，最后合并成新的CSV文件就行。下面给你两种可行的方案，按需选择：

方案一：用Python自带csv模块（无需额外安装库） #

这个方案适合文件规模不算特别大的情况，不用装额外依赖，代码也直观：

第一步：收集三个文件的ID，找出共同存在的ID #

import csv

# 替换成你的三个CSV文件实际路径
file_paths = ["file1.csv", "file2.csv", "file3.csv"]

# 存储每个文件的ID集合（集合求交集效率超高）
id_collections = []

for file_path in file_paths:
    current_ids = set()
    with open(file_path, 'r', encoding='utf-8') as f:
        # 用DictReader可以通过列名访问数据，更清晰
        reader = csv.DictReader(f)
        for row in reader:
            # 这里假设ID列的表头是"ID"，如果你的表头不一样记得修改！
            current_ids.add(row["ID"].strip())
    id_collections.append(current_ids)

# 计算三个ID集合的交集，也就是三个文件都有的ID
shared_ids = id_collections[0] & id_collections[1] & id_collections[2]

第二步：提取共同ID对应的行，写入新CSV #

# 先获取表头（默认用第一个文件的表头，要是三个文件表头有差异可以手动指定）
with open(file_paths[0], 'r', encoding='utf-8') as f:
    header = csv.DictReader(f).fieldnames

# 创建并写入新的CSV文件
with open("shared_rows.csv", 'w', encoding='utf-8', newline='') as output_file:
    writer = csv.DictWriter(output_file, fieldnames=header)
    writer.writeheader()  # 先写入表头

    # 遍历每个原文件，把属于共同ID的行写进去
    for file_path in file_paths:
        with open(file_path, 'r', encoding='utf-8') as f:
            reader = csv.DictReader(f)
            for row in reader:
                if row["ID"].strip() in shared_ids:
                    writer.writerow(row)

方案二：用Pandas库（适合大文件，代码更简洁） #

如果你的CSV文件行数特别多（比如几百万行），用Pandas处理会更高效，代码也更短，不过需要先安装Pandas：pip install pandas

import pandas as pd

# 读取三个CSV文件
df1 = pd.read_csv("file1.csv")
df2 = pd.read_csv("file2.csv")
df3 = pd.read_csv("file3.csv")

# 找出三个文件都存在的ID
shared_ids = df1["ID"].intersection(df2["ID"]).intersection(df3["ID"])

# 从每个DataFrame中筛选出共同ID的行，然后合并
shared_df = pd.concat([
    df1[df1["ID"].isin(shared_ids)],
    df2[df2["ID"].isin(shared_ids)],
    df3[df3["ID"].isin(shared_ids)]
])

# 如果同一个ID在单个文件里有多行，可以去重（按需选择）
shared_df = shared_df.drop_duplicates(subset=["ID"])

# 保存到新CSV文件，index=False是为了不写入自动生成的索引列
shared_df.to_csv("shared_rows.csv", index=False)

额外提醒 #

• 如果ID不是唯一判断标准，你需要用多列（比如ID+Date）来识别“相似行”，只需要把集合里的元素改成多字段拼接的字符串就行。比如方案一里的current_ids.add(f"{row['ID'].strip()}_{row['Date'].strip()}")，这样就能按多列匹配共同行了。
• 注意CSV文件的编码，如果读取时出现乱码，可以试试把encoding='utf-8'改成encoding='gbk'或者encoding='utf-8-sig'。

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