pyspark：使用自定义时间序列数据的滚动平均值

Question

你好我的基础数据框是这样的。

'|stockId|timeStamp|stockPrice|'
+-------+---------+----------+
|    101|        1|      53.0|
|    101|        2|      15.0|
|    101|        3|      57.0|
|    101|        4|      71.0|
|    101|        5|      86.0|

这是我的代码，它通过 window 和平均 window 转换 days.followed 。

days=lambda i:i*86400
W=Window.partitionBy(F.col('stockId')).orderBy(F.col('epoch_time').cast("timestamp").cast("long")).rangeBetween(-days(3),0)
Df=.withColumn("current_timestamp",F.unix_timestamp(F.lit(timestamp),'yyyy-MM-dd HH:mm:ss').cast("timestamp"))\
.withColumn("epoch",F.unix_timestamp("current_timestamp"))\
.withColumn("epoch_time",F.concat(F.col("epoch")+F.col("timeStamp")))\
.withColumn("moving_avg",F.avg("stockPrice").over(W))

这是我的结果。

+-------+---------+----------+-------------------+----------+----------+-----------------+
|stockId|timeStamp|stockPrice|  current_timestamp|     epoch|epoch_time|       moving_avg|
+-------+---------+----------+-------------------+----------+----------+-----------------+
|    101|        1|      53.0|2020-01-21 10:53:43|1579584223|1579584224|48.21782178217822|
|    101|        2|      15.0|2020-01-21 10:53:43|1579584223|1579584225|48.21782178217822|
|    101|        3|      57.0|2020-01-21 10:53:43|1579584223|1579584226|48.21782178217822|
|    101|        4|      86.0|2020-01-21 10:53:43|1579584223|1579584227|48.21782178217822|

预期输出

+-------+---------+----------+-------------------+----------+----------+-----------------+
|stockId|timeStamp|stockPrice|  current_timestamp|     epoch|epoch_time|       moving_avg|
+-------+---------+----------+-------------------+----------+----------+-----------------+
|    101|        3|      57.0|2020-01-21 10:53:43|1579584223|1579584226|41.67|
|    101|        4|      71.0|2020-01-21 10:53:43|1579584223|1579584227|47.67|
|    101|        5|      71.0|2020-01-21 10:53:43|1579584223|1579584227|71.33|

Answer 1

W=Window.partitionBy(F.col('stockId')).orderBy(F.col('epoch_time').cast("timestamp").cast("long")).rangeBetween(-2,0)

+-------+---------+----------+-------------------+----------+----------+------
|stockId|timeStamp|stockPrice|  current_timestamp|     epoch|epoch_time|       moving_avg|
+-------+---------+----------+-------------------+----------+----------+-----------------+
|    101|        3|      57.0|2020-01-21 10:53:43|1579584223|1579584226|41.67|
|    101|        4|      71.0|2020-01-21 10:53:43|1579584223|1579584227|47.67|
|    101|        5|      71.0|2020-01-21 10:53:43|1579584223|1579584227|71.33|

保持滚动后window rangeBetween(-2,0) 平均值从 current row 变为 3rd row of the table。