Python中如何快速对大数据量DataFrame执行模糊匹配？

针对2000条字符串与500万条字符串的模糊匹配场景，原逐行全量比对的方式时间复杂度过高，以下是几个能显著提速的优化方向：

1. 替换为更高效的模糊匹配库 #

fuzzy-match性能偏慢，推荐改用RapidFuzz（底层基于C实现，速度比原生Python库快几十倍）。它支持批量处理和候选集过滤，能大幅减少计算时间。

示例代码： #

import time
from rapidfuzz import process, fuzz

start = time.time()

# 提前将大表的Key_ID转为列表，避免重复读取
siren_keys = Base_Siren_Data["Key_ID"].tolist()

# 提取每个字符串的最优匹配
def get_best_match(key):
    result = process.extractOne(key, siren_keys, scorer=fuzz.JaroWinkler, score_cutoff=0.85)
    return result[0] if result else 'nan'

data_1_8_Projets_tmp['MachedValue'] = data_1_8_Projets_tmp['Key_ID'].apply(get_best_match)

print(time.time() - start, ' seconds')

2. 预过滤候选集，减少比对次数 #

直接全量比对500万条数据效率极低，可通过字符特征预筛选缩小候选范围，再对候选集做模糊匹配：

实现思路： #

• 为500万条Key_ID构建n-gram（比如2-gram、3-gram）倒排索引，记录每个n-gram对应的字符串集合
• 对每个待匹配的字符串，提取它的n-gram，找到所有包含共同n-gram的字符串作为候选集
• 仅在候选集内做模糊匹配，避免全量遍历

示例代码： #

from collections import defaultdict
from rapidfuzz import process, fuzz

# 构建3-gram倒排索引
def build_ngram_index(strings, n=3):
    ngram_index = defaultdict(set)
    for s in strings:
        if len(s) < n:
            ngram_index[s].add(s)
            continue
        # 生成所有3-gram片段
        for i in range(len(s) - n + 1):
            ngram = s[i:i+n]
            ngram_index[ngram].add(s)
    return ngram_index

# 预构建索引
siren_keys = Base_Siren_Data["Key_ID"].tolist()
ngram_index = build_ngram_index(siren_keys)

def get_filtered_best_match(key):
    n = 3
    candidate_set = set()
    if len(key) < n:
        candidate_set.update(ngram_index.get(key, set()))
    else:
        for i in range(len(key) - n + 1):
            ngram = key[i:i+n]
            candidate_set.update(ngram_index.get(ngram, set()))
    # 候选集为空直接返回nan
    if not candidate_set:
        return 'nan'
    # 仅在候选集内做模糊匹配
    result = process.extractOne(key, list(candidate_set), scorer=fuzz.JaroWinkler, score_cutoff=0.85)
    return result[0] if result else 'nan'

data_1_8_Projets_tmp['MachedValue'] = data_1_8_Projets_tmp['Key_ID'].apply(get_filtered_best_match)

3. 并行化处理 #

由于2000条待匹配数据相互独立，可通过多进程/多线程并行处理，进一步缩短时间：

示例代码（用multiprocessing）： #

import multiprocessing as mp
from rapidfuzz import process, fuzz

siren_keys = Base_Siren_Data["Key_ID"].tolist()

def worker(key):
    result = process.extractOne(key, siren_keys, scorer=fuzz.JaroWinkler, score_cutoff=0.85)
    return result[0] if result else 'nan'

if __name__ == "__main__":
    start = time.time()
    # 用CPU核心数创建进程池
    with mp.Pool(mp.cpu_count()) as pool:
        matches = pool.map(worker, data_1_8_Projets_tmp['Key_ID'].tolist())
    data_1_8_Projets_tmp['MachedValue'] = matches
    print(time.time() - start, ' seconds')

4. 利用数据库全文索引（可选） #

如果500万条数据可以导入数据库（比如SQLite、PostgreSQL），可先通过数据库的全文索引做初步筛选，再把候选集拉回Python做模糊匹配，适合需要多次重复匹配的场景。

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