如何使用pyspark数据框查找std dev分区或分组数据？

Question

代码：

w = Window().partitionBy("ticker").orderBy("date") 
x = s_df.withColumn("daily_return", (col("close") - lag("close", 1).over(w))/lag("close", 1).over(w)) 

什么s_df样子：

+----------+------+------+------+------+--------+------+ 
|  date| open| high| low| close| volume|ticker| 
+----------+------+------+------+------+--------+------+ 
|2016-11-02| 111.4|112.35|111.23|111.59|28331709| AAPL| 
|2016-11-01|113.46|113.77|110.53|111.49|43825812| AAPL| 
|2016-10-31|113.65|114.23| 113.2|113.54|26419398| AAPL| 
+----------+------+------+------+------+--------+------+ 

看起来什么X像当年：

+----------+--------------------+ 
|  date| avg(daily_return)| 
+----------+--------------------+ 
|2015-12-28|0.004124786535090563| 
|2015-11-20|0.006992226387807268| 
|2015-12-29| 0.01730500286123971| 

我想找到AVG（daily_return）的标准偏差为每一组代收人。

我的尝试：

x.agg(stddev("avg(daily_return)")).over(w) 

我得到这个错误：

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

是什么，我试图做不可能的，或者是有另一种方式做到这一点？

Answer 1

ticker比data不同的维度，这样就不能要这两个聚合在一起。您可以：因为你只有一个股票date，ticker组和标准偏差是不确定的

s_df_w_daily_rets = s_df.withColumn("daily_return", 
    (col("close") - lag("close", 1).over(w))/lag("close", 1).over(w)) 

s_df_w_daily_rets.groupBy("date").agg(avg("daily_return")) 
s_df_w_daily_rets.groupBy("ticker").agg(stddev("daily_return")) 

groupBy("date","ticker").agg(..)没有意义。

Answer 2

您可以

x.groupBy("A").agg(stddev("D"))