Zastosowanie funkcji okna do obliczenia różnic w pySpark

Question

używam pySpark i utworzyły mój dataframe z dwóch kolumn reprezentujących codziennie cen aktywów w następujący sposób:

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"]) 

uzyskać po zastosowaniu df.show():

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

Co jest w porządku i wszystko. Chciałbym mieć inną kolumnę zawierającą dnia na dzień powrotów kolumnie cenowej, czyli coś podobnego

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

Po wielu badań, jak mi powiedziano, że jest to najbardziej skutecznie osiągnięte przez nałożeniem pyspark.sql.window funkcji, ale nie jestem w stanie zobaczyć, jak.

Answer 1

Możesz wstawić kolumnę z poprzedniego dnia, używając funkcji lag, i dodać dodatkową kolumnę, która odzwierciedla rzeczywisty powrót z dwóch kolumn, ale może być konieczne podanie informacji o sposobie podzielenia danych i/lub zamówienia na partycje to zrobić lagi, coś takiego:

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| 
+---+-----+-------+--------------+--------------------+ 

Oto już wprowadzenie do Window functions in Spark.