用零填充缺失的销售价值并在 PySpark 中计算 3 个月的平均值答案

【问题标题】：Filling Missing sales value with zero and calculate 3 month average in PySpark用零填充缺失的销售价值并在 PySpark 中计算 3 个月的平均值
【发布时间】：2021-02-05 16:42:35
【问题描述】：

我想添加零销售额的缺失值并在 pyspark 中计算 3 个月平均值

My Input :
  product    specialty    date       sales
  A           pharma      1/3/2019    50
  A           pharma      1/4/2019    60
  A           pharma      1/5/2019    70
  A           pharma      1/8/2019    80
  A           ENT         1/8/2019    50
  A           ENT         1/9/2019    65
  A           ENT         1/11/2019   40

my output:
   product    specialty    date       sales    3month_avg_sales
   A           pharma      1/3/2019    50       16.67
   A           pharma      1/4/2019    60       36.67
   A           pharma      1/5/2019    70        60
   A           pharma      1/6/2019     0        43.33
   A           pharma      1/7/2019     0        23.33
   A           pharma      1/8/2019    80        26.67
   A           ENT         1/8/2019    50        16.67
   A           ENT         1/9/2019    65        38.33
   A           ENT         1/10/2019    0        38.33  
   A           ENT         1/11/2019   40        35
 



