Excel 数据匹配实战：五种身份证比对算法性能对比

现在，我们开始设计算法。

算法一：暴力双重循环 (Brute Force Double Loop)

原理与步骤

这是最直观、最基础的方法，但也是效率最低的一种。

1. 加载数据：将两个 Excel 表分别读入 Pandas DataFrame。
2. 嵌套循环：外层遍历大表（140,000 次），内层遍历小表（5,000 次）。
3. 比对收集：对每一对 ID 进行比较，相等则标记。
4. 保存结果：提取标记的行保存。

Python 实现

def algorithm_1_brute_force(small_file, large_file, output_file):
    print("算法 1：暴力双重循环 - 开始执行")
    start_time = time.time()
    
    # 1. 加载数据
    small_df = pd.read_excel(small_file)
    large_df = pd.read_excel(large_file)
    
    # 确保 id_card 列是字符串类型，避免因数据类型导致的匹配失败
    small_df['id_card'] = small_df['id_card'].astype(str)
    large_df['id_card'] = large_df['id_card'].astype(str)
    
    # 获取小表的身份证号列表
    small_ids = small_df['id_card'].tolist()
    matched_rows = []
    large_ids = large_df['id_card'].tolist()
    
    # 2. 嵌套循环比对
    for i, large_id in enumerate(large_ids):
        for small_id in small_ids:
            if large_id == small_id:
                matched_rows.append(i)
                break
    
    # 3. 收集结果
    result_df = large_df.iloc[matched_rows]
    
    # 4. 保存结果
    result_df.to_excel(output_file, index=False)
    
    end_time = time.time()
    execution_time = end_time - start_time
    print(f"算法 1 完成。找到 {len(result_df)} 条匹配记录。耗时：{execution_time:.4f} 秒")
    return execution_time

优劣对比

• 优点：逻辑清晰，无需复杂数据结构知识。
• 缺点：时间复杂度极高 O(n*m)。总计 7 亿次比较。在普通 Python 环境中，预计需要 30 分钟到数小时。绝对不推荐用于生产环境。

算法二：利用 Pandas 的 isin() 方法

原理与步骤

这是对暴力算法的极大优化，利用了 Pandas 内置的高效向量化操作。

1. 加载数据：同算法一。
2. 创建查询集合：将小表 ID 转换为 Python 集合（Set）。集合基于哈希表，查找复杂度平均为 O(1)。
3. 向量化筛选：使用 isin() 方法返回布尔序列。
4. 布尔索引：直接筛选出匹配行。

Python 实现

def algorithm_2_pandas_isin(small_file, large_file, output_file):
    print("算法 2：Pandas isin() - 开始执行")
    start_time = time.time()
    
    # 1. 加载数据
    small_df = pd.read_excel(small_file)
    large_df = pd.read_excel(large_file)
    small_df['id_card'] = small_df['id_card'].astype(str)
    large_df['id_card'] = large_df['id_card'].astype(str)
    
    # 2. 创建查询集合
    target_set = set(small_df['id_card'])
    
    # 3. 向量化筛选
    mask = large_df['id_card'].isin(target_set)
    
    # 4. 使用布尔索引获取结果
    result_df = large_df[mask]
    
    # 5. 保存结果
    result_df.to_excel(output_file, index=False)
    
    end_time = time.time()
    execution_time = end_time - start_time
    print(f"算法 2 完成。找到 {len(result_df)} 条匹配记录。耗时：{execution_time:.4f} 秒")
    return execution_time

优劣对比

• 优点：代码简洁，速度极快。底层由 C 优化，近似 O(n)。
• 缺点：需将小表数据加载到内存构建集合（本场景下完全不是问题）。
• 预计运行时间：1 秒以内。这是处理此类问题的首选标准方法。

算法三：Pandas Merge（合并）

原理与步骤

利用数据库 INNER JOIN 思想，使用 Pandas 的合并功能。

1. 加载数据：同算法一。
2. 执行内连接：使用 pd.merge()，以'身份证号'为键进行内连接。
3. 保存结果：连接结果即为所需数据。

Python 实现

def algorithm_3_pandas_merge(small_file, large_file, output_file):
    print("算法 3：Pandas Merge - 开始执行")
    start_time = time.time()
    
    # 1. 加载数据
    small_df = pd.read_excel(small_file)
    large_df = pd.read_excel(large_file)
    small_df['id_card'] = small_df['id_card'].astype(str)
    large_df['id_card'] = large_df['id_card'].astype(str)
    
    # 2. 执行内连接
    # 注意：这里使用 small_df[['id_card']] 是为了避免其他同名列产生后缀
    result_df = pd.merge(large_df, small_df[['id_card']], on='id_card', how='inner')
    
    # 3. 保存结果
    result_df.to_excel(output_file, index=False)
    
    end_time = time.time()
    execution_time = end_time - start_time
    print(f"算法 3 完成。找到 {len(result_df)} 条匹配记录。耗时：{execution_time:.4f} 秒")
    return execution_time

优劣对比

• 优点：语义清晰，底层经过高度优化，性能与 isin() 在同一数量级。
• 缺点：比 isin() 略多协调列的开销，但在本场景下差异微乎其微。
• 预计运行时间：1 秒左右。与算法二同为首选标准方法。

算法四：使用数据库（SQLite）

原理与步骤

将数据加载到内存数据库，使用 SQL 语句让引擎完成高效查找。

