Spark Window Functions – rangeBetween dates

by

Spark >= 2.3

Since Spark 2.3 it is possible to use interval objects using SQL API, but the DataFrame API support is still work in progress.

df.createOrReplaceTempView("df")

spark.sql(
    """SELECT *, mean(some_value) OVER (
        PARTITION BY id 
        ORDER BY CAST(start AS timestamp) 
        RANGE BETWEEN INTERVAL 7 DAYS PRECEDING AND CURRENT ROW
     ) AS mean FROM df""").show()

## +---+----------+----------+------------------+       
## | id|     start|some_value|              mean|
## +---+----------+----------+------------------+
## |  1|2015-01-01|      20.0|              20.0|
## |  1|2015-01-06|      10.0|              15.0|
## |  1|2015-01-07|      25.0|18.333333333333332|
## |  1|2015-01-12|      30.0|21.666666666666668|
## |  2|2015-01-01|       5.0|               5.0|
## |  2|2015-01-03|      30.0|              17.5|
## |  2|2015-02-01|      20.0|              20.0|
## +---+----------+----------+------------------+

Spark < 2.3

As far as I know it is not possible directly neither in Spark nor Hive. Both require ORDER BY clause used with RANGE to be numeric. The closest thing I found is conversion to timestamp and operating on seconds. Assuming start column contains date type:

from pyspark.sql import Row

row = Row("id", "start", "some_value")
df = sc.parallelize([
    row(1, "2015-01-01", 20.0),
    row(1, "2015-01-06", 10.0),
    row(1, "2015-01-07", 25.0),
    row(1, "2015-01-12", 30.0),
    row(2, "2015-01-01", 5.0),
    row(2, "2015-01-03", 30.0),
    row(2, "2015-02-01", 20.0)
]).toDF().withColumn("start", col("start").cast("date"))

A small helper and window definition:

from pyspark.sql.window import Window
from pyspark.sql.functions import mean, col


# Hive timestamp is interpreted as UNIX timestamp in seconds*
days = lambda i: i * 86400

Finally query:

w = (Window()
   .partitionBy(col("id"))
   .orderBy(col("start").cast("timestamp").cast("long"))
   .rangeBetween(-days(7), 0))

df.select(col("*"), mean("some_value").over(w).alias("mean")).show()

## +---+----------+----------+------------------+
## | id|     start|some_value|              mean|
## +---+----------+----------+------------------+
## |  1|2015-01-01|      20.0|              20.0|
## |  1|2015-01-06|      10.0|              15.0|
## |  1|2015-01-07|      25.0|18.333333333333332|
## |  1|2015-01-12|      30.0|21.666666666666668|
## |  2|2015-01-01|       5.0|               5.0|
## |  2|2015-01-03|      30.0|              17.5|
## |  2|2015-02-01|      20.0|              20.0|
## +---+----------+----------+------------------+

Far from pretty but works.

* Hive Language Manual, Types

More Related Contents:

  1. Spark SQL window function with complex condition
  2. How to aggregate over rolling time window with groups in Spark
  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. How to flatten a struct in a Spark dataframe?
  20. How to access element of a VectorUDT column in a Spark DataFrame?
  21. Pyspark ‘NoneType’ object has no attribute ‘_jvm’ error
  22. OVER clause in Oracle
  23. Reading csv files with quoted fields containing embedded commas
  24. Create Spark DataFrame. Can not infer schema for type
  25. spark dataframe drop duplicates and keep first
  26. Referencing current row in FILTER clause of window function
  27. PySpark: How to fillna values in dataframe for specific columns?
  28. When to cache a DataFrame?
  29. How to use a ring data structure in window functions
  30. PySpark error: AttributeError: ‘NoneType’ object has no attribute ‘_jvm’

Leave a Comment