坐标数据的 PySpark Dataframe 转换答案

【问题标题】：PySpark Dataframe transformation for coordinate data坐标数据的 PySpark Dataframe 转换
【发布时间】：2023-02-02 20:25:29
【问题描述】：

我在 PySpark 中有以下数据框，其中每个条目都是具有“恒定”值 ID、开始和停止以及不同坐标的旅程的位置。

Id	Lat	Lon	Start	Stop
1	40.5	40	A	B
1	41.0	45	A	B
1	40.5	40	A	B
2	31.4	59	A	C
2	34.5	60	A	C
2	37.0	61	A	C
...

我想将此数据框转换为

Id	Start	Stop	Trajectory
1	A	B	Vector of Coordinates
2	A	C	Vector of Coordinates
...

我的第一个想法是遍历数据框中的每一行，为每个新的 Id 创建一个“旅行对象”，其中包含 Id、Start 和 Stop 以及一个轨迹列表，然后将纬度和经度作为 Coordinate 对象附加到轨迹列表中，然而，由于数据集很大，这可能在计算上很昂贵。

我以前从未使用过 Spark，因此可能有一些巧妙的技巧可以使它变得非常高效。

【问题讨论】：

标签： python pyspark apache-spark-sql

【解决方案1】：

我不确定你到底需要什么输出，但检查此代码作为改进或讨论的开始。当然，您可以更改 groupBy 中使用的列，添加排序或创建点的方式（我只是添加 lat 和 long，由 :) 分隔

import pyspark.sql.functions as F

inputData = [
    ("1", "40.5", "40", "A", "B"),
    ("1", "41.0", "45", "A", "B"),
    ("1", "40.5", "40", "A", "B"),
    ("2", "31.4", "59", "A", "C"),
    ("2", "34.5", "60", "A", "C"),
    ("2", "37.0", "61", "A", "C"),
]

df = spark.createDataFrame(inputData, schema=["id", "Lat", "Lon", "Start", "Stop"])

aggregatedDf = (
    df.withColumn("Point", F.concat_ws(':', F.col("Lat"), F.col("Lon")))
    .groupBy("id", "Start", "Stop")
    .agg(F.collect_list("Point").alias('Trajectory'))
)

aggregatedDf.show(truncate = False)

示例输出：

+---+-----+----+---------------------------+
|id |Start|Stop|Trajectory                 |
+---+-----+----+---------------------------+
|1  |A    |B   |[40.5:40, 41.0:45, 40.5:40]|
|2  |A    |C   |[31.4:59, 34.5:60, 37.0:61]|
+---+-----+----+---------------------------+

【讨论】：