Pyspark levenshtein Join 卡在同一个阶段

如何解决Pyspark levenshtein Join 卡在同一个阶段

我想根据 Levenshtein 距离执行连接。

我有 2 个数据框：

• 数据：13 万行的小数据框。

• siren：大型数据框，表示数据库 stockunitelegale 的表 SIREN，有 20M 行。

我尝试广播小数据帧并执行与大数据集的连接。 Spark Job 总是卡在 join 的最后三个任务上...

data=broadcast(data)
df=siren.join(spk,1-(fn.levenshtein(data["description_nom"],siren["denominationUniteLegale"])/ fn.greatest(fn.length(data["description_nom"]),fn.length(siren["denominationUniteLegale"])))>0.7)
df.write.csv("join_siren.csv")

我还尝试增加执行程序的数量、内核数量和分配的内存，但在处理方面我有相同的行为。

--num-executors 20 --conf "spark.executor.memory=12g" --conf "spark.executor.cores=15" --conf "spark.sql.shuffle.partitions=1000" --conf "spark.default.parallelism=1000" --conf "spark.driver.maxResultSize=8g" --conf "spark.sql.autobroadcastJoinThreshold=20485760" --conf "spark.network.timeout=10000000"

请问您有针对此类问题的解决方案吗？

解决方法

即使你的一个数据框很小，你本质上做的是一个带有这个连接的笛卡尔积。因为为了让 spark 知道哪个行组合满足您的条件，它必须首先从连接的左侧和右侧数据帧创建行的每个单个组合。这意味着 130k 乘以 20m，即两万亿六千亿行。挺大的。

我的建议是首先加入一些其他条件，这会减少模糊名称匹配的搜索空间。例如，您可以使用诸如邮政编码之类的东西。其他一些示例可能是在空间上拆分列 description_nom 和 denominationUniteLegale 并加入任何标记匹配。然后过滤

    1-(fn.levenshtein(data["description_nom"],siren["denominationUniteLegale"])/ fn.greatest(fn.length(data["description_nom"]),fn.length(siren["denominationUniteLegale"])))>0.7

这取决于您在数据框中可用的内容。您将不得不尝试一些有效的方法。例如，使用邮政编码建议，您可以这样做：

df=siren.join(data,data["postcode"]==siren["codePostalEtablissement"])
df=df.filter(1-(fn.levenshtein(data["description_nom"],fn.length(siren["denominationUniteLegale"])))>0.7)
df.write.csv("join_siren.csv")

当然，这只是一个示例，您可以在连接中组合不同的条件，例如在邮政编码或街道名称或名称标记匹配上连接。但是您仍然需要检查它是否不会像使用 cartesianJoin 那样生成太多行。为此，您必须在最坏的情况下检查您的任何状况。以邮政编码为例，如果按邮政编码分组计算data和siren中的行数，最大的组合是什么。假设您在 postcode 中有一个名为 data 的列，您可以按如下方式获取此信息：

dataPostcodeCount = data.groupBy('postcode').count()
sirenPostcodeCount = siren.groupBy('codePostalEtablissement').count()
postCodeCounts = dataPostcodeCount \
    .join(sirenPostcodeCount,data['postcode'] == siren['codePostalEtablissement']) \
    .select('postcode',(data['count']*siren['count']).alias('count'))
postCodeCounts.sortBy(postCodeCounts['count'].desc()).show()