Solución:
Puede traer la columna del día anterior usando la función de retraso y agregar una columna adicional que haga un retorno real del día a día de las dos columnas, pero es posible que tenga que decirle a Spark cómo particionar sus datos y / o ordenar que haga un retraso. , algo como esto:
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|
+---+-----+-------+--------------+--------------------+
Aquí hay una introducción más extensa a las funciones de Windows en Spark.
Retraso La función puede ayudarlo a resolver su caso de uso.
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'] )
¡Haz clic para puntuar esta entrada!
(Votos: 0 Promedio: 0)