无需额外库实现类pandas.merge的大DataFrame高效差异提取方法

针对内存不足无法用pandas.merge合并百万级DataFrame的场景，以下是无需额外依赖、低内存消耗的解决方案，可提取仅存在于单个DataFrame的行并标记来源，同时覆盖“内容相同但索引不同”的情况：

核心思路 #

利用行内容哈希值唯一标识行（忽略索引），通过集合运算快速找出仅存在于单个DataFrame的行，全程仅需处理共同字段，大幅降低内存占用。

具体步骤 #

1. 预处理：优化数据类型+确定共同字段 #

先统一两个DataFrame的共同字段，并压缩数据类型减少内存占用：

import pandas as pd
import hashlib

# 假设你的DataFrame是df1、df2
common_cols = list(set(df1.columns) & set(df2.columns))
common_cols.sort()  # 固定字段顺序，避免拼接结果不一致

# 压缩数据类型：字符串转分类，数值类型向下转换
for col in common_cols:
    # 字符串字段转category
    if df1[col].dtype == 'object':
        df1[col] = df1[col].astype('category')
        df2[col] = df2[col].astype('category')
    # 数值字段向下压缩
    elif pd.api.types.is_numeric_dtype(df1[col]):
        cast_type = 'integer' if df1[col].dtype.kind == 'i' else 'float'
        df1[col] = pd.to_numeric(df1[col], downcast=cast_type)
        df2[col] = pd.to_numeric(df2[col], downcast=cast_type)

2. 生成行哈希值 #

基于共同字段生成每行的唯一哈希值（忽略索引）：

def get_row_hash(row, cols):
    # 拼接字段并处理空值，确保空值在两个DF中处理逻辑一致
    combined = '|'.join(str(row[col] if pd.notna(row[col]) else 'NaN') for col in cols).encode('utf-8')
    return hashlib.md5(combined).hexdigest()

# 生成哈希列
df1['row_hash'] = df1.apply(lambda x: get_row_hash(x, common_cols), axis=1)
df2['row_hash'] = df2.apply(lambda x: get_row_hash(x, common_cols), axis=1)

3. 找出独有行并标记来源 #

通过集合差集筛选仅存在于单个DataFrame的行，添加来源标记：

# 计算仅存在于单个DF的哈希集合
df1_unique_hashes = set(df1['row_hash']) - set(df2['row_hash'])
df2_unique_hashes = set(df2['row_hash']) - set(df1['row_hash'])

# 筛选目标行
df1_only = df1[df1['row_hash'].isin(df1_unique_hashes)].copy()
df1_only['source'] = 'df1'

df2_only = df2[df2['row_hash'].isin(df2_unique_hashes)].copy()
df2_only['source'] = 'df2'

# 合并最终结果
result = pd.concat([df1_only, df2_only], ignore_index=True)
# 可选：删除哈希列
result.drop('row_hash', axis=1, inplace=True)

超大规模数据的分块处理 #

如果DataFrame大到无法一次性加载到内存，可分块处理（以从CSV读取为例）：

chunk_size = 100000  # 每块10万行，可根据内存调整

# 第一步：批量计算所有行的哈希值，存入集合
def collect_hashes(file_path, cols):
    hashes = set()
    for chunk in pd.read_csv(file_path, chunksize=chunk_size):
        # 预处理数据类型
        for col in cols:
            if chunk[col].dtype == 'object':
                chunk[col] = chunk[col].astype('category')
            elif pd.api.types.is_numeric_dtype(chunk[col]):
                cast_type = 'integer' if chunk[col].dtype.kind == 'i' else 'float'
                chunk[col] = pd.to_numeric(chunk[col], downcast=cast_type)
        # 生成哈希并加入集合
        chunk_hashes = chunk.apply(lambda x: get_row_hash(x, cols), axis=1)
        hashes.update(chunk_hashes)
    return hashes

df1_hashes = collect_hashes('df1.csv', common_cols)
df2_hashes = collect_hashes('df2.csv', common_cols)

# 第二步：分块筛选独有行
def filter_unique_rows(file_path, cols, unique_hashes, source_tag):
    result = pd.DataFrame()
    for chunk in pd.read_csv(file_path, chunksize=chunk_size):
        chunk['row_hash'] = chunk.apply(lambda x: get_row_hash(x, cols), axis=1)
        filtered = chunk[chunk['row_hash'].isin(unique_hashes)].copy()
        filtered['source'] = source_tag
        result = pd.concat([result, filtered], ignore_index=True)
    return result

df1_only = filter_unique_rows('df1.csv', common_cols, df1_hashes - df2_hashes, 'df1')
df2_only = filter_unique_rows('df2.csv', common_cols, df2_hashes - df1_hashes, 'df2')

final_result = pd.concat([df1_only, df2_only], ignore_index=True)
final_result.drop('row_hash', axis=1, inplace=True)

注意事项 #

• 哈希冲突概率极低，若需更高安全性，可将md5替换为sha256；
• 空值处理需统一，避免因空值表示差异导致哈希不同；
• 分块大小可根据电脑内存灵活调整，优先保证单块能顺利加载。

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