如何使用 Window() 在 PySpark 中计算滚动中位数? [英] How to calculate rolling median in PySpark using Window()?

问题描述

如何计算前 3 个值的窗口大小的美元滚动中位数?

How do I calculate rolling median of dollar for a window size of previous 3 values?

输入数据

dollars timestampGMT       
25      2017-03-18 11:27:18
17      2017-03-18 11:27:19
13      2017-03-18 11:27:20
27      2017-03-18 11:27:21
13      2017-03-18 11:27:22
43      2017-03-18 11:27:23
12      2017-03-18 11:27:24

预期输出数据

dollars timestampGMT          rolling_median_dollar
25      2017-03-18 11:27:18   median(25)
17      2017-03-18 11:27:19   median(17,25)
13      2017-03-18 11:27:20   median(13,17,25)
27      2017-03-18 11:27:21   median(27,13,17)
13      2017-03-18 11:27:22   median(13,27,13)
43      2017-03-18 11:27:23   median(43,13,27)
12      2017-03-18 11:27:24   median(12,43,13)

下面的代码确实移动了平均值，但 PySpark 没有 F.median().

Below code does moving avg but PySpark doesn't have F.median().

pyspark:使用时间序列数据的滚动平均值

编辑 1:挑战是中值()函数不退出.我做不到

EDIT 1: The challenge is median() function doesn't exit. I cannot do

df = df.withColumn('rolling_average', F.median("dollars").over(w))

如果我想要移动平均线，我可以做到

If I wanted moving average I could have done

df = df.withColumn('rolling_average', F.avg("dollars").over(w))

编辑 2:尝试使用 approxQuantile()

EDIT 2: Tried using approxQuantile()

windfun = Window().partitionBy().orderBy(F.col(date_column)).rowsBetwe‌​en(-3, 0) sdf.withColumn("movingMedian", sdf.approxQuantile(col='a', probabilities=[0.5], relativeError=0.00001).over(windfun)) 

但出现错误

AttributeError: 'list' object has no attribute 'over'

编辑 3

请给出没有 Udf 的解决方案，因为它不会从催化剂优化中受益.

Please give solution without Udf since it won't benefit from catalyst optimization.

推荐答案

一种方法是将 $dollars 列收集为每个窗口的列表，然后使用udf:

One way is to collect the $dollars column as a list per window, and then calculate the median of the resulting lists using an udf:

from pyspark.sql.window import Window
from pyspark.sql.functions import *
import numpy as np 
from pyspark.sql.types import FloatType

w = (Window.orderBy(col("timestampGMT").cast('long')).rangeBetween(-2, 0))
median_udf = udf(lambda x: float(np.median(x)), FloatType())

df.withColumn("list", collect_list("dollars").over(w)) \
  .withColumn("rolling_median", median_udf("list")).show(truncate = False)

+-------+---------------------+------------+--------------+
|dollars|timestampGMT         |list        |rolling_median|
+-------+---------------------+------------+--------------+
|25     |2017-03-18 11:27:18.0|[25]        |25.0          |
|17     |2017-03-18 11:27:19.0|[25, 17]    |21.0          |
|13     |2017-03-18 11:27:20.0|[25, 17, 13]|17.0          |
|27     |2017-03-18 11:27:21.0|[17, 13, 27]|17.0          |
|13     |2017-03-18 11:27:22.0|[13, 27, 13]|13.0          |
|43     |2017-03-18 11:27:23.0|[27, 13, 43]|27.0          |
|12     |2017-03-18 11:27:24.0|[13, 43, 12]|13.0          |
+-------+---------------------+------------+--------------+

这篇关于如何使用 Window() 在 PySpark 中计算滚动中位数?的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！