row = Row("Product", "specialty","Date", "Sales")
df = sc.parallelize([row("A","pharma", "1/3/2019", 50),row("A","pharma", "1/4/2019", 60),row("A", "pharma","01/05/2019", 70),row("A","pharma", "1/8/2019", 80),row("A","ENT", "1/8/2019", 50),row("A","ENT", "1/9/2019", 65),row("A","ENT", "1/11/2019", 40)]).toDF()
w = Window.partitionBy("product","specialty).orderBy("date")
df.withColumn("new_data_date", expr("add_months(data_date, 1)"))
df.withcolumn("sales",F.where(col("date") isin col("new_data_date")
 df=df.withColumn('index', (year('Date') - 2020) * 12 + month('Date')).withColumn('avg',sum('Sales').over(w) / 3)

我很惊讶地添加了销售价值为零的日期价值缺失的地方。并计算 3 个月的平均值。

【问题讨论】：

你的 spark 版本是什么？
Spark 2.4 版本
能否请您粘贴生成输出的代码，看来您发布的不是正确的

标签： date apache-spark pyspark missing-data

【解决方案1】：

对于缺失值，你可以这样做

df.fillna(0, subset=['sales'])

对于3个月的平均值，你可以找到一个很好的答案here，只是注意正确解析时间戳并将窗口开始日期更改为-90

更新

这段代码应该可以完成您寻找的工作

days = lambda i: i * 86400
w = (Window.orderBy(f.col("timestampGMT").cast('long')).rangeBetween(-days(90), 0))

missings_df = sparkSession.createDataFrame([ ('A', 'pharma', '1/6/2019', 0)], ['product', 'specialty', 'date', 'sales'])

df = (df
      .union(missings_df) # adding missing row
      .withColumn('timestampGMT', f.to_date('date', 'd/M/yyyy').cast('timestamp')) # cast to timestamp
      .withColumn('rolling_average', f.avg("sales").over(w)) # rolling average on 90 days
     )

【讨论】：

我想添加另一行输入中缺少日期的数据。在上面的输入中 1/6/2019 is missing 。我将获得产品 A、专业制药、日期为 2019 年 1 月 6 日且销售价值为零的数据。
然后检查这个答案：stackoverflow.com/a/52692858/7306659

【解决方案2】：

您可以使用 SparkSQL 内置函数transform + sequence 来创建缺失的月份并设置它们的销售额=0，使用窗口聚合函数计算所需的end_date 和最终的 3 个月平均销售额。下面我将代码分成三步进行说明，大家可以根据自己的需求进行合并。

注意：这假设每个不同月份最多有一条记录，并且所有日期值的 day=1，否则使用 @987654326 将日期截断到 month 级别@ 和/或定义重复条目的逻辑。

from pyspark.sql import functions as F, Window

df = spark.createDataFrame([
    ('A', 'pharma', '1/3/2019', 50), ('A', 'pharma', '1/4/2019', 60), 
    ('A', 'pharma', '1/5/2019', 70), ('A', 'pharma', '1/8/2019', 80), 
    ('A', 'ENT', '1/8/2019', 50), ('A', 'ENT', '1/9/2019', 65),
    ('A', 'ENT', '1/11/2019', 40)
], ['product', 'specialty', 'date', 'sales'])

df = df.withColumn('date', F.to_date('date', 'd/M/yyyy'))

Step-1： 设置 WinSpec w1 并使用窗口聚合函数 lead 查找下一个日期（w1），将其转换为前几个月以设置日期序列：

w1 = Window.partitionBy('product', 'specialty').orderBy('date')

df1 = df.withColumn('end_date', F.coalesce(F.add_months(F.lead('date').over(w1),-1),'date'))
+-------+---------+----------+-----+----------+
|product|specialty|      date|sales|  end_date|
+-------+---------+----------+-----+----------+
|      A|      ENT|2019-08-01|   50|2019-08-01|
|      A|      ENT|2019-09-01|   65|2019-10-01|
|      A|      ENT|2019-11-01|   40|2019-11-01|
|      A|   pharma|2019-03-01|   50|2019-03-01|
|      A|   pharma|2019-04-01|   60|2019-04-01|
|      A|   pharma|2019-05-01|   70|2019-07-01|
|      A|   pharma|2019-08-01|   80|2019-08-01|
+-------+---------+----------+-----+----------+

步骤 2： 使用months_between(end_date, date) 计算两个日期之间的月数，并使用转换函数遍历sequence(0, #months)，创建一个named_struct，其中date=add_months(date,i) 和sales= IF(i=0,sales,0)，使用inline_outer 来分解结构数组：

df2 = df1.selectExpr("product", "specialty", """
       inline_outer(
         transform(
            sequence(0,int(months_between(end_date, date))),
            i -> (add_months(date,i) as date, IF(i=0,sales,0) as sales)
         )
       )
   """)
+-------+---------+----------+-----+
|product|specialty|      date|sales|
+-------+---------+----------+-----+
|      A|      ENT|2019-08-01|   50|
|      A|      ENT|2019-09-01|   65|
|      A|      ENT|2019-10-01|    0|
|      A|      ENT|2019-11-01|   40|
|      A|   pharma|2019-03-01|   50|
|      A|   pharma|2019-04-01|   60|
|      A|   pharma|2019-05-01|   70|
|      A|   pharma|2019-06-01|    0|
|      A|   pharma|2019-07-01|    0|
|      A|   pharma|2019-08-01|   80|
+-------+---------+----------+-----+

第 3 步：使用以下 WinSpec w2 和聚合函数计算平均值：

N = 3

w2 = Window.partitionBy('product', 'specialty').orderBy('date').rowsBetween(-N+1,0)

df_new = df2.select("*", F.round(F.sum('sales').over(w2)/N,2).alias(f'{N}month_avg_sales'))
+-------+---------+----------+-----+----------------+
|product|specialty|      date|sales|3month_avg_sales|
+-------+---------+----------+-----+----------------+
|      A|      ENT|2019-08-01|   50|           16.67|
|      A|      ENT|2019-09-01|   65|           38.33|
|      A|      ENT|2019-10-01|    0|           38.33|
|      A|      ENT|2019-11-01|   40|            35.0|
|      A|   pharma|2019-03-01|   50|           16.67|
|      A|   pharma|2019-04-01|   60|           36.67|
|      A|   pharma|2019-05-01|   70|            60.0|
|      A|   pharma|2019-06-01|    0|           43.33|
|      A|   pharma|2019-07-01|    0|           23.33|
|      A|   pharma|2019-08-01|   80|           26.67|
+-------+---------+----------+-----+----------------+

【讨论】：