Applying a Window function to calculate differences in pySpark

by

You can bring the previous day column by using lag function, and add additional column that does actual day-to-day return from the two columns, but you may have to tell spark how to partition your data and/or order it to do lag, something like this:

from pyspark.sql.window import Window
import pyspark.sql.functions as func
from pyspark.sql.functions import lit

dfu = df.withColumn('user', lit('tmoore'))

df_lag = dfu.withColumn('prev_day_price',
                        func.lag(dfu['price'])
                                 .over(Window.partitionBy("user")))

result = df_lag.withColumn('daily_return', 
          (df_lag['price'] - df_lag['prev_day_price']) / df_lag['price'] )

>>> result.show()
+---+-----+-------+--------------+--------------------+
|day|price|   user|prev_day_price|        daily_return|
+---+-----+-------+--------------+--------------------+
|  1| 33.3| tmoore|          null|                null|
|  2| 31.1| tmoore|          33.3|-0.07073954983922816|
|  3| 51.2| tmoore|          31.1|         0.392578125|
|  4| 21.3| tmoore|          51.2|  -1.403755868544601|
+---+-----+-------+--------------+--------------------+

Here is longer introduction into Window functions in Spark.

More Related Contents:

  1. TypeError: Column is not iterable – How to iterate over ArrayType()?
  2. Apache spark dealing with case statements
  3. Finding duplicates from large data set using Apache Spark
  4. How to make good reproducible Apache Spark examples
  5. Using a column value as a parameter to a spark DataFrame function
  6. Spark Window Functions – rangeBetween dates
  7. Spark SQL window function with complex condition
  8. Avoid performance impact of a single partition mode in Spark window functions
  9. How to improve performance for slow Spark jobs using DataFrame and JDBC connection?
  10. AttributeError: ‘DataFrame’ object has no attribute ‘map’
  11. Spark RDD to DataFrame python
  12. Python Spark Cumulative Sum by Group Using DataFrame
  13. How to save/insert each DStream into a permanent table
  14. Convert spark DataFrame column to python list
  15. Pyspark: Pass multiple columns in UDF
  16. Reading parquet files from multiple directories in Pyspark
  17. pyspark: count distinct over a window
  18. How to aggregate over rolling time window with groups in Spark
  19. pyspark: rolling average using timeseries data
  20. Manually create a pyspark dataframe
  21. combine text from multiple rows in pyspark
  22. Best way to get result count before LIMIT was applied
  23. Calculating Cumulative Sum in PostgreSQL
  24. Spark Dataframe validating column names for parquet writes
  25. Change nullable property of column in spark dataframe
  26. Why does Spark think this is a cross / Cartesian join
  27. Pyspark: Replacing value in a column by searching a dictionary
  28. Pivot String column on Pyspark Dataframe
  29. Reduce a key-value pair into a key-list pair with Apache Spark
  30. pyspark: Efficiently have partitionBy write to same number of total partitions as original table

Leave a Comment