优化基于fuzzywuzzy的两个DataFrame模糊匹配性能

问题背景 #

现有两个Excel数据：

• df_source：6.6万条已入库书籍数据，含link/name/author字段
• df_sort：3.6万条待入库书籍数据，含author/name字段

需通过模糊匹配识别重复书籍（存在作者/书名字符串差异，如"Stephen Edwin King it" vs "S. King it"），原代码采用双重循环调用fuzzywuzzy.fuzz.ratio，因O(nm)的时间复杂度（6.6万3.6万=237.6亿次匹配），运行效率极低，仅处理少量数据就耗时极长。

优化方案 #

1. 替换底层匹配引擎，提升单匹配速度 #

fuzzywuzzy默认用纯Python实现编辑距离计算，速度极慢。安装python-Levenshtein后，fuzzywuzzy会自动切换为C语言实现的Levenshtein算法，单匹配速度可提升10-100倍。

pip install python-Levenshtein

无需修改原有匹配逻辑，直接运行原代码即可获得显著提速。

2. 用更高效的模糊匹配库替代fuzzywuzzy #

rapidfuzz是fuzzywuzzy的高性能替代库，API兼容且速度更快，还支持向量化操作，避免循环开销：

pip install rapidfuzz pandas

3. 预过滤减少匹配次数（核心优化） #

直接全量匹配成本太高，先通过规则预过滤候选集，只对可能匹配的数据做模糊计算：

• 提取作者姓氏（如从"Stephen Edwin King"取"King"，从"S. King"取"King"）
• 提取书名核心关键词（如取书名前3个单词或去除标点后的核心词）
• 先通过姓氏+书名关键词做近似过滤，缩小需要模糊匹配的范围

示例代码（预过滤+rapidfuzz）：

import pandas as pd
from rapidfuzz import fuzz
from tqdm import tqdm

# 读取数据
df_sort = pd.read_excel('data/sort.xlsx', sheet_name='Sheet 1', names=["author", "name"], dtype="string")
df_source = pd.read_excel('data/list.xlsx', sheet_name='Sheet 1', names=["link", "name", "author"], dtype="string")

# 预处理：提取作者姓氏和书名核心词
def get_last_name(author_str):
    if pd.isna(author_str):
        return ""
    parts = author_str.strip().split()
    return parts[-1] if parts else ""

def get_title_keywords(title_str):
    if pd.isna(title_str):
        return ""
    cleaned = ''.join([c for c in title_str if c.isalnum() or c.isspace()]).strip()
    return ' '.join(cleaned.split()[:3])

df_source['last_name'] = df_source['author'].apply(get_last_name)
df_source['title_key'] = df_source['name'].apply(get_title_keywords)
df_sort['last_name'] = df_sort['author'].apply(get_last_name)
df_sort['title_key'] = df_sort['name'].apply(get_title_keywords)

# 存储结果的列表（避免逐行修改DataFrame）
duplicate_list = []
result_list = []

# 遍历待入库数据
for idx_sort, row_sort in tqdm(df_sort.iterrows(), total=len(df_sort)):
    sort_auth = row_sort['author']
    sort_name = row_sort['name']
    sort_ln = row_sort['last_name']
    sort_tk = row_sort['title_key']
    
    # 预过滤：只保留作者姓氏匹配、且书名关键词有交集的候选
    candidates = df_source[
        (df_source['last_name'] == sort_ln) & 
        (df_source['title_key'].str.contains('|'.join(sort_tk.split()), na=False))
    ]
    
    if len(candidates) == 0:
        result_list.append({"author": sort_auth, "name": sort_name})
        continue
    
    # 对候选集做模糊匹配
    max_ratio = 0
    best_match = None
    for idx_src, row_src in candidates.iterrows():
        src_str = f"{row_src['author']} {row_src['name']}"
        sort_str = f"{sort_auth} {sort_name}"
        ratio = fuzz.ratio(src_str, sort_str)
        if ratio > max_ratio:
            max_ratio = ratio
            best_match = row_src
    
    if max_ratio > 70:
        duplicate_list.append({
            "ratio": max_ratio,
            "author source": best_match['author'],
            "name source": best_match['name'],
            "author sort": sort_auth,
            "name sort": sort_name
        })
    else:
        result_list.append({"author": sort_auth, "name": sort_name})

# 转换为DataFrame并保存
duplicate = pd.DataFrame(duplicate_list)
result = pd.DataFrame(result_list)

duplicate.to_csv('out/duplicate.csv', index=False)
result.to_csv('out/result.csv', index=False)

4. 优化DataFrame操作 #

原代码中用df.loc[len(df.index)]逐行添加数据，每次都会创建新的DataFrame对象，效率极低。改为先将数据存入列表，最后一次性转换为DataFrame，可大幅提升写入速度。

5. 并行处理（可选） #

如果预过滤后仍有大量候选，可使用swifter库实现并行计算，进一步利用多核CPU资源：

pip install swifter

# 对每个待入库行并行计算最佳匹配
def find_best_match(row_sort):
    sort_auth = row_sort['author']
    sort_name = row_sort['name']
    sort_ln = row_sort['last_name']
    sort_tk = row_sort['title_key']
    
    candidates = df_source[
        (df_source['last_name'] == sort_ln) & 
        (df_source['title_key'].str.contains('|'.join(sort_tk.split()), na=False))
    ]
    
    if len(candidates) == 0:
        return {"match_type": "new", "author": sort_auth, "name": sort_name}
    
    max_ratio = 0
    best_match = None
    for idx_src, row_src in candidates.iterrows():
        src_str = f"{row_src['author']} {row_src['name']}"
        sort_str = f"{sort_auth} {sort_name}"
        ratio = fuzz.ratio(src_str, sort_str)
        if ratio > max_ratio:
            max_ratio = ratio
            best_match = row_src
    
    if max_ratio > 70:
        return {
            "match_type": "duplicate",
            "ratio": max_ratio,
            "author source": best_match['author'],
            "name source": best_match['name'],
            "author sort": sort_auth,
            "name sort": sort_name
        }
    else:
        return {"match_type": "new", "author": sort_auth, "name": sort_name}

# 并行处理
results = df_sort.swifter.apply(find_best_match, axis=1)

# 拆分结果
duplicate_list = [res for res in results if res['match_type'] == 'duplicate']
result_list = [res for res in results if res['match_type'] == 'new']

# 后续保存逻辑同上

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