评估来自两个数据帧的所有行组合

Posted

技术标签:

【中文标题】评估来自两个数据帧的所有行组合【英文标题】:Evaluate all combinations of rows from two dataframes 【发布时间】:2016-11-25 14:57:48 【问题描述】:

我有两个大型 spark DataFrame,都包含坐标。我们称它们为地点和地点:

loc = [('01', 0.2, 0.9), ('02', 0.3, 0.6), ('03', 0.8, 0.1)]
locations = sqlContext.createDataFrame(loc, schema=['id', 'X', 'Y'])

site = [('A', 0.7, 0.1), ('B', 0.3, 0.7), ('C', 0.9, 0.3), ('D', 0.3, 0.8)]
sites = sqlContext.createDataFrame(site, schema=['name', 'X', 'Y'])

地点:

+---+---+---+
| id|  X|  Y|
+---+---+---+
| 01|0.2|0.9|
| 02|0.3|0.6|
| 03|0.8|0.1|
+---+---+---+

网站:

+----+---+---+
|name|  X|  X|
+----+---+---+
|   A|0.7|0.1|
|   B|0.3|0.7|
|   C|0.9|0.3|
|   D|0.3|0.8|
+----+---+---+

现在我想以有效的方式计算离站点最近的位置。所以我得到类似的东西:

+----+---+
|name| id|
+----+---+
|   A| 03|
|   B| 02|
|   C| 03|
|   D| 01|
+----+---+

我正在考虑首先制作一个包含所有信息的大型数据框,然后使用 map/reduce 来获取最接近所有站点的位置 ID。但是,我不知道这是否是正确的方法,或者我将如何使用 spark 进行此操作。目前我使用这个:

closest_locations = []
for s in sites.rdd.collect():
    min_dist = float('inf')
    min_loc = None
    for l in locations.rdd.collect():
        dist = (l.X - s.X)**2 + (l.Y - s.Y)**2
        if dist < min_dist:
            min_dist = dist
            min_loc = l.id
    closest_locations.append((s.name, min_loc))

selected_locations = sqlContext.createDataFrame(closest_locations, schema=['name', 'id'])

但我想要一个更像火花的方法,因为上面显然很慢。如何有效地评估两个 spark 数据帧的所有行组合?

【问题讨论】:

【参考方案1】:

你可以:

from pyspark.sql.functions import udf, struct
from pyspark.sql import DoubleType


dist = udf(lamdba x1, y1, x2, y2: (x1 - x2)**2 + (y1 - y1)**2, DoubleType())

locations.join(sites).withColumn("dist", dist(
    locations.X, locations.Y, sites.X, sites.Y)).select(
  "name", struct("id", "dist")
).rdd.reduceByKey(lambda x, y: min(x, y, key=lambda x: x[1]))

【讨论】:

以上是关于评估来自两个数据帧的所有行组合的主要内容,如果未能解决你的问题,请参考以下文章

为 pyspark 数据帧的每一行评估多个 if elif 条件

缓存后正在重新评估 Spark 数据帧

R:从一个数据帧中提取行,基于列名匹配来自另一个数据帧的值

R如何排列数据帧的所有行,以便在列表中返回所有可能的行组合?

Python Pandas从现有数据帧的所有行组合创建新的数据帧

源代码基于扩展贝叶斯方法融合的作战效能评估研究