如何高效利用Spark并行性实现PySpark DataFrame文本相似性匹配？

核心思路：缩小计算范围+利用Spark分布式并行 #

• 按city列做分区/分组处理，避免全量笛卡尔积：
  • 对df_1和df_2都按city分区，让每个Spark任务只处理单个城市的两组数据，把原来数百万级的关联量缩小到城市内的小批量计算（比如某城市df_1有1000行、df_2有500行，仅需处理50万对，而非全量2.1亿对）。
  • 实现时可直接设置DataFrame的分区键为city，或用groupBy("city")触发分组处理，让Spark自动将同城市数据分配到同一executor。

优化广播策略，避免全量广播 #

• 不要直接广播整个df_2，改为按城市广播对应子集：
  • 先将df_2按city分组，每个城市的子集单独广播给处理该城市df_1的任务。通过mapGroups或自定义分区逻辑实现，每个分区仅加载对应城市的df_2数据，大幅降低单广播变量的内存占用与传输耗时。

用向量化Pandas UDF并行计算相似性 #

• 普通UDF逐行处理效率低，换成**Pandas UDF（向量化UDF）**批量处理分区数据：
  • 先统一预处理店铺名称：转小写、去除标点/特殊字符、过滤通用后缀（如“店”“有限公司”），减少干扰项。
  • 编写批量计算相似性的Pandas UDF，示例伪代码：

    from pyspark.sql.functions import pandas_udf, col
    import pandas as pd
    from fuzzywuzzy import fuzz
    
    def match_names(df1_names: pd.Series, df2_names: pd.Series) -> pd.Series:
        # 文本预处理
        df1_clean = df1_names.str.lower().str.replace(r'[^\w\s]', '', regex=True)
        df2_clean = df2_names.str.lower().str.replace(r'[^\w\s]', '', regex=True)
        # 批量匹配，设置相似度阈值（如80）
        def check_match(name):
            similarity_scores = df2_clean.apply(lambda x: fuzz.ratio(name, x))
            return any(similarity_scores >= 80)
        return df1_clean.apply(check_match)
    
    # 按city分组后应用UDF，生成匹配标记
    matched_df = df_1.groupBy("city").applyInPandas(
        lambda df: df.assign(
            is_matched=match_names(
                df["shop_name"], 
                df_2.filter(col("city") == df["city"].iloc[0])["google_shop_name"].toPandas()
            )
        ),
        schema=df_1.schema.add("is_matched", "boolean")
    )
    

先做粗过滤，减少无效计算 #

• 在精确计算相似性前，先通过规则过滤明显不匹配的对：
  • 比如筛选名称长度差≤3的行，或名称包含至少一个共同关键词的行，示例代码：

    from pyspark.sql.functions import length, array_intersect, split, size
    
    # 按city关联后做粗过滤，得到候选匹配对
    candidate_pairs = df_1.join(df_2, on="city") \
        .filter(abs(length(col("shop_name")) - length(col("google_shop_name"))) <= 3) \
        .filter(size(array_intersect(split(col("shop_name"), " "), split(col("google_shop_name"), " "))) >= 1)
    

  • 再对候选对计算精确相似度，最后将匹配结果回写到df_1。

调整Spark配置优化性能 #

• 增大executor内存：如设置spark.executor.memory=8g、spark.driver.memory=4g，避免内存溢出。
• 调整广播阈值：若单城市df_2子集超过默认广播阈值（10MB），可设置spark.sql.autoBroadcastJoinThreshold=50m，让Spark自动广播更大数据集。
• 增加executor核心数：如spark.executor.cores=4，提升并行处理能力。

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