窗口上的 Spark 计算方差

Posted 2023-03-31

【中文标题】窗口上的 Spark 计算方差

【英文标题】：Spark computing variance over a window

【发布时间】：2021-01-04 10:30:14

【问题描述】：

我有一个结构如下的数据框：

+------------------+
|   id  |     value|
+------------------+
|  user0|     100  |
|  user1|     102  |
|  user0|     109  |
|  user2|     103  |
|  user1|     108  |
|  user0|     119  |
|  user0|     140  |
|  user0|     142  |
+------------------+

我想计算每个 ID 的每一行和前面的差异，为此我尝试了以下代码：

import pyspark.sql.functions as F

w_vv = Window.partitionBy('id')  
df=df.withColumn('variances',F.round(F.var_pop("value"),2).over(w_vv.rowsBetween(Window.unboundedPreceding,0)))

这是理想的期望输出

+--------------------------------------------------------------+
|   User|  value|                                     variances|
+--------------------------------------------------------------+
|  user0| value1|         -                                    |
|  user1| value1|         -                                    |
|  user0| value2|  variance(value2,value1)                     |
|  user1| value2|  variance(value2,value1)                     |
|  user1| value3|  variance(value3,value2,value1)              |
|  user0| value3|  variance(value4,value3,value2,value1)       |
|  user0| value4|  variance(value4,value3,value2,value1)       |
|  user0| value5|  variance(value5,value4,value3,value2,value1)|
+--------------------------------------------------------------+

前面的输出以数字为例：

+---------------------------+
|   User|  value|  variances|
+---------------------------+
|  user0| 2|         -      |
|  user1| 4|         -      |
|  user0| 3| 0.25           |
|  user1| 3| 0.25           |
|  user1| 9| 6.9            |
|  user0| 7| 4.7            |
|  user0| 3| 3.7            |     
|  user0| 4| 3              |
+---------------------------+

但是代码返回以下错误

grouping expressions sequence is empty, and '`timestamp`' is not an aggregate function.  
Wrap '(var_pop(CAST(`value` AS DOUBLE)) AS `_w0`)' in windowing function(s) or wrap  
'`timestamp`' in first() (or first_value) if you don't care which value you get.;;

我知道应该在 groupBy 上使用聚合函数，但我不知道如何对其进行编码以使其工作，有什么想法吗？谢谢

【问题讨论】：

你能添加数字结果期望吗？

@thebluephantom 确定，完成。

【参考方案1】：

您应该将窗口附加到var_pop，而不是round：

w_vv = Window.partitionBy('id')  

df = df.withColumn('variances',
    F.round(
        F.var_pop("value")
         .over(w_vv.rowsBetween(Window.unboundedPreceding,0)
    , 2)
)