如何得到每种产品的最小库存和最大库存之间的差额？

Question

我想得到每个产品的最大库存和最小库存，并计算这些值之间的差异。如果股票等于null或为空，则应将其替换为0。

这是我的密码：

import org.apache.spark.sql.functions._
import org.apache.spark.sql.expressions._

val windowSpec = Window.partitionBy("product_pk")
val sales = df
             .select($"product_pk",$"stock")
             .withColumn("sold",max($"stock")-min($"stock").over(windowSpec))

但是当我运行我的代码时，我会得到这个错误。我不明白为什么不能将Window与product_pk结合使用。

诊断: User类抛出异常: org.apache.spark.sql.AnalysisException:分组表达式序列为空，“mytable.product_pk”不是聚合函数。在窗口函数中将'(max(mytable.stock)包装为_w0‘)，或者在first() (或first_value)中包装'mytable.product_pk’，如果您不关心得到哪个值。

还是我应该使用groupBy product_pk？

Answer 1

当前，您在错误的上下文中使用max (未指定窗口)，请尝试：

val sales = df
  .select($"product_pk",$"stock")
  .withColumn("sold",max($"stock").over(windowSpec)-min($"stock").over(windowSpec))

您还可以使用groupBy：

val sales = df
  .groupBy($"product_pk",$"stock")
    .agg(
      max($"stock").as("max_stock"),
      min($"stock").as("min_stock")
    )
  .withColumn("sold",coalesce($"max_stock"-$"min_stock",lit(0.0)))