如何使用（Py）Spark對數據集中數據點之間的距離進行求和？

Question

我有一段時間內用戶Lat/Lon格式位置的數據集。我想計算這些用戶旅行的距離。示例數據集：

|時間戳|用戶|緯度|經度| | 1462838468 | 49B4361512443A4DA ... | 39.777982 | -7.054599 | | 1462838512 | 49B4361512443A4DA ... | 39.777982 | -7.054599 | | 1462838389 | 49B4361512443A4DA ... | 39.777982 | -7.054599 | | 1462838497 | 49B4361512443A4DA ... | 39.777982 | -7.054599 | | 1465975885 | 6E9E0581E2A032FD8 ... | 37.118362 | -8.205041 | | 1457723815 | 405C238E25FE0B9E7 ... | 37.177322 | -7.426781 | | 1457897289 | 405C238E25FE0B9E7 ... | 37.177922 | -7.447443 | | 1457899229 | 405C238E25FE0B9E7 ... | 37.177922 | -7.447443 | | 1457972626 | 405C238E25FE0B9E7 ... | 37.18059 | -7.46128 | | 1458062553 | 405C238E25FE0B9E7 ... | 37.177322 | -7.426781 | | 1458241825 | 405C238E25FE0B9E7 ... | 37.178172 | -7.444512 | | 1458244457 | 405C238E25FE0B9E7 ... | 37.178172 | -7.444512 | | 1458412513 | 405C238E25FE0B9E7 ... | 37.177322 | -7.426781 | | 1458412292 | 405C238E25FE0B9E7 ... | 37.177322 | -7.426781 | | 1465197963 | 6E9E0581E2A032FD8 ... | 37.118362 | -8.205041 | | 1465202192 | 6E9E0581E2A032FD8 ... | 37.118362 | -8.205041 | | 1465923817 | 6E9E0581E2A032FD8 ... | 37.118362 | -8.205041 | | 1465923766 | 6E9E0581E2A032FD8 ... | 37.118362 | -8.205041 | | 1465923748 | 6E9E0581E2A032FD8 ... | 37.118362 | -8.205041 | | 1465923922 | 6E9E0581E2A032FD8 ... | 37.118362 | -8.205041 |

我想過使用自定義聚合函數，但似乎沒有Python支持。此外，操作需要在特定順序的相鄰點上完成，所以我不知道自定義聚合器是否可以工作。

我也看過reduceByKey，但距離函數似乎並未滿足運算符的要求。

有沒有辦法在Spark中以有效的方式執行此操作？

Answer 1

它看起來像一個窗口函數的工作。假設我們定義距離爲：

from pyspark.sql.window import Window 

w = Window().partitionBy("User").orderBy("Timestamp") 

和連續觀測之間的計算距離使用lag：

from pyspark.sql.functions import acos, cos, sin, lit, toRadians 

def dist(long_x, lat_x, long_y, lat_y): 
    return acos(
     sin(toRadians(lat_x)) * sin(toRadians(lat_y)) + 
     cos(toRadians(lat_x)) * cos(toRadians(lat_y)) * 
      cos(toRadians(long_x) - toRadians(long_y)) 
    ) * lit(6371.0) 

您可以按照定義窗口

from pyspark.sql.functions import lag 

df.withColumn("dist", dist(
    "longitude", "latitude", 
    lag("longitude", 1).over(w), lag("latitude", 1).over(w) 
).alias("dist")) 

之後，你可以執行標準的聚集。

Answer 2

您可以使用SparkSQL中的Window函數來計算兩點之間的距離，然後使用groupby來計算每個用戶的距離。

下面是一個代碼示例：https://gist.github.com/mr1azl/71f376e7efb6637a43b26a83e90d304a