使用Pandas对比大文件时的内存管理问题咨询

你当前的代码一次性将两个4-5GB的文件全量加载进DataFrame，后续的拼接、排序、去重操作又会额外占用数倍内存，内存需求远超8GB，所以调整heap size无法解决问题。以下是几个可行的优化思路：

1. 哈希集合（Set）逐行遍历——最省内存方案 #

核心思路是只将其中一个文件的记录唯一标识存入内存，再逐行扫描另一个文件筛选差异，完全避免全量加载：

示例代码 #

import datetime

def get_record_key(line):
    # 提取关键字段生成唯一标识（对应代码中三个对账字段）
    parts = line.strip().split(',')
    if len(parts) >= 3:
        return ','.join([parts[0], parts[1], parts[2]])
    return None

def find_unique_records(file_a_path, file_b_path, output_path):
    start_time = datetime.datetime.now()
    print(f"加载文件 {file_a_path} 记录标识", start_time)
    
    # 存储文件A的所有唯一标识
    record_keys = set()
    with open(file_a_path, 'r', encoding='utf-8') as f:
        for line in f:
            key = get_record_key(line)
            if key:
                record_keys.add(key)
    print(f"文件 {file_a_path} 加载完成，共 {len(record_keys)} 条记录", datetime.datetime.now())
    
    # 遍历文件B，筛选不在A中的记录并写入输出
    print(f"查找文件 {file_b_path} 独有记录", datetime.datetime.now())
    with open(file_b_path, 'r', encoding='utf-8') as f_in, open(output_path, 'a', encoding='utf-8') as f_out:
        for line in f_in:
            key = get_record_key(line)
            if key and key not in record_keys:
                f_out.write(line)
    print(f"文件 {file_b_path} 独有记录写入完成", datetime.datetime.now())
    print(f"总耗时: {datetime.datetime.now() - start_time}\n")

# 查找源文件独有的记录（源不在目标中）
find_unique_records(TargetFile, SourceFile, diffFile)
# 查找目标文件独有的记录（目标不在源中），追加到同一输出文件
find_unique_records(SourceFile, TargetFile, diffFile)

该方案内存仅需存储一个文件的记录标识（短字符串），4GB文件转成Set后仅需几百MB内存，完全避开DataFrame的额外内存开销。

2. Pandas分块读取——保留Pandas习惯的方案 #

通过read_csv的chunksize参数分批次加载数据，避免一次性占用大量内存：

示例代码 #

import pandas as pd
import datetime

chunk_size = 100000  # 每次加载10万条，可根据内存调整

# 第一步：存储目标文件的所有记录标识
target_keys = set()
print("分块加载目标文件", datetime.datetime.now())
for chunk in pd.read_csv(TargetFile, names=["CardholdNumber", "MemberNo", "CrdSponsorNo"],
                         dtype={"CardholdNumber": str, "MemberNo": str, "CrdSponsorNo": str},
                         chunksize=chunk_size):
    chunk['key'] = chunk['CardholdNumber'] + ',' + chunk['MemberNo'] + ',' + chunk['CrdSponsorNo']
    target_keys.update(chunk['key'].tolist())
print("目标文件加载完成", datetime.datetime.now())

# 第二步：筛选源文件独有的记录并写入
print("查找源文件独有记录", datetime.datetime.now())
with open(diffFile, 'a', encoding='utf-8') as f_out:
    for chunk in pd.read_csv(SourceFile, names=["CardholdNumber", "MemberNo", "CrdSponsorNo"],
                             dtype={"CardholdNumber": str, "MemberNo": str, "CrdSponsorNo": str},
                             chunksize=chunk_size):
        chunk['key'] = chunk['CardholdNumber'] + ',' + chunk['MemberNo'] + ',' + chunk['CrdSponsorNo']
        unique_chunk = chunk[~chunk['key'].isin(target_keys)]
        unique_chunk.to_csv(f_out, header=False, index=False)
print("源文件独有记录写入完成", datetime.datetime.now())

# 反向操作：查找目标文件独有的记录
source_keys = set()
print("分块加载源文件", datetime.datetime.now())
for chunk in pd.read_csv(SourceFile, names=["CardholdNumber", "MemberNo", "CrdSponsorNo"],
                         dtype={"CardholdNumber": str, "MemberNo": str, "CrdSponsorNo": str},
                         chunksize=chunk_size):
    chunk['key'] = chunk['CardholdNumber'] + ',' + chunk['MemberNo'] + ',' + chunk['CrdSponsorNo']
    source_keys.update(chunk['key'].tolist())
print("源文件加载完成", datetime.datetime.now())

print("查找目标文件独有记录", datetime.datetime.now())
with open(diffFile, 'a', encoding='utf-8') as f_out:
    for chunk in pd.read_csv(TargetFile, names=["CardholdNumber", "MemberNo", "CrdSponsorNo"],
                             dtype={"CardholdNumber": str, "MemberNo": str, "CrdSponsorNo": str},
                             chunksize=chunk_size):
        chunk['key'] = chunk['CardholdNumber'] + ',' + chunk['MemberNo'] + ',' + chunk['CrdSponsorNo']
        unique_chunk = chunk[~chunk['key'].isin(source_keys)]
        unique_chunk.to_csv(f_out, header=False, index=False)
print("目标文件独有记录写入完成", datetime.datetime.now())

3. Dask自动分块处理——适合复杂操作的方案 #

Dask兼容Pandas语法，自动对大文件分块并行处理，无需手动管理内存：

示例代码 #

import dask.dataframe as dd

# 读取两个文件
dd_target = dd.read_csv(TargetFile, names=["CardholdNumber", "MemberNo", "CrdSponsorNo"],
                        dtype={"CardholdNumber": str, "MemberNo": str, "CrdSponsorNo": str})
dd_source = dd.read_csv(SourceFile, names=["CardholdNumber", "MemberNo", "CrdSponsorNo"],
                        dtype={"CardholdNumber": str, "MemberNo": str, "CrdSponsorNo": str})

# 添加来源标识
dd_target['Source'] = 'rods10'
dd_source['Source'] = 'prod10'

# 合并后筛选不重复记录
dd_combined = dd.concat([dd_target, dd_source])
dd_unique = dd_combined.drop_duplicates(subset=["CardholdNumber", "MemberNo", "CrdSponsorNo"], keep=False)

# 写入输出文件（single_file=True确保生成单个文件）
dd_unique.to_csv(diffFile, single_file=True, index=False)

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