从Spark中，

val w = Window.partitionBy("item","loc").orderBy("month").rangeBetween(-2, -1)
df.withColumn("month", 'month.cast("int"))
  .withColumn("avg", avg(when('qty =!= lit(0), 'qty)).over(w)).show()

+-----+-----+-----+----+---+----+
| item|  loc|month|year|qty| avg|
+-----+-----+-----+----+---+----+
|watch|delhi|    1|2020| 10| 0.0|
|watch|delhi|    2|2020|  0|10.0|
|watch|delhi|    3|2020| 20|10.0|
|watch|delhi|    4|2020| 30|20.0|
|watch|delhi|    5|2020| 40|25.0|
|watch|delhi|    6|2020| 50|35.0|
+-----+-----+-----+----+---+----+

我认为这是一个有条件的印支平均数：

select 
    t.*,
    coalesce(avg(nullif(qty, 0)) over(partition by item, loc order by month), 0) qty_avg
from mytable t
``` `nullif()` 产量 `null` 为了 `0` 值-哪个 `avg()` 然后忽略。我把整个Windows都包上了 `coalesce()` ，因为你似乎想要 `0` 如果有 `null` 仅限值。

它可以在spark中使用lag函数和windowframe来完成

import org.apache.spark.sql.expressions.Window
import org.apache.spark.sql.functions._
import org.apache.spark.sql.types.IntegerType

df.withColumn("month", col("month").cast(IntegerType))
.withColumn("avg", when(lag("qty", 2, 0).over(w) =!= lit(0) && lag("qty", 1, 0).over(w) =!= lit(0),
(lag("qty", 2, 0).over(w) + lag("qty", 1, 0).over(w)).divide(lit(2)))
.when(lag("qty", 1, 0).over(w) =!= lit(0),lag("qty", 1, 0).over(w)).otherwise(lag("qty", 2, 0)
.over(w))).show()

输出：

+-----+-----+-----+----+---+----+
| item|  loc|month|year|qty| avg|
+-----+-----+-----+----+---+----+
|watch|delhi|    1|2020| 10|   0|
|watch|delhi|    2|2020|  0|  10|
|watch|delhi|    3|2020| 20|  10|
|watch|delhi|    4|2020| 30|  20|
|watch|delhi|    5|2020| 40|25.0|
|watch|delhi|    6|2020| 50|35.0|
+-----+-----+-----+----+---+----+

在sql中，可以将窗口函数与窗口框架说明符一起使用：

select t.*,
       coalesce(avg(nullif(qty, 0)) over (partition by item, loc
                                          order by month
                                          rows between 2 preceding and 1 preceding
                                         ),
                0) as qty_avg
from t;