Python中文本相似度两两比较的并行化实现方案咨询

问题背景 #

现有一个包含10K行、2列的DataFrame，第一列为客户消息，cnt列为消息的出现频次。需对任意两行文本做相似度两两比较（总计N²次计算），现需实现该过程的并行化来提升效率。

样本数据 #

message cnt
0     ABC    1
1     abc    2
2     cba    1
3    abcd    3
4    dcsa    5
5    adcd    6
6    abcd    2
7     cba    3

预期输出 #

message  cnt         similarity
0     ABC    1  [1, 0, 0, 0, 0, 0, 0, 0]
1     abc    2  [0, 1, 0, 1, 0, 0, 1, 0]
2     cba    1  [0, 0, 1, 0, 0, 0, 0, 1]
3    abcd    3  [0, 1, 0, 1, 0, 0, 1, 0]
4    dcsa    5  [0, 0, 0, 0, 1, 0, 0, 0]
5    adcd    6  [0, 0, 0, 0, 0, 1, 0, 0]
6    abcd    2  [0, 1, 0, 1, 0, 0, 1, 0]
7     cba    3  [0, 0, 1, 0, 0, 0, 0, 1]

串行可复现代码 #

import pandas as pd
from difflib import SequenceMatcher

df = pd.DataFrame({'message': ['ABC','abc','cba','abcd','dcsa','adcd','abcd','cba'], 
                  'cnt': [1, 2, 1, 3, 5, 6, 2, 3]})

def similarity_score(s1, s2):
    coef = SequenceMatcher(None, s1, s2).ratio()
    return 1 if coef >= 0.80 else 0

def similarity(x, df_messages):
    return [similarity_score(x, msg) for msg in df_messages]

df['similarity'] = df['message'].apply(lambda x: similarity(x, df['message'])).astype(str)

并行化实现方案 #

方案1：使用multiprocessing模块 #

利用Python标准库实现进程级并行，适配CPU密集型任务：

import pandas as pd
from difflib import SequenceMatcher
from multiprocessing import Pool, cpu_count

df = pd.DataFrame({'message': ['ABC','abc','cba','abcd','dcsa','adcd','abcd','cba'], 
                  'cnt': [1, 2, 1, 3, 5, 6, 2, 3]})

def similarity_score(s1, s2):
    coef = SequenceMatcher(None, s1, s2).ratio()
    return 1 if coef >= 0.80 else 0

def compute_similarity_for_row(row_msg, all_messages):
    return [similarity_score(row_msg, msg) for msg in all_messages]

if __name__ == '__main__':
    all_messages = df['message'].tolist()
    # 基于CPU核心数创建进程池
    with Pool(cpu_count()) as pool:
        similarity_results = pool.starmap(compute_similarity_for_row, [(msg, all_messages) for msg in all_messages])
    df['similarity'] = [str(res) for res in similarity_results]
    print(df)

方案2：使用joblib库 #

joblib封装了多进程逻辑，代码更简洁易读：

import pandas as pd
from difflib import SequenceMatcher
from joblib import Parallel, delayed

df = pd.DataFrame({'message': ['ABC','abc','cba','abcd','dcsa','adcd','abcd','cba'], 
                  'cnt': [1, 2, 1, 3, 5, 6, 2, 3]})

def similarity_score(s1, s2):
    coef = SequenceMatcher(None, s1, s2).ratio()
    return 1 if coef >= 0.80 else 0

def compute_similarity(row_msg, all_messages):
    return [similarity_score(row_msg, msg) for msg in all_messages]

all_messages = df['message'].tolist()
# n_jobs=-1表示调用全部可用CPU核心
similarity_results = Parallel(n_jobs=-1)(delayed(compute_similarity)(msg, all_messages) for msg in all_messages)
df['similarity'] = [str(res) for res in similarity_results]
print(df)

方案3：使用Dask处理超大数据量 #

如果数据量远超内存，可借助Dask实现并行计算+内存优化：

import dask.dataframe as dd
import pandas as pd
from difflib import SequenceMatcher

df = pd.DataFrame({'message': ['ABC','abc','cba','abcd','dcsa','adcd','abcd','cba'], 
                  'cnt': [1, 2, 1, 3, 5, 6, 2, 3]})
# 转换为Dask DataFrame，设置分区数
ddf = dd.from_pandas(df, npartitions=4)

def similarity_score(s1, s2):
    coef = SequenceMatcher(None, s1, s2).ratio()
    return 1 if coef >= 0.80 else 0

def compute_similarity(row_msg, all_messages):
    return [similarity_score(row_msg, msg) for msg in all_messages]

all_messages = df['message'].tolist()
# 并行计算相似度列
ddf['similarity'] = ddf['message'].apply(lambda x: compute_similarity(x, all_messages), meta=('similarity', object))
# 转回Pandas DataFrame
result_df = ddf.compute()
result_df['similarity'] = result_df['similarity'].astype(str)
print(result_df)

额外优化提示 #

• 针对10K行数据，先提取唯一消息计算相似度矩阵，再映射回原DataFrame，可大幅减少重复计算量
• 替换SequenceMatcher为更高效的算法（如fuzzywuzzy.ratio、TF-IDF余弦相似度），能进一步提升计算速度

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