Using the Window function to calculate differences in pySpark

Question

I use pySpark and set up my data frame with two columns representing the daily price of assets as follows:

 ind = sc.parallelize(range(1,5)) prices = sc.parallelize([33.3,31.1,51.2,21.3]) data = ind.zip(prices) df = sqlCtx.createDataFrame(data,["day","price"]) 

I get after applying df.show() :

 +---+-----+ |day|price| +---+-----+ | 1| 33.3| | 2| 31.1| | 3| 51.2| | 4| 21.3| +---+-----+ 

Which is good and all. I would like to have another column that contains daily returns of the price column, i.e. something like

(price(day2)-price(day1))/(price(day1))

After much research, I was told that this is most effectively achieved by using the pyspark.sql.window functions, but I cannot figure out how to do this.

Answer 1

You can transfer the column of the previous day using the delay function and add an additional column that performs the actual daily return of the two columns, but you may have to tell the pair how to separate your data and / or order a delay., Something like of 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's a longer introduction to window features in Spark .

Answer 2

The delay function can help you decide your use case.

 from pyspark.sql.window import Window import pyspark.sql.functions as func ### Defining the window Windowspec=Window.orderBy("day") ### Calculating lag of price at each day level prev_day_price= df.withColumn('prev_day_price', func.lag(dfu['price']) .over(Windowspec)) ### Calculating the average result = prev_day_price.withColumn('daily_return', (prev_day_price['price'] - prev_day_price['prev_day_price']) / prev_day_price['price'] )