如何优化Spark环境下Fuzzywuzzy匹配两个大型数据集的性能

核心问题定位 #

你当前代码的性能瓶颈来自两个致命问题：

1. 完全浪费Spark并行能力：将240万条全量数据collect到Driver节点单线程跑for循环，6核CPU完全闲置，所有计算压力都压在单个进程上。
2. 无剪枝的全量匹配：每条大数据集记录都要和3.8万条小数据集全量做模糊匹配，总计算量高达912亿次，7小时运行时长属于正常情况。

具体优化措施 #

1. 分布式并行改造，用满硬件资源 #

小数据集仅3.8万条，体积非常小，直接广播到所有工作节点，大数据集按分区分布式处理，每个分区的记录独立做匹配，完全利用你的6核CPU资源。

2. 前置剪枝，大幅降低比对次数 #

车辆匹配场景下很多字段可以做前置过滤，不需要全量走模糊匹配：

• 按品牌+车型分组过滤：把小数据集按(小写(name_make), 小写(name_model))做分组键，大数据集每条记录也生成对应分组键，仅和同组的小数据集记录做匹配，大部分场景下单组候选记录不会超过100条，比对量直接下降2个数量级。
• 数值字段前置过滤：排量、功率、座位数、车门数这些数值字段如果差值超过阈值（比如排量差超过50cc，功率差超过5kw），直接跳过匹配，不用走字符串相似度计算。

3. 匹配逻辑提速 #

• 依赖替换：安装python-Levenshtein库，fuzzywuzzy调用时会自动切换为C实现的匹配算法，比纯Python实现快10~20倍。
• 分字段加权算分：不要把所有字段拼接成整串匹配，拆分字段分别算分再加权求和：品牌占20分、车型占25分、燃料类型占15分、排量占20分、变速箱占10分、车身类型占10分，数值类字段直接精确判断得分，仅对品牌、车型这类非标准字符串做模糊匹配，既提升准确率又提升速度。

优化后代码示例 #

from fuzzywuzzy import fuzz
import pyspark.sql.functions as F

# 预处理小数据集，按品牌车型分组后广播
header2 = ["name_make", "name_model", "name_fueltype", "cap_ccm", "pwr_km_base", "pwr_kw_base",
           "name_transmission_et", "name_body_et", "name_drivetype_et", "cnt_door", "cnt_seat", "id"]
df2_pre = df2.withColumn("match_key", F.concat(F.lower(F.col("name_make")), F.lit("_"), F.lower(F.col("name_model"))))
# 转成分组字典广播到所有节点
df2_group_dict = df2_pre.rdd.map(lambda x: (x["match_key"], x.asDict())).groupByKey().mapValues(list).collectAsMap()
broadcast_df2 = sparkSession.sparkContext.broadcast(df2_group_dict)

# 单条记录匹配逻辑
def match_single_record(row):
    row_dict = row.asDict()
    # 生成分组匹配键
    match_key = f"{row_dict['name_make_ad'].lower()}_{row_dict['name_model_ad'].lower()}"
    candidates = broadcast_df2.value.get(match_key, [])
    if not candidates:
        return (row_dict["concatenated1"], None)
    best_score = 0
    best_id = None
    for cand in candidates:
        # 数值字段前置过滤
        if abs(row_dict["engine_capacity_ccm"] - cand["cap_ccm"]) > 50:
            continue
        if abs(row_dict["engine_power_kw"] - cand["pwr_kw_base"]) > 5:
            continue
        if row_dict["nr_seats_ad"] != cand["cnt_seat"]:
            continue
        # 分字段算分
        score = 0
        score += fuzz.ratio(row_dict["name_fueltype_ad"].lower(), cand["name_fueltype"].lower()) * 0.15
        score += fuzz.ratio(row_dict["name_transmission_ad"].lower(), cand["name_transmission_et"].lower()) * 0.1
        score += fuzz.ratio(row_dict["name_body_ad"].lower(), cand["name_body_et"].lower()) * 0.1
        score += fuzz.ratio(row_dict["name_drivetype_ad"].lower(), cand["name_drivetype_et"].lower()) * 0.1
        # 数值字段直接给固定分
        score += 20 if abs(row_dict["engine_capacity_ccm"] - cand["cap_ccm"]) < 10 else 10
        score += 20 if abs(row_dict["engine_power_kw"] - cand["pwr_kw_base"]) < 3 else 10
        score += 15 if row_dict["nr_seats_ad"] == cand["cnt_seat"] else 0
        # 记录最优匹配
        if score > best_score and score >= 50:
            best_score = score
            best_id = cand["id"]
    return (row_dict["concatenated1"], best_id)

# 分布式处理大数据集
header1 = ["name_make_ad", "name_model_ad", "name_fueltype_ad", "engine_capacity_ccm", "engine_power_km",
           "engine_power_kw", "name_transmission_ad", "name_body_ad", "name_drivetype_ad", "nr_door_ad",
           "nr_seats_ad"]
df1_pre = df1.withColumn("concatenated1", F.concat_ws(", ", *[F.col(x) for x in header1]))
matches_rdd = df1_pre.rdd.map(match_single_record).filter(lambda x: x[1] is not None)
dfIndexes = matches_rdd.toDF(["concatenated1", "id"])

# 关联得到最终结果
dfMapping = df1_pre.join(dfIndexes, on="concatenated1", how="inner")\
    .join(df2, on="id", how="inner")\
    .drop("concatenated1", "id")

本地运行调优 #

启动Spark会话时调整配置，充分利用你的16G内存：

sparkSession = SparkSession.builder \
    .master("local[6]") \
    .config("spark.driver.memory", "8g") \
    .config("spark.executor.memory", "4g") \
    .config("spark.python.worker.memory", "2g") \
    .getOrCreate()

按上述方案调整后，你的数据量预计10~30分钟即可完成全部匹配。

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