PySpark实现固定3行简单移动平均（SMA）并返回空值

要实现仅当窗口包含完整3行时计算简单移动平均，否则返回NULL，可以通过窗口函数结合条件判断来处理，以下是几种可行方案：

方案一：SQL方式（基于窗口内计数判断） #

修改原SQL语句，新增窗口内的计数逻辑，仅当计数等于3时返回平均值：

select 
    case 
        when count(v1) over (order by id1 rows between 2 preceding and current row) = 3 
        then avg(v1) over (order by id1 rows between 2 preceding and current row) 
        else null 
    end as v1 
from x

执行后结果：

[Row(v1=None), Row(v1=None), Row(v1=10.0), Row(v1=15.0)]

方案二：SQL方式（基于行号判断） #

利用row_number()获取当前行的顺序号，当行号≥3时（即存在前2行数据）再计算平均值：

select 
    case 
        when row_number() over (order by id1) >= 3 
        then avg(v1) over (order by id1 rows between 2 preceding and current row) 
        else null 
    end as v1 
from x

方案三：DataFrame API方式 #

如果习惯用DataFrame操作，写法如下：

from pyspark.sql.window import Window
from pyspark.sql import functions as F

# 定义3行窗口（前2行+当前行）
window_spec = Window.orderBy("id1").rowsBetween(-2, 0)
# 定义行号窗口
row_num_window = Window.orderBy("id1")

df_result = df.withColumn(
    "v1",
    F.when(
        F.row_number().over(row_num_window) >= 3,
        F.avg("v1").over(window_spec)
    ).otherwise(F.lit(None))
)

df_result.show()

输出结果：

+---+----+
|id1|  v1|
+---+----+
|  1|null|
|  2|null|
|  3|10.0|
|  4|15.0|
+---+----+

原理说明 #

两种核心思路本质一致：

• 要么统计当前窗口内的记录数量，确认达到3条后计算平均
• 要么通过行号判断当前行是否处于第3行及以后，确保存在足够的前置数据

这样就能避免窗口数据不足时计算部分平均，返回符合预期的NULL值。

内容的提问来源于stack exchange，提问作者jangorecki