Фрейм данных группы Pyspark за интервал времени - PullRequest
Новая
       98

Фрейм данных группы Pyspark за интервал времени

1 голос
/ 26 мая 2020

У меня есть отсортированный фрейм данных PYSPARK («временная метка» и «корабль» возрастают): 

+----------------------+------+
|        timestamp     | ship |
+----------------------+------+
| 2018-08-01 06:01:00  |    1 |
| 2018-08-01 06:01:30  |    1 |
| 2018-08-01 09:00:00  |    1 |
| 2018-08-01 09:00:00  |    2 |
| 2018-08-01 10:15:43  |    2 |
| 2018-08-01 11:00:01  |    3 |
| 2018-08-01 06:00:13  |    4 |
| 2018-08-01 13:00:00  |    4 |
| 2018-08-13 14:00:00  |    5 |
| 2018-08-13 14:15:03  |    5 |
| 2018-08-13 14:45:08  |    5 |
| 2018-08-13 14:50:00  |    5 |
+-----------------------------+

Я хочу добавить новый столбец в фрейм данных под названием «поездка». Поездка определяется как номер корабля, который отправляется в течение 2 часов с момента начала записи судна в фрейм данных. Если в течение двух часов номер корабля изменится, новый номер рейса должен быть добавлен в столбец «поездка» фрейма данных.

Желаемый результат выглядит так: 

+----------------------+------+-------+
|        timestamp     | ship | trip  |
+----------------------+------+-------+
| 2018-08-01 06:01:00  |    1 |    1  | # start new ship number
| 2018-08-01 06:01:30  |    1 |    1  | # still within 2 hours of same ship number
| 2018-08-01 09:00:00  |    1 |    2  | # more than 2 hours of same ship number = new trip
| 2018-08-01 09:00:00  |    2 |    3  | # new ship number = new trip
| 2018-08-01 10:15:43  |    2 |    3  | # still within 2 hours of same ship number
| 2018-08-01 11:00:01  |    3 |    4  | # new ship number = new trip
| 2018-08-01 06:00:13  |    4 |    5  | # new ship number = new trip
| 2018-08-01 13:00:00  |    4 |    6  | # more than 2 hours of same ship number = new trip
| 2018-08-13 14:00:00  |    5 |    7  | # new ship number = new trip
| 2018-08-13 14:15:03  |    5 |    7  | # still within 2 hours of same ship number
| 2018-08-13 14:45:08  |    5 |    7  | # still within 2 hours of same ship number
| 2018-08-13 14:50:00  |    5 |    7  | # still within 2 hours of same ship number
+-----------------------------+-------+

В Pandas это будет сделано как таковое: 

dt_trip = 2 # time duration trip per ship (in hours)
total_time = df['timestamp'] - df.groupby('name')['timestamp'].transform('min')
trips = total_time.dt.total_seconds().fillna(0)//(dt_trip*3600)
df['trip'] = df.groupby(['name', trips]).ngroup()+1

Как это будет сделано в PYSPARK?

1 Ответ

2 голосов
/ 26 мая 2020

Попробуйте это, используя window functions, row_number(), collect_list() и incremental sum сверх созданных условий . 

from pyspark.sql import functions as F
from pyspark.sql.window import Window

w1=Window().partitionBy("ship").orderBy(F.unix_timestamp("timestamp")).rangeBetween(-7199, Window.currentRow)
w2=Window().partitionBy("ship").orderBy("timestamp")
w3=Window().orderBy("ship","timestamp")

df.withColumn("trip", F.sum(F.when(F.row_number().over(w2)==1, F.lit(1))\
                       .when(F.size(F.collect_list("ship").over(w1))==1, F.lit(1))\
                       .otherwise(F.lit(0))).over(w3)).orderBy("ship","timestamp").show()

#+-------------------+----+----+
#|          timestamp|ship|trip|
#+-------------------+----+----+
#|2018-08-01 06:01:00|   1|   1|
#|2018-08-01 06:01:30|   1|   1|
#|2018-08-01 09:00:00|   1|   2|
#|2018-08-01 09:00:00|   2|   3|
#|2018-08-01 10:15:43|   2|   3|
#|2018-08-01 11:00:01|   3|   4|
#|2018-08-01 06:00:13|   4|   5|
#|2018-08-01 13:00:00|   4|   6|
#|2018-08-13 14:00:00|   5|   7|
#|2018-08-13 14:15:03|   5|   7|
#|2018-08-13 14:45:08|   5|   7|
#|2018-08-13 14:50:00|   5|   7|
#+-------------------+----+----+
...