How to aggregate over rolling time window with groups in Spark

by

Revised answer:

You can use a simple window functions trick here. A bunch of imports:

from pyspark.sql.functions import coalesce, col, datediff, lag, lit, sum as sum_
from pyspark.sql.window import Window

window definition:

w = Window.partitionBy("group_by").orderBy("date")

Cast date to DateType:

df_ = df.withColumn("date", col("date").cast("date"))

Define following expressions:

# Difference from the previous record or 0 if this is the first one
diff = coalesce(datediff("date", lag("date", 1).over(w)), lit(0))

# 0 if diff <= 30, 1 otherwise
indicator = (diff > 30).cast("integer")

# Cumulative sum of indicators over the window
subgroup = sum_(indicator).over(w).alias("subgroup")

Add subgroup expression to the table:

df_.select("*", subgroup).groupBy("group_by", "subgroup").avg("get_avg")

+--------+--------+------------+
|group_by|subgroup|avg(get_avg)|
+--------+--------+------------+
|  group1|       0|         5.0|
|  group2|       0|        20.0|
|  group2|       1|         8.0|
+--------+--------+------------+

first is not meaningful with aggregations, but if column is monotonically increasing you can use min. Otherwise you’ll have to use window functions as well.

Tested using Spark 2.1. May require subqueries and Window instance when used with earlier Spark release.

The original answer (not relevant in the specified scope)

Since Spark 2.0 you should be able to use a window function:

Bucketize rows into one or more time windows given a timestamp specifying column. Window starts are inclusive but the window ends are exclusive, e.g. 12:05 will be in the window [12:05,12:10) but not in [12:00,12:05). 

from pyspark.sql.functions import window

df.groupBy(window("date", windowDuration="30 days")).count()

but you can see from the result,

+---------------------------------------------+-----+
|window                                       |count|
+---------------------------------------------+-----+
|[2016-01-30 01:00:00.0,2016-02-29 01:00:00.0]|1    |
|[2015-12-31 01:00:00.0,2016-01-30 01:00:00.0]|2    |
|[2016-03-30 02:00:00.0,2016-04-29 02:00:00.0]|1    |
+---------------------------------------------+-----+

you’ll have to be a bit careful when it comes to timezones.

More Related Contents:

  1. Spark Window Functions – rangeBetween dates
  2. Spark SQL window function with complex condition
  3. How to select the first row of each group?
  4. Querying Spark SQL DataFrame with complex types
  5. Concatenate columns in Apache Spark DataFrame
  6. Avoid performance impact of a single partition mode in Spark window functions
  7. Does SparkSQL support subquery?
  8. Including null values in an Apache Spark Join
  9. Spark sql queries vs dataframe functions
  10. Explode (transpose?) multiple columns in Spark SQL table
  11. How to get other columns when using Spark DataFrame groupby?
  12. get TopN of all groups after group by using Spark DataFrame
  13. Why using a UDF in a SQL query leads to cartesian product?
  14. pyspark: count distinct over a window
  15. What is the difference between cube, rollup and groupBy operators?
  16. Difference between === null and isNull in Spark DataDrame
  17. Spark functions vs UDF performance?
  18. Find maximum row per group in Spark DataFrame
  19. Spark Dataframe distinguish columns with duplicated name
  20. Can I do a max(count(*)) in SQL?
  21. SPARK SQL – case when then
  22. Why is Apache-Spark – Python so slow locally as compared to pandas?
  23. Spark load data and add filename as dataframe column
  24. PySpark create new column with mapping from a dict
  25. Using UDF ignores condition in when
  26. Total size of serialized results of 16 tasks (1048.5 MB) is bigger than spark.driver.maxResultSize (1024.0 MB)
  27. Multiple condition filter on dataframe
  28. How to return a “Tuple type” in a UDF in PySpark?
  29. How to connect HBase and Spark using Python?
  30. spark.ml StringIndexer throws ‘Unseen label’ on fit()

Leave a Comment