多文件拆分大型DataFrame重复行定位及内存溢出问题求解

你遇到的这个MemoryError很典型——把所有ID都塞进内存里的集合和字典，文件数量多、数据量大的时候肯定会撑爆内存。咱们换个分阶段+磁盘暂存的思路，不用把所有数据都放在内存里就能搞定，具体分三步来：

第一步：先检查单个文件内部的重复行 #

别着急跨文件找重复，先把每个文件自己内部的重复ID找出来，这部分处理起来简单，还能避免后续无效的跨文件检查。

import os
import pandas as pd

def check_single_file_duplicates(target_folder):
    for root, _, files in os.walk(target_folder):
        for file in files:
            # 只处理Excel文件，避免其他干扰文件
            if not file.lower().endswith('.xlsx'):
                continue
            full_path = os.path.join(root, file)
            # 只读取需要的ID列，减少内存加载量
            df = pd.read_excel(
                full_path,
                header=0,
                sheet_name="Results",
                usecols=['BvD ID number']  # 只加载目标列，省内存
            )
            # 找出当前文件中重复的ID（keep=False保留所有重复行）
            duplicate_ids = df[df.duplicated('BvD ID number', keep=False)]['BvD ID number'].unique()
            for dup_id in duplicate_ids:
                print(f"Duplicate row: ID '{dup_id}' appears multiple times in file '{file}'")

第二步：用哈希分桶把ID分散到临时文件里 #

这是解决内存问题的核心：把所有ID按哈希值分到多个临时文件（桶）里，同一个ID一定会被分到同一个桶中。这样每个桶里的ID数量就只有原来的1/N（N是桶的数量），内存压力直接降下来。

import hashlib
from collections import defaultdict

def get_bucket_index(id_str, total_buckets=100):
    # 用MD5哈希取模，保证相同ID分到同一个桶
    hash_result = hashlib.md5(id_str.encode('utf-8')).hexdigest()
    hash_int = int(hash_result, 16)
    return hash_int % total_buckets

def build_id_buckets(target_folder, total_buckets=100, temp_dir='temp_id_buckets'):
    # 创建临时文件夹存储桶文件
    os.makedirs(temp_dir, exist_ok=True)
    # 初始化所有桶文件的句柄
    bucket_handles = [
        open(os.path.join(temp_dir, f'bucket_{i}.txt'), 'w', encoding='utf-8')
        for i in range(total_buckets)
    ]
    
    for root, _, files in os.walk(target_folder):
        for file in files:
            if not file.lower().endswith('.xlsx'):
                continue
            full_path = os.path.join(root, file)
            df = pd.read_excel(
                full_path,
                header=0,
                sheet_name="Results",
                usecols=['BvD ID number']
            )
            # 遍历当前文件的每一行ID，写入对应桶
            for _, row in df.iterrows():
                id_str = str(row['BvD ID number'])
                bucket_idx = get_bucket_index(id_str, total_buckets)
                # 写入格式：ID\t文件名（方便后续解析）
                bucket_handles[bucket_idx].write(f"{id_str}\t{file}\n")
    
    # 关闭所有桶文件
    for handle in bucket_handles:
        handle.close()

第三步：在每个桶里找跨文件的重复ID #

现在每个桶里的ID数量已经很小了，直接把每个桶的内容加载到内存里，就能轻松找出跨文件的重复ID了。

def check_cross_file_duplicates(temp_dir='temp_id_buckets'):
    for bucket_filename in os.listdir(temp_dir):
        if not bucket_filename.startswith('bucket_'):
            continue
        bucket_path = os.path.join(temp_dir, bucket_filename)
        # 用字典存储每个ID对应的所有文件名
        id_to_files = defaultdict(list)
        
        with open(bucket_path, 'r', encoding='utf-8') as f:
            for line in f:
                line = line.strip()
                if not line:
                    continue
                id_str, file_name = line.split('\t', 1)
                id_to_files[id_str].append(file_name)
        
        # 检查每个ID对应的文件列表
        for id_str, file_list in id_to_files.items():
            # 去重文件名（避免同一个文件多次出现同一个ID的情况，已经在单文件检查过了）
            unique_files = list(set(file_list))
            if len(unique_files) >= 2:
                # 按照你需要的格式输出
                # 假设文件名是类似"file#10.xlsx"的格式，提取数字部分
                file_nums = [f.split('#')[-1].split('.')[0] for f in unique_files]
                # 如果只有两个重复文件，按示例格式输出
                if len(file_nums) ==2:
                    print(f"Duplicate row: ID_{id_str} is contained in both file #{file_nums[0]} and file #{file_nums[1]}")
                # 如果有多个文件重复，也可以扩展输出
                else:
                    file_str = ", ".join([f"file #{num}" for num in file_nums])
                    print(f"Duplicate row: ID_{id_str} is contained in {file_str}")

主流程调用 #

把上面的函数串起来，再加上临时文件清理的步骤：

if __name__ == '__main__':
    import sys
    import shutil
    
    if len(sys.argv) !=2:
        print("Usage: python duplicate_checker.py <path_to_excel_folder>")
        sys.exit(1)
    
    target_folder = sys.argv[1]
    
    print("=== Checking duplicates within single files ===")
    check_single_file_duplicates(target_folder)
    
    print("\n=== Building ID buckets (this may take a while) ===")
    build_id_buckets(target_folder)
    
    print("\n=== Checking duplicates across files ===")
    check_cross_file_duplicates()
    
    # 清理临时文件夹
    shutil.rmtree('temp_id_buckets')
    print("\nDone! Temporary files cleaned up.")

为什么这个方案能解决内存问题？ #

• 原来的方案把所有ID都存在内存里，数据量大时直接溢出；现在分桶后，每次只处理一个桶的内容，内存占用只有原来的1/100（默认桶数量是100），完全可控。
• 只读取需要的ID列，避免加载Excel中其他无关数据，进一步减少内存消耗。
• 哈希分桶保证同一个ID一定会出现在同一个桶里，不会漏掉任何跨文件重复的情况。

可选优化建议 #

• 调整total_buckets参数：如果内存还是紧张，就把桶的数量调大（比如200或500），每个桶的文件会更小。
• 分块读取大Excel：如果单个Excel文件特别大，可以用pd.read_excel(chunksize=1000)分块加载，每次只处理1000行。
• 用更快的哈希函数：如果觉得MD5慢，可以换成Python内置的hash()函数，但要注意字符串哈希的一致性（不同进程可能有差异，但单进程没问题）。

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