Spark DataFrame如何将列中0值填充为最近的前置非零值

核心思路 #

要实现按ID顺序，将count列的0值替换为同列靠前位置最近的非零值，直接利用Spark内置窗口函数即可实现，不需要额外开发UDF，执行性能更高：

1. 先将count列中值为0的记录转换为null——Spark内置的last函数支持忽略空值的特性，但无法直接识别0值，所以先做值替换
2. 定义按ID升序排列的窗口，窗口范围覆盖从数据集起点到当前行的所有记录
3. 调用带ignorenulls=True参数的last函数，取窗口内最近的非空值填充当前行，即可得到目标结果

代码实现（PySpark版本） #

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

# 初始化Spark会话
spark = SparkSession.builder.appName("fill_zero").getOrCreate()

# 构造示例原始数据（可替换为你自己的DataFrame读取逻辑）
source_data = [
    (0, 12000), (1, 12000), (2, 12000), (3, 12000),
    (4, 0), (5, 0), (6, 0), (7, 0),
    (8, 1400), (9, 1400)
]
df = spark.createDataFrame(source_data, schema=["ID", "count"])

# 定义窗口规则：按ID升序，范围为从首行到当前行
fill_window = Window.orderBy("ID").rowsBetween(Window.unboundedPreceding, Window.currentRow)

# 执行填充逻辑
result_df = df.withColumn(
    "count",
    F.last(
        F.when(F.col("count") != 0, F.col("count")),
        ignorenulls=True
    ).over(fill_window)
)

# 打印结果验证
result_df.show()

注意事项 #

• 必须保证窗口按ID（或你业务上定义先后顺序的字段）排序，否则会因为数据顺序错乱导致填充结果错误
• 如果需要按指定维度分组填充（比如不同品类、不同用户下分别做前向填充），只需要在窗口定义时加上.partitionBy("分组列名")即可
• Scala/Spark SQL版本的实现逻辑完全一致，仅语法有少量区别，核心都是「0值转null + 支持忽略空值的last窗口计算」

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