Áp dụng một hàm cửa sổ để tính toán sự khác biệt trong pySpark

Question

Tôi đang sử dụng pySpark, và đã thiết lập dataframe của tôi với hai cột đại diện cho một giá tài sản hàng ngày như sau:

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

tôi nhận được khi áp dụng df.show():

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

Điều đó là tốt và tất cả. Tôi muốn có một cột chứa ngày-to-ngày trở lại của cột giá, tức là một cái gì đó giống như

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

Sau nhiều nghiên cứu, tôi nói rằng điều này được thực hiện một cách hiệu quả nhất thông qua việc áp dụng pyspark.sql.window chức năng, nhưng tôi không thể xem như thế nào.

Answer 1

Bạn có thể đưa cột ngày trước bằng cách sử dụng chức năng lag và thêm cột bổ sung thực sự trả lại hàng ngày từ hai cột, nhưng bạn có thể phải biết cách phân chia dữ liệu và/hoặc thứ tự của bạn nó để làm lag, một cái gì đó như thế này:

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

Đây là giới thiệu dài hơn vào Window functions in Spark.