PySpark LSH approxSimilarityJoin运行过慢，求优化方案

我需要为以下两个数据集计算Jaccard指数，当前实现代码如下：

数据集结构 #

df_1(89k条记录):
id     desc_array
1      [dde,jhb..]
3      [bjdhwb,hdjgvg,gd,...]
4      [bvcdhgs,hhj,...]
...

df_2(500条记录):
id     desc_array
1      [dde,jhb..]
2      [bjdhwb,hdjgvg,gd,...]
5      [bvcdhgs,hhj,...]
...

当前实现代码 #

model = Pipeline(stages=[
        HashingTF(inputCol = "desc_array", outputCol="vectors"),
        MinHashLSH(inputCol = "vectors", outputCol="lsh", numHashTables=1000)
    ]).fit(df_1)

db_hashed = model.transform(df_1)

model_final = Pipeline(stages=[
        HashingTF(inputCol = "desc_array", outputCol="vectors"),
        MinHashLSH(inputCol = "vectors", outputCol="lsh", numHashTables=1000)
    ]).fit(df_2)

db_hashed_final = model.transform(df_2)

db_matches = model.stages[-1].approxSimilarityJoin(db_hashed_final, db_hashed, 0.99, distCol="JaccardDistance")
db_matches.write.parquet(file_location)

问题现状 #

当前该任务已运行约4小时仍未完成，数据集规模（89k*500）不大，不应耗时如此之久。当前Spark配置参数如下：

spark_config = {}
spark_config["spark.executor.memory"] = "32G"
spark_config["spark.executor.memoryOverhead"] = "32G"
spark_config["spark.executor.cores"] = "32"
spark_config["spark.driver.memory"] = "2G"

优化方案 #

1. 修正模型训练逻辑，减少哈希表数量 #

• 错误点：当前分别训练两个模型，会导致HashingTF和MinHashLSH的哈希规则不一致，匹配结果无效，同时额外增加训练时间。
• 哈希表优化：numHashTables=1000设置过高，会大幅增加计算和存储开销，通常10-50个哈希表就足够保证近似计算精度。

修改后的代码：

# 仅训练一次模型，保证哈希规则统一
model = Pipeline(stages=[
        HashingTF(inputCol = "desc_array", outputCol="vectors"),
        MinHashLSH(inputCol = "vectors", outputCol="lsh", numHashTables=20)
    ]).fit(df_1)

# 用同一模型转换两个数据集
db_hashed = model.transform(df_1)
db_hashed_final = model.transform(df_2)

# 执行近似匹配
db_matches = model.stages[-1].approxSimilarityJoin(db_hashed_final, db_hashed, 0.99, distCol="JaccardDistance")

2. 调整Spark资源配置，避免资源浪费 #

当前配置的250个executor、32核/executor属于严重资源过载，小数据集不需要这么多资源，过多的executor会导致集群调度开销剧增：

spark_config = {}
spark_config["spark.executor.memory"] = "8G"        # 小数据集无需32G内存
spark_config["spark.executor.memoryOverhead"] = "2G"
spark_config["spark.executor.cores"] = "4"           # 降低单executor核数，减少资源竞争
spark_config["spark.executor.instances"] = "8"       # 减少executor数量到合理范围
spark_config["spark.driver.memory"] = "2G"

3. 优化数据分区 #

检查数据集分区数，过多的小分区会增加调度开销，手动设置合理的分区数：

# 调整为executor总核数的2-3倍即可
db_hashed = db_hashed.repartition(24)
# df_2仅500条记录，分区数无需过多
db_hashed_final = db_hashed_final.repartition(8)

4. 调整相似度阈值，发挥MinHashLSH的过滤作用 #

当前设置的0.99距离阈值（对应Jaccard相似度≥0.01）会导致几乎所有数据对都被匹配，近似join退化为全量笛卡尔积，计算量暴增：

• 如果实际需求是寻找高相似度配对，将阈值调整为0.1（对应相似度≥0.9），让MinHashLSH发挥过滤作用。
• 如果确实需要匹配所有低相似度配对，MinHashLSH不适合，直接计算笛卡尔积更高效：

from pyspark.sql.functions import udf
from pyspark.sql.types import DoubleType

# 自定义UDF计算Jaccard相似度
def jaccard_similarity(arr1, arr2):
    set1 = set(arr1)
    set2 = set(arr2)
    intersection = len(set1 & set2)
    union = len(set1 | set2)
    return intersection / union if union != 0 else 0.0

jaccard_udf = udf(jaccard_similarity, DoubleType())

# 笛卡尔积后计算相似度并过滤
cross_df = db_hashed_final.crossJoin(db_hashed)
result_df = cross_df.withColumn("JaccardSimilarity", jaccard_udf("desc_array", "desc_array"))
result_df = result_df.filter(result_df.JaccardSimilarity >= 0.01)
result_df.write.parquet(file_location)

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