如何加速十万级姓名列表的模糊匹配？

针对10万级姓名记录的模糊匹配效率问题，先拆解现有代码的核心瓶颈：

• 时间复杂度为O(N*M)：外层遍历待筛查姓名，内层遍历所有参考姓名计算相似度，10万×10万的计算量直接导致效率极低
• 重复计算排序姓名：每次调用相似度函数都要重新拆分、排序字符串，无缓存机制
• Pandas apply的额外开销：本质是循环，远慢于向量化操作

以下是可落地的优化思路：

1. 预处理排序姓名，消除重复计算 #

提前为两个数据集生成排序后的姓名列，避免每次计算都重复执行拆分、排序逻辑：

import pandas as pd

# 读取数据
df_name_reference = pd.read_csv('path_to_file')
df_name_to_screen = pd.read_csv('path_to_file')

# 预生成排序姓名列
def sort_name(name):
    if pd.isna(name) or not isinstance(name, str):
        return ""
    return ' '.join(sorted(name.split())).strip()

df_name_reference['sorted_fullname'] = df_name_reference['fullname'].apply(sort_name)
df_name_to_screen['sorted_fullname'] = df_name_to_screen['fullname'].apply(sort_name)

2. 替换为RapidFuzz库，提升单计算速度 #

RapidFuzz是C++实现的模糊匹配库，速度比原生Levenshtein快数倍，支持批量计算：

from rapidfuzz import fuzz

# 批量计算单个待筛查姓名与所有参考姓名的最大相似度
def get_batch_scores(ref_name, ref_sorted, ref_names, ref_sorted_names):
    # 批量计算原姓名相似度
    scores_raw = fuzz.ratio(ref_name, ref_names, score_cutoff=0.0)
    # 批量计算排序后姓名相似度
    scores_sorted = fuzz.ratio(ref_sorted, ref_sorted_names, score_cutoff=0.0)
    # 取每个姓名的最高分数
    return [max(a, b) for a, b in zip(scores_raw, scores_sorted)]

# 转换为列表，减少Pandas索引开销
ref_names_list = df_name_reference['fullname'].tolist()
ref_sorted_list = df_name_reference['sorted_fullname'].tolist()

screening_results = []
for _, row in df_name_to_screen.iterrows():
    scores = get_batch_scores(row['fullname'], row['sorted_fullname'], ref_names_list, ref_sorted_list)
    screening_results.append(pd.DataFrame({
        'screened_name': row['fullname'],
        'scores': scores,
        'reference_name': ref_names_list
    }))

3. 预过滤减少计算量 #

通过姓名长度、首字母等特征快速过滤不可能匹配的样本，大幅减少计算次数：

def pre_filter(ref_name, ref_names, ref_sorted_names, length_tol=3):
    ref_len = len(ref_name)
    # 仅保留长度差在阈值内的姓名
    filtered_indices = [
        i for i, name in enumerate(ref_names)
        if abs(len(name) - ref_len) <= length_tol
    ]
    return (
        filtered_indices,
        [ref_names[i] for i in filtered_indices],
        [ref_sorted_names[i] for i in filtered_indices]
    )

# 结合预过滤的批量计算
screening_results = []
for _, row in df_name_to_screen.iterrows():
    ref_name = row['fullname']
    ref_sorted = row['sorted_fullname']
    # 预过滤
    filtered_indices, filtered_names, filtered_sorted = pre_filter(ref_name, ref_names_list, ref_sorted_list)
    # 仅计算过滤后的样本
    scores = get_batch_scores(ref_name, ref_sorted, filtered_names, filtered_sorted)
    # 补全未过滤样本的分数为0
    full_scores = [0.0] * len(ref_names_list)
    for i, idx_ref in enumerate(filtered_indices):
        full_scores[idx_ref] = scores[i]
    screening_results.append(pd.DataFrame({
        'screened_name': ref_name,
        'scores': full_scores,
        'reference_name': ref_names_list
    }))

4. 用近似最近邻(ANN)将复杂度降至O(N*logM) #

对于10万级数据，O(N*M)的复杂度完全不可行，推荐用FAISS构建向量索引，快速查找相似姓名：

import faiss
from sklearn.feature_extraction.text import TfidfVectorizer

# 用字符n-gram构建TF-IDF向量（适配姓名匹配场景）
vectorizer = TfidfVectorizer(analyzer='char', ngram_range=(2,3))
ref_vectors = vectorizer.fit_transform(df_name_reference['fullname']).toarray()

# 构建FAISS索引
index = faiss.IndexFlatL2(ref_vectors.shape[1])
index.add(ref_vectors)

# 查找Top-K相似姓名后再计算精确相似度
k = 100  # 取最相似的100个样本，可按需调整
screening_results = []
for _, row in df_name_to_screen.iterrows():
    ref_name = row['fullname']
    # 转换待筛查姓名为向量
    query_vec = vectorizer.transform([ref_name]).toarray()
    # 快速查找Top-K相似
    distances, indices = index.search(query_vec, k)
    # 对Top-K样本计算精确相似度
    top_k_refs = df_name_reference.iloc[indices[0]]
    scores = get_batch_scores(
        ref_name, row['sorted_fullname'],
        top_k_refs['fullname'].tolist(),
        top_k_refs['sorted_fullname'].tolist()
    )
    screening_results.append(pd.DataFrame({
        'screened_name': ref_name,
        'reference_name': top_k_refs['fullname'].tolist(),
        'scores': scores
    }))

5. 多进程并行处理 #

利用CPU多核并行处理独立的姓名匹配任务：

from multiprocessing import Pool

def process_single_row(row):
    ref_name = row['fullname']
    ref_sorted = row['sorted_fullname']
    filtered_indices, filtered_names, filtered_sorted = pre_filter(ref_name, ref_names_list, ref_sorted_list)
    scores = get_batch_scores(ref_name, ref_sorted, filtered_names, filtered_sorted)
    full_scores = [0.0] * len(ref_names_list)
    for i, idx_ref in enumerate(filtered_indices):
        full_scores[idx_ref] = scores[i]
    return pd.DataFrame({
        'screened_name': ref_name,
        'scores': full_scores,
        'reference_name': ref_names_list
    })

# 转换为列表并行处理
rows_list = [row for _, row in df_name_to_screen.iterrows()]
with Pool(processes=4) as pool:  # 进程数根据CPU核心数调整
    screening_results = pool.map(process_single_row, rows_list)

额外建议 #

• 统一姓名格式：提前转换为小写、去除特殊字符（.、-）、标准化中间名缩写，减少无效差异
• 设置相似度阈值：仅保留分数高于阈值（如0.8）的结果，减少数据量和计算量

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