Pyspark - 根据时间和位置计算组中的平均速度答案

【问题标题】：Pyspark - calculate average velocity in group based on time and locationPyspark - 根据时间和位置计算组中的平均速度
【发布时间】：2020-05-31 02:15:29
【问题描述】：

我有以下 PYSPARK 数据框：

+-------------------+----+---------+------+
|      timestamplast|ship|  X_pos  |time_d|
+-------------------+----+---------+------+
|2019-08-01 00:00:00|   1|     3   |   0  |
|2019-08-01 00:00:09|   1|     4   |   9  |
|2019-08-01 00:00:20|   1|     5   |  11  |
|2019-08-01 00:00:27|   1|     9   |   7  |
|2019-08-01 00:00:38|   2|     3   |   0  |
|2019-08-01 00:00:39|   2|     8   |   1  |
|2019-08-01 00:00:57|   2|     20  |  18  |
+-------------------+----+---------+------+

其中 timestamplast 是日期时间，time_d 是组“ship”内的时间差（当新的“ship”启动时，time_d 为零。我想计算组“ship”内的平均速度并将结果附加到数据帧基于时差和位置X_pos

ship==1 的平均速度为：(1/9 + 1/11 + 4/7)/3 = 0.26 m/s。 ship==2 的平均速度为：(5/1 + 12/18 /2 = 2.83 m/s。

编辑： ship==1 的平均速度为：((4-3)/(9) + (5-4)/(11) + (9-5)/(7))/3 = 0.26 m/s。 ship==2 的平均速度为：((8-3)/1 + ((20-8)/18)) /2 = 2.83 m/s。

结果应如下所示：

+-------------------+----+---------+------+-----------+
|      timestamplast|name|     X   |time_d| avg_vel_x |
+-------------------+----+---------+------+-----------|
|2019-08-01 00:00:00|   1|     3   |   0  |     0.26  |
|2019-08-01 00:00:09|   1|     4   |   9  |     0.26  |
|2019-08-01 00:00:20|   1|     5   |  11  |     0.26  |
|2019-08-01 00:00:27|   1|     9   |   7  |     0.26  |
|2019-08-01 00:00:38|   2|     3   |   0  |     2.83  |
|2019-08-01 00:00:39|   2|     8   |   1  |     2.83  |
|2019-08-01 00:00:57|   2|     20  |  18  |     2.83  |
+-------------------+----+---------+------+-----------|

【问题讨论】：

你能解释一下计算部分吗？
我的错，在 Pandas 中应该是：df['vel_x'] = ((df.groupby("name")['X'].diff().fillna(0)) /(df['time_d']).replace([np.inf,-1*np.inf],0)).fillna(0)。然后：df['avg_vel'] = df.groupby(['name'])['vel_x'].transform('mean')
for ship == 2 你的输出总和不正确..你能检查一下吗？
为什么最后把1除以3、2除以2？另外请不要再编辑问题，因为这会浪费大家的时间
确实你是对的......这是漫长的一天。

标签： pyspark group-by pyspark-dataframes

【解决方案1】：

pandas 中的transform 可以通过类似于 sql 的 pyspark 中的 windows 函数进行复制，ship == 1 的预期输出应该是 0.26。你可以试试：

import pyspark.sql.functions as F

w = Window.partitionBy('ship')
pct_change=((F.col("X_pos")-F.lag("X_pos").over(w.orderBy("timestamplast")))
                                                /F.col("time_d"))
df.withColumn("avg_vel_x",F.round(F.sum(pct_change).over(w)
                                 /(F.count("ship").over(w)-1),2)).show()

+-------------------+----+-----+------+---------+
|      timestamplast|ship|X_pos|time_d|avg_vel_x|
+-------------------+----+-----+------+---------+
|2019-08-01 00:00:00|   1|    3|     0|     0.26|
|2019-08-01 00:00:09|   1|    4|     9|     0.26|
|2019-08-01 00:00:20|   1|    5|    11|     0.26|
|2019-08-01 00:00:27|   1|    9|     7|     0.26|
|2019-08-01 00:00:38|   2|    3|     0|     2.83|
|2019-08-01 00:00:39|   2|    8|     1|     2.83|
|2019-08-01 00:00:57|   2|   20|    18|     2.83|
+-------------------+----+-----+------+---------+

【讨论】：