1. 创建内存数据库：使用 sqlite3 模块。
2. 导入数据：将 DataFrame 写入数据库表。
3. 执行 SQL 查询：编写 SELECT ... WHERE ... IN (...) 或 JOIN 语句。
4. 获取结果：读回 Pandas DataFrame。

Python 实现

import sqlite3

def algorithm_4_sqlite(small_file, large_file, output_file):
    print("算法 4：SQLite 内存数据库 - 开始执行")
    start_time = time.time()
    
    # 1. 加载数据
    small_df = pd.read_excel(small_file)
    large_df = pd.read_excel(large_file)
    small_df['id_card'] = small_df['id_card'].astype(str)
    large_df['id_card'] = large_df['id_card'].astype(str)
    
    # 2. 创建内存数据库连接
    conn = sqlite3.connect(':memory:')
    
    # 3. 导入数据到数据库
    small_df.to_sql('small_table', conn, index=False)
    large_df.to_sql('large_table', conn, index=False)
    
    # 4. 编写并执行 SQL 查询
    query = """ SELECT large_table.* FROM large_table WHERE large_table.id_card IN (SELECT id_card FROM small_table) """
    result_df = pd.read_sql_query(query, conn)
    
    # 5. 关闭连接
    conn.close()
    
    # 6. 保存结果
    result_df.to_excel(output_file, index=False)
    
    end_time = time.time()
    execution_time = end_time - start_time
    print(f"算法 4 完成。找到 {len(result_df)} 条匹配记录。耗时：{execution_time:.4f} 秒")
    return execution_time

优劣对比

• 优点：数据库引擎专为查询设计，适合海量数据或复杂逻辑。
• 缺点：多了数据导入导出的 I/O 开销，对于小规模数据略慢于纯 Pandas。
• 预计运行时间：1-3 秒。数据量极大时优势显现。

算法五：分块处理 (Chunking)

原理与步骤

此算法并非为了提速，而是为了解决内存不足的问题。当大文件无法一次性读入内存时使用。

1. 加载小数据：全部读入内存并建集合 S。
2. 分块读取大数据：使用 read_excel(chunksize=...) 分批读取。
3. 逐块处理：对每个块使用 isin(S) 筛选。
4. 累积结果：追加结果或逐块写入文件。

Python 实现

def algorithm_5_chunking(small_file, large_file, output_file, chunksize=10000):
    print("算法 5：分块处理 - 开始执行")
    start_time = time.time()
    
    # 1. 加载小数据并创建集合
    small_df = pd.read_excel(small_file)
    small_df['id_card'] = small_df['id_card'].astype(str)
    target_set = set(small_df['id_card'])
    
    chunks_result_list = []
    
    # 2. 分块读取大数据
    chunk_reader = pd.read_excel(large_file, chunksize=chunksize)
    for chunk in chunk_reader:
        chunk['id_card'] = chunk['id_card'].astype(str)
        
        # 3. 处理当前块
        mask = chunk['id_card'].isin(target_set)
        filtered_chunk = chunk[mask]
        chunks_result_list.append(filtered_chunk)
        print(f"已处理一个数据块，该块找到 {len(filtered_chunk)} 条匹配记录。")
    
    # 4. 合并结果并保存
    if chunks_result_list:
        final_result_df = pd.concat(chunks_result_list, ignore_index=True)
    else:
        final_result_df = pd.DataFrame()
    
    final_result_df.to_excel(output_file, index=False)
    
    end_time = time.time()
    execution_time = end_time - start_time
    print(f"算法 5 完成。找到 {len(final_result_df)} 条匹配记录。耗时：{execution_time:.4f} 秒")
    return execution_time

优劣对比

• 优点：内存友好，可处理大于内存的文件。
• 缺点：多次 I/O 导致速度稍慢，实现较复杂。
• 预计运行时间：2-5 秒。仅在内存不足时使用。

总结与最终对比

我们将五种算法的优缺点和适用场景总结如下：

最终结论与建议：

1. 对于你的具体问题（5000 条 vs 140000 条）：
   • 毫不犹豫地选择算法二（isin()）或算法三（merge()）。它们是专门为这种表格数据操作设计的，代码简洁、效率最高。两者性能差异可以忽略不计，按个人编码喜好选择即可。
2. 如果你的小表也变得非常大（例如上百万条）：
   • 算法二和三仍然有效，但需确保机器有足够 RAM。
3. 如果你的大表巨大（例如几十 GB，无法读入内存）：
   • 选择算法五（分块处理），它是解决此类问题的标准范式。
4. 如果你的匹配逻辑变得非常复杂（不仅仅是相等匹配）：
   • 考虑使用算法四（SQLite），利用 SQL 强大的表达能力来编写复杂查询。
5. 算法一（暴力循环）：永远只存在于教学示例中，用于提醒大家时间复杂度的重要性。

代码执行：你可以创建一个主函数来运行和比较这些算法（除了算法一）。

if __name__ == '__main__':
    files = ('small.xlsx', 'large.xlsx')
    times = {}
    times['alg_2'] = algorithm_2_pandas_isin(*files, 'result_2.xlsx')
    times['alg_3'] = algorithm_3_pandas_merge(*files, 'result_3.xlsx')
    times['alg_4'] = algorithm_4_sqlite(*files, 'result_4.xlsx')
    times['alg_5'] = algorithm_5_chunking(*files, 'result_5.xlsx', chunksize=50000)
    
    print("\n=== 所有算法耗时对比 ===")
    for alg, t in times.items():
        print(f"{alg}: {t:.4f} 秒")

在实际运行中，你会看到算法二和三以绝对优势胜出。