与前一天 pyspark 相比具有平均值的列答案

【问题标题】：Column with avg over previous day pyspark与前一天 pyspark 相比具有平均值的列
【发布时间】：2019-02-11 21:42:34
【问题描述】：

我想使用 pyspark 创建一个新列，它是前一天销售额的平均值。

考虑这些值在不同的时间戳。

例如转换这个：

| Date       | value |
|------------|-------|
| 2019/02/11 | 30    |
| 2019/02/11 | 40    |
| 2019/02/11 | 20    |
| 2019/02/12 | 10    |
| 2019/02/12 | 15    |

到这里

| Date       | value | avg  |
|------------|-------|------|
| 2019/02/11 | 30    | null |
| 2019/02/11 | 40    | null |
| 2019/02/11 | 20    | null |
| 2019/02/12 | 10    | 30   |
| 2019/02/12 | 15    | 30   |

我的想法：

使用过滤器和聚合函数来获取平均值但它的抛出错误。不知道我哪里做错了。

df = df.withColumn("avg",lit((df.filter(df["date"] == date_sub("date",1)).agg({"value": "avg"}))))

【问题讨论】：

你也可以看看这个答案：stackoverflow.com/questions/47622447/…

标签： python apache-spark dataframe pyspark apache-spark-sql

【解决方案1】：

您可以使用 windows 函数来执行此操作，但您必须创建一个新列来处理日期。我为您添加了几行示例：

df.withColumn(
  "rnk",
  F.dense_rank().over(Window.partitionBy().orderBy("date"))
).withColumn(
  "avg",
  F.avg("value").over(Window.partitionBy().orderBy("rnk").rangeBetween(-1,-1))
).show()

+----------+-----+---+----+
|      date|value|rnk| avg|
+----------+-----+---+----+
|2018-01-01|   20|  1|null|
|2018-01-01|   30|  1|null|
|2018-01-01|   40|  1|null|
|2018-01-02|   40|  2|30.0|
|2018-01-02|   30|  2|30.0|
|2018-01-03|   40|  3|35.0|
|2018-01-03|   40|  3|35.0|
+----------+-----+---+----+

您也可以使用聚合来做到这一点：

agg_df = df.withColumn("date", F.date_add("date", 1)).groupBy('date').avg("value")
df.join(agg_df, how="full_outer", on="date").orderBy("date").show()

+----------+-----+----------+
|      date|value|avg(value)|
+----------+-----+----------+
|2018-01-01|   20|      null|
|2018-01-01|   30|      null|
|2018-01-01|   40|      null|
|2018-01-02|   30|      30.0|
|2018-01-02|   40|      30.0|
|2018-01-03|   40|      35.0|
|2018-01-03|   40|      35.0|
|2018-01-04| null|      40.0|
+----------+-----+----------+

【讨论】：

rangeBetween(-1,-1) 为什么要使用这个，你能解释一下吗？
当然。它采用当前行中“rnk”中的值并添加-1（基本上删除1）。因此，当您查看 rnk = 2 的行时，它只需要值为 1 的行。

【解决方案2】：

第 0 步：创建数据帧

from pyspark.sql.window import Window
from pyspark.sql.functions import col, avg, lag
df = sqlContext.createDataFrame([('2019/02/11',30),('2019/02/11',40),('2019/02/11',20),
                                 ('2019/02/12',10),('2019/02/12',15),
                                 ('2019/02/13',10),('2019/02/13',20)],['Date','value'])

第一步：先计算平均值，然后使用windows函数得到滞后1天。

my_window = Window.partitionBy().orderBy('Date')
df_avg_previous = df.groupBy('Date').agg(avg(col('value')).alias('avg'))
df_avg_previous = df_avg_previous.withColumn('avg', lag(col('avg'),1).over(my_window))
df_avg_previous.show()
+----------+----+
|      Date| avg|
+----------+----+
|2019/02/11|null|
|2019/02/12|30.0|
|2019/02/13|12.5|
+----------+----+

第 2 步：最后使用 left 连接来连接两个数据帧。

df = df.join(df_avg_previous, ['Date'],how='left').orderBy('Date')
df.show()
+----------+-----+----+
|      Date|value| avg|
+----------+-----+----+
|2019/02/11|   40|null|
|2019/02/11|   20|null|
|2019/02/11|   30|null|
|2019/02/12|   10|30.0|
|2019/02/12|   15|30.0|
|2019/02/13|   10|12.5|
|2019/02/13|   20|12.5|
+----------+-----+----+

【讨论】：