Использование Windows для группировки 5 минут - PullRequest
Новая
       27

Использование Windows для группировки 5 минут

1 голос
/ 26 апреля 2020

CSV-файл: 

#+----+-----------+-------------------+
#|col1|       col2|          timestamp|
#+----+-----------+-------------------+
#|   0|Town Street|01-02-2017 06:01:00|
#|   0|Town Street|01-02-2017 06:03:00|
#|   0|Town Street|01-02-2017 06:05:00|
#|   0|Town Street|01-02-2017 06:06:00|
#|   0|Town Street|02-02-2017 10:01:00|
#|   0|Town Street|02-02-2017 10:05:00|
#+----+-----------+-------------------+

сравните время каждой даты, чтобы увидеть, есть ли разница в 5 минут, если их считать их

вывод: 

 #+----+-----------+-------------------+
#|col1|       col2|          timestamp|
#+----+-----------+-------------------+
#|   0|Town Street|01-02-2017 06:01:00|
#|   0|Town Street|01-02-2017 06:03:00|
#|   0|Town Street|01-02-2017 06:05:00|
#|   0|Town Street|01-02-2017 06:06:00|
#|   0|Town Street|02-02-2017 10:01:00|
#|   0|Town Street|02-02-2017 10:05:00|
#+----+-----------+-------------------+

Код прямо сейчас: 

from pyspark.sql import SQLContext
import pyspark.sql.functions as F

    def my_main(sc, my_dataset_dir):
        sqlContext = SQLContext(sc)
        df = sqlContext.read.csv(my_dataset_dir,sep=';').rdd.zipWithIndex().filter(lambda x: x[1] > 1).map(lambda x: x[0]).toDF(['status','title','datetime'])

Этот код просто дает нулевой результат в течение 5 мин.

1 Ответ

1 голос
/ 26 апреля 2020

Не уверен, что это именно то, что вы хотите, но это должно подтолкнуть вас sh в правильном направлении . Вы можете преобразовать свою метку времени в timestamptype и datetype. Для создания window до partitionBy даты и rangebetween отметка времени в seconds(300). 

#df.show()
#sampledataframe
#+----+-----------+-------------------+
#|col1|       col2|          timestamp|
#+----+-----------+-------------------+
#|   0|Town Street|01-02-2017 06:01:00|
#|   0|Town Street|01-02-2017 06:03:00|
#|   0|Town Street|01-02-2017 06:05:00|
#|   0|Town Street|01-02-2017 06:06:00|
#|   0|Town Street|02-02-2017 10:01:00|
#|   0|Town Street|02-02-2017 10:05:00|
#+----+-----------+-------------------+

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

w=Window().partitionBy("date").orderBy(F.col("timestamp").cast("long")).rangeBetween(Window.currentRow,60*5)

df.withColumn("timestamp", F.to_timestamp("timestamp",'MM-dd-yyyy HH:mm:ss'))\
  .withColumn("date", F.to_date("timestamp"))\
  .withColumn('collect', F.size(F.collect_list("timestamp").over(w))).filter("collect>1")\
  .select(F.date_format("date","yyyy-MM-dd").alias("date"), F.array(F.date_format("timestamp","HH:mm:ss"),F.col("collect")).alias("time"))\
  .orderBy("date").show()

#+----------+-------------+
#|      date|         time|
#+----------+-------------+
#|2017-01-02|[06:01:00, 4]|
#|2017-01-02|[06:05:00, 2]|
#|2017-01-02|[06:03:00, 3]|
#|2017-02-02|[10:01:00, 2]|
#+----------+-------------+
...