如何在没有直接连接列的两个数据框之间找到最接近的匹配行？

Posted 2023-04-17

【中文标题】如何在没有直接连接列的两个数据框之间找到最接近的匹配行？

【英文标题】：How to find the closest matching rows in between two dataframes that has no direct join columns?

【发布时间】：2020-07-20 05:45:40

【问题描述】：

对于 pyspark 数据帧中的每一组坐标，我需要在另一个数据帧中找到最近的一组坐标

我有一个带有坐标数据的 pyspark 数据框（数据框 a）：

    +------------------+-------------------+
    |      latitude_deg|      longitude_deg|
    +------------------+-------------------+
    |    40.07080078125| -74.93360137939453|
    |         38.704022|        -101.473911|
    |       59.94919968|     -151.695999146|
    | 34.86479949951172| -86.77030181884766|
    |           35.6087|         -91.254898|
    |        34.9428028|        -97.8180194|

另一个类似的（数据框 b）：（仅显示几行以供理解）

    +-----+------------------+-------------------+
    |ident|      latitude_deg|      longitude_deg|
    +-----+------------------+-------------------+
    |  00A|    30.07080078125| -24.93360137939453|
    | 00AA|         56.704022|        -120.473911|
    | 00AK|       18.94919968|     -109.695999146|
    | 00AL| 76.86479949951172| -67.77030181884766|
    | 00AR|           10.6087|         -87.254898|
    | 00AS|        23.9428028|        -10.8180194|

是否有可能以某种方式合并数据帧以使数据帧 a 中的每一行都具有与数据帧 b 最接近的标识：

    +------------------+-------------------+-------------+
    |      latitude_deg|      longitude_deg|closest_ident|
    +------------------+-------------------+-------------+
    |    40.07080078125| -74.93360137939453|          12A|
    |         38.704022|        -101.473911|         14BC|
    |       59.94919968|     -151.695999146|         278A|
    | 34.86479949951172| -86.77030181884766|         56GH|
    |           35.6087|         -91.254898|         09HJ|
    |        34.9428028|        -97.8180194|         09BV|

到目前为止我所尝试的：

我有一个 pyspark UDF 来计算定义的 2 对坐标之间的半正弦距离。

    udf_get_distance = F.udf(get_distance)

它是这样工作的：

    df = (df.withColumn(“ABS_DISTANCE”, udf_get_distance(
         df.latitude_deg_a, df.longitude_deg_a,
         df.latitude_deg_b, df.longitude_deg_b,)
    ))

如果能提供任何帮助，我将不胜感激。非常感谢

【参考方案1】：

您需要先进行 crossJoin。像这样的

joined_df=source_df1.crossJoin(source_df2)

然后你可以像你提到的那样调用你的udf，根据距离生成rownum并过滤掉接近的

from pyspark.sql.functions import row_number,Window
rwindow=Window.partitionBy("latitude_deg_a","latitude_deg_b").orderBy("ABS_DISTANCE")

udf_result_df = joined_df.withColumn(“ABS_DISTANCE”, udf_get_distance(
         df.latitude_deg_a, df.longitude_deg_a,
         df.latitude_deg_b, df.longitude_deg_b).withColumn("rownum",row_number().over(rwindow)).filter("rownum = 1")

注意：将返回类型添加到您的 udf