pyspark: rolling average using timeseries data

by

I figured out the correct way to calculate a moving/rolling average using this stackoverflow:

Spark Window Functions – rangeBetween dates

The basic idea is to convert your timestamp column to seconds, and then you can use the rangeBetween function in the pyspark.sql.Window class to include the correct rows in your window.

Here’s the solved example:

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


#function to calculate number of seconds from number of days
days = lambda i: i * 86400

df = spark.createDataFrame([(17, "2017-03-10T15:27:18+00:00"),
                        (13, "2017-03-15T12:27:18+00:00"),
                        (25, "2017-03-18T11:27:18+00:00")],
                        ["dollars", "timestampGMT"])
df = df.withColumn('timestampGMT', df.timestampGMT.cast('timestamp'))

#create window by casting timestamp to long (number of seconds)
w = (Window.orderBy(F.col("timestampGMT").cast('long')).rangeBetween(-days(7), 0))

df = df.withColumn('rolling_average', F.avg("dollars").over(w))

This results in the exact column of rolling averages that I was looking for:

dollars   timestampGMT            rolling_average
17        2017-03-10 15:27:18.0   17.0
13        2017-03-15 12:27:18.0   15.0
25        2017-03-18 11:27:18.0   19.0

More Related Contents:

  1. Avoid performance impact of a single partition mode in Spark window functions
  2. pyspark: count distinct over a window
  3. How to melt Spark DataFrame?
  4. Unpivot in spark-sql/pyspark
  5. Spark Window Functions – rangeBetween dates
  6. Spark SQL window function with complex condition
  7. Spark Dataframe validating column names for parquet writes
  8. How to check if spark dataframe is empty?
  9. How to improve performance for slow Spark jobs using DataFrame and JDBC connection?
  10. TypeError: Column is not iterable – How to iterate over ArrayType()?
  11. How to save/insert each DStream into a permanent table
  12. How to fix ‘TypeError: an integer is required (got type bytes)’ error when trying to run pyspark after installing spark 2.4.4
  13. Pyspark : forward fill with last observation for a DataFrame
  14. Adding a group count column to a PySpark dataframe
  15. Multiple Spark applications with HiveContext
  16. How to loop through each row of dataFrame in pyspark
  17. Why does Spark think this is a cross / Cartesian join
  18. How to find count of Null and Nan values for each column in a PySpark dataframe efficiently?
  19. Pyspark: Pass multiple columns in UDF
  20. Rename more than one column using withColumnRenamed
  21. pyspark: Efficiently have partitionBy write to same number of total partitions as original table
  22. Spark gives a StackOverflowError when training using ALS
  23. How to aggregate over rolling time window with groups in Spark
  24. AWS EMR – ModuleNotFoundError: No module named ‘pyarrow’
  25. How can I access python variable in Spark SQL?
  26. How to turn off scientific notation in pyspark?
  27. Spark ALS predictAll returns empty
  28. Why is the fold action necessary in Spark?
  29. How to add a SparkListener from pySpark in Python?
  30. PySpark error: AttributeError: ‘NoneType’ object has no attribute ‘_jvm’

Leave a Comment