通过在另一个数据框中的所有行上迭代数据框中的一行来检查最小值答案

【问题标题】：Check the minimum by iterating one row in a dataframe over all the rows in another dataframe通过在另一个数据框中的所有行上迭代数据框中的一行来检查最小值
【发布时间】：2021-12-10 13:07:15
【问题描述】：

假设我有以下两个数据框：

DF1:
+----------+----------+----------+
|     Place|Population|    IndexA|     
+----------+----------+----------+
|         A|       Int|       X_A|
|         B|       Int|       X_B|
|         C|       Int|       X_C|
+----------+----------+----------+

DF2:
+----------+----------+
|      City|    IndexB|     
+----------+----------+
|         D|       X_D|      
|         E|       X_E|   
|         F|       X_F|  
|      ....|      ....|
|        ZZ|      X_ZZ|
+----------+----------+

上面的数据框通常要大得多。

我想确定从DF1 到每个Place 到哪个City(DF2) 的最短距离。可以根据索引计算距离。所以对于DF1 中的每一行，我必须遍历DF2 中的每一行，并根据索引的计算寻找最短距离。对于距离计算，定义了一个函数：

val distance = udf(
      (indexA: Long, indexB: Long) => {
        h3.instance.h3Distance(indexA, indexB)
      })

我尝试了以下方法：

val output =  DF1.agg(functions.min(distance(col("IndexA"), DF2.col("IndexB"))))

但是这个，代码编译但是我得到以下错误：

线程“主”org.apache.spark.sql.AnalysisException 中的异常：已解决的属性
Places#316,Population#330,IndexAx#338L 中缺少 H3Index#220L 在运算符 !Aggregate
[min(if ((isnull(IndexA#338L) OR isnull(IndexB#220L))) null else UDF(knownnotnull(IndexA#338L), knownnotnull(IndexB#220L))) AS min(UDF(IndexA, IndexB))#346].

所以我想我在从DF1 获取一行时迭代DF2 中的每一行时做错了，但我找不到解决方案。

我做错了什么？我的方向是否正确？

【问题讨论】：

标签： scala dataframe apache-spark apache-spark-sql uber-api

【解决方案1】：

您收到此错误是因为您使用的索引列仅存在于 DF2 而不是您尝试执行聚合的 DF1。

为了使该字段可访问并确定与所有点的距离，您需要

交叉连接DF1 和Df2 使Df1 的每个索引匹配DF2 的每个索引
使用您的 udf 确定距离
在这个带有距离的新十字架上找到最小值

这可能看起来像：

import org.apache.spark.sql.expressions.Window
import org.apache.spark.sql.functions.{col, min, udf}

val distance = udf(
      (indexA: Long, indexB: Long) => {
        h3.instance.h3Distance(indexA, indexB)
      })

val resultDF = DF1.crossJoin(DF2)
    .withColumn("distance", distance(col("IndexA"), col("IndexB")))
    //instead of using a groupby then matching the min distance of the aggregation with the initial df. I've chosen to use a window function min to determine the min_distance of each group (determined by Place) and filter by the city with the min distance to each place
    .withColumn("min_distance", min("distance").over(Window.partitionBy("Place")))
    .where(col("distance") === col("min_distance"))
    .drop("min_distance")

这将产生一个数据框，其中包含来自数据框和附加列 distance 的列。

NB。您当前将一个 df 中的每个项目与另一个 df 中的每个项目进行比较的方法是一项昂贵的操作。如果您有机会尽早过滤（例如加入启发式列，即其他可能表明某个地方可能更靠近城市的列），建议您这样做。

让我知道这是否适合你。

【讨论】：

此方法有效。这确实是一项非常昂贵的手术。但是在高效使用.cache() 之后，确实将成本降到最低。谢谢@ggordon
还有一点，A地和B市之间的距离最小。但是上面的解只显示了最小距离，而没有显示最小距离是到哪个城市。我该怎么做？
距离最小的城市在最终数据框的City列中

【解决方案2】：

如果您只有几个城市（少于或大约 1000 个），您可以通过将城市收集到一个数组中来避免 crossJoin 和 Window 洗牌，然后使用这个收集的数组为每个地方执行距离计算：

import org.apache.spark.sql.functions.{array_min, col, struct, transform, typedLit, udf}

val citiesIndexes = df2.select("City", "IndexB")
  .collect()
  .map(row => (row.getString(0), row.getLong(1)))

val result = df1.withColumn(
  "City",
  array_min(
    transform(
      typedLit(citiesIndexes),
      x => struct(distance(col("IndexA"), x.getItem("_2")), x.getItem("_1"))
    )
  ).getItem("col2")
)

这段代码适用于 Spark 3 及更高版本。如果您使用的 Spark 版本小于 3.0，则应将 array_min(...).getItem("col2") 部分替换为用户定义的函数。

【讨论】：