如何加速Python中大数据集下的Levenshtein距离嵌套计算循环？

核心优化方向 #

针对10000×50000=5亿次计算需求，从减少重复开销、并行计算、替换高效库三个维度入手，可将耗时压缩至1-2小时内：

1. 消除重复实例创建开销 #

原代码每次循环都初始化abd.DiscountedLevenshtein()实例，会产生大量无意义的对象创建损耗，将实例初始化移到循环外：

# 仅初始化一次实例
levenshtein = abd.DiscountedLevenshtein()

for a in range(10000):
    listA_element = listA[a]
    for b in range(50000):
        listB_element = listB[b]
        score = levenshtein.sim(listA_element, listB_element)

这一步至少能节省15%-25%的运行时间。

2. 多进程并行拆分任务 #

Python的GIL对CPU密集型任务限制极大，用多进程将大任务拆分为多个子任务并行处理：

• 把列表A拆分为与CPU核心数匹配的子列表，每个进程负责一个子列表与整个列表B的计算
• 示例代码如下：

from multiprocessing import Pool

levenshtein = abd.DiscountedLevenshtein()

def process_chunk(chunk):
    chunk_results = []
    for a_element in chunk:
        element_scores = []
        for b_element in listB:
            score = levenshtein.sim(a_element, b_element)
            element_scores.append(score)
        chunk_results.append(element_scores)
    return chunk_results

# 根据CPU核心数设置进程数（推荐为核心数的80%）
num_processes = 12
chunk_size = len(listA) // num_processes
# 拆分listA为多个子块
chunks = [listA[i:i+chunk_size] for i in range(0, len(listA), chunk_size)]

with Pool(num_processes) as pool:
    all_results = pool.map(process_chunk, chunks)

# 合并所有进程的结果
final_results = [item for sublist in all_results for item in sublist]

若使用16核CPU，并行后可将耗时压缩至原单线程的1/8~1/12。

3. 替换为C实现的高效库 #

如果abd.DiscountedLevenshtein性能一般，换成纯C实现的Levenshtein库，速度能提升10-100倍：

• 优先选用python-Levenshtein库（官方维护的C扩展），若需要折扣Levenshtein逻辑，可基于该库源码修改实现
• 普通Levenshtein距离计算示例：

import Levenshtein

def process_chunk(chunk):
    chunk_results = []
    for a_element in chunk:
        # 用列表推导式进一步提速
        element_scores = [Levenshtein.distance(a_element, b) for b in listB]
        chunk_results.append(element_scores)
    return chunk_results

4. 缓存重复元素计算结果 #

若列表B存在大量重复元素，先统计重复项，计算一次后缓存结果，避免重复计算：

from collections import defaultdict

# 缓存listB元素的索引列表
b_cache = defaultdict(list)
for idx, b_element in enumerate(listB):
    b_cache[b_element].append(idx)

levenshtein = abd.DiscountedLevenshtein()
# 预初始化结果矩阵
results = [[0]*len(listB) for _ in range(len(listA))]

for a_idx, a_element in enumerate(listA):
    for b_element, b_indices in b_cache.items():
        score = levenshtein.sim(a_element, b_element)
        # 填充所有重复元素的位置
        for b_idx in b_indices:
            results[a_idx][b_idx] = score

若listB重复率超过30%，这一步能大幅减少计算量。

预期效果 #

结合以上优化手段：

• 实例优化+多进程+高效库的组合，可将10小时的耗时压缩至1-2小时以内
• 若存在大量重复元素，缓存优化能进一步缩短运行时间

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