apply groupby over window in a continuous manner pyspark

user_7641766 0
Viewed 362

I want to apply groupby with a time window of 60 minutes but it only collects the value in the hour it has appeared and does not show anything for a window where there is no value.

I want it in a way that for the window without any value it gives 0 so as to have the data in a more continuous fashion.

for example:

df = sc.parallelize(
  [Row(datetime='2015/01/01 03:00:36', value = 2.0),
   Row(datetime='2015/01/01 03:40:12', value = 3.0),
   Row(datetime='2015/01/01 05:25:30', value = 1.0)]).toDF()

df1 = df.select(sf.unix_timestamp(sf.column("datetime"), 'yyyy/MM/dd HH:mm:ss').cast(TimestampType()).alias("timestamp"), sf.column("value"))

df1.groupBy(sf.window(sf.col("timestamp"), "60 minutes")).agg(sf.sum("value")).show(truncate = False)

the output i get is:

+------------------------------------------+----------+
|window                                    |sum(value)|
+------------------------------------------+----------+
|[2015-01-01 03:00:00, 2015-01-01 04:00:00]|5.0       |
|[2015-01-01 05:00:00, 2015-01-01 06:00:00]|1.0       |
+------------------------------------------+----------+

whereas, I would rather want the output to be:

+------------------------------------------+----------+
|window                                    |sum(value)|
+------------------------------------------+----------+
|[2015-01-01 03:00:00, 2015-01-01 04:00:00]|5.0       |
|[2015-01-01 04:00:00, 2015-01-01 05:00:00]|0.0       |
|[2015-01-01 05:00:00, 2015-01-01 06:00:00]|1.0       |
+------------------------------------------+----------+

Edit:

How do i then extend it to double groupby and equal number of windows for each "name":

df = sc.parallelize(
  [Row(name = 'ABC', datetime = '2015/01/01 03:00:36', value = 2.0),
   Row(name = 'ABC', datetime = '2015/01/01 03:40:12', value = 3.0),
   Row(name = 'ABC', datetime = '2015/01/01 05:25:30', value = 1.0),
   Row(name = 'XYZ', datetime = '2015/01/01 05:15:30', value = 2.0)]).toDF()

df1 = df.select('name', sf.unix_timestamp(sf.column("datetime"), 'yyyy/MM/dd HH:mm:ss').cast(TimestampType()).alias("timestamp"), sf.column("value"))

df1.show(truncate = False)

>>>+----+-------------------+-----+
|name|timestamp          |value|
+----+-------------------+-----+
|ABC |2015-01-01 03:00:36|2.0  |
|ABC |2015-01-01 03:40:12|3.0  |
|ABC |2015-01-01 05:25:30|1.0  |
|XYZ |2015-01-01 05:15:30|2.0  |
+----+-------------------+-----+

and i want the result to be:

+----+------------------------------------------+----------+
|name|window                                    |sum(value)|
+----+------------------------------------------+----------+
|ABC |[2015-01-01 03:00:00, 2015-01-01 04:00:00]|5.0       |
|ABC |[2015-01-01 04:00:00, 2015-01-01 05:00:00]|0.0       |
|ABC |[2015-01-01 05:00:00, 2015-01-01 06:00:00]|1.0       |
|XYZ |[2015-01-01 03:00:00, 2015-01-01 04:00:00]|0.0       |
|XYZ |[2015-01-01 04:00:00, 2015-01-01 05:00:00]|0.0       |
|XYZ |[2015-01-01 05:00:00, 2015-01-01 06:00:00]|2.0       |
+----+------------------------------------------+----------+
pyspark-dataframes pyspark apache-spark-sql apache-spark python
1 Answers

This is actually the behavior of grouping by window as you have no corresponding rows between hours 4 and 5.

However, you can make it work by generating the intervals in a separate dataframe using sequence function and starting from min(timestamp) to max(timestamp) truncated to hours. Then, using transfrom function on the generated sequence to create structs of strat and end time of each bucket :

from pyspark.sql import functions as sf

buckets = df1.agg(
    sf.expr("""transform(
                sequence(date_trunc('hour', min(timestamp)), 
                         date_trunc('hour', max(timestamp)), 
                         interval 1 hour
                ),
                x -> struct(x as start, x + interval 1 hour as end)
              )
    """).alias("buckets")
).select(sf.explode("buckets").alias("window"))

buckets.show(truncate=False)
#+------------------------------------------+
#|window                                    |
#+------------------------------------------+
#|[2015-01-01 03:00:00, 2015-01-01 04:00:00]|
#|[2015-01-01 04:00:00, 2015-01-01 05:00:00]|
#|[2015-01-01 05:00:00, 2015-01-01 06:00:00]|
#+------------------------------------------+

Now, you join with the original dataframe and groupby window column to sum the value :

df2 = buckets.join(
    df1,
    (sf.col("timestamp") >= sf.col("window.start")) &
    (sf.col("timestamp") < sf.col("window.end")),
    "left"
).groupBy("window").agg(
    sf.sum(sf.coalesce(sf.col("value"), sf.lit(0))).alias("sum")
)

df2.show(truncate=False)
#+------------------------------------------+---+
#|window                                    |sum|
#+------------------------------------------+---+
#|[2015-01-01 04:00:00, 2015-01-01 05:00:00]|0.0|
#|[2015-01-01 03:00:00, 2015-01-01 04:00:00]|5.0|
#|[2015-01-01 05:00:00, 2015-01-01 06:00:00]|1.0|
#+------------------------------------------+---+
user_13865510
Related