PySpark如何将列中null值替换为同列上方最近非空值

该需求属于典型的**向前空值填充（Forward Fill）**场景，即保留列中已有的非空值，将空值替换为同列上方最近的非空值，全程使用Spark原生函数实现，无自定义UDF性能损耗。

核心注意前提 #

Spark是分布式计算引擎，数据读取后无默认行顺序保证，必须先生成和原始CSV行序一致的自增ID作为排序依据，否则会出现行序错乱、填充结果错误的问题。

完整实现代码 #

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, last, monotonically_increasing_id
from pyspark.sql.window import Window

# 初始化Spark会话（实际使用时按自身环境配置参数）
spark = SparkSession.builder.appName("country_col_fill").getOrCreate()

# 读取CSV数据源，替换为你的实际文件路径
# df = spark.read.csv("your_file_path.csv", header=True, inferSchema=True)

# 以下为样例数据构造，实际使用时替换为上方的CSV读取代码
sample_data = [
    ("BR", 1995), (None, 1997), (None, 1999), (None, 2001), (None, 2003), (None, 2005),
    ("NO", 1995), (None, 1997), (None, 1999), (None, 2001), (None, 2003), (None, 2005),
    ("RO", 1995), (None, 1997), (None, 1999), (None, 2001), (None, 2003), (None, 2005),
    ("AC", 1995), (None, 1997)
]
df = spark.createDataFrame(sample_data, schema=["C", "Y"])

# 步骤1：生成和原始读取顺序一致的单调自增行ID，作为排序依据
df = df.withColumn("row_seq", monotonically_increasing_id())

# 步骤2：定义填充窗口：按行序升序，窗口范围覆盖从第一行到当前行
fill_window = Window.orderBy("row_seq").rowsBetween(Window.unboundedPreceding, Window.currentRow)

# 步骤3：执行填充，取窗口内最后一个非空的C值覆盖空值，最后删除辅助行ID列
filled_result = df.withColumn(
    "C",
    last(col("C"), ignorenulls=True).over(fill_window)
).drop("row_seq")

# 输出验证结果
filled_result.show()

实现逻辑说明 #

• last(col("C"), ignorenulls=True)：会忽略窗口范围内的null值，返回窗口中最后一个（即距离当前行最近的上方）非空C列值
• 窗口范围设置为从分区首行到当前行，保证每次取值都只参考当前行及以上的内容，不会取到下方行的值
• 如果实际数据需要按维度分组填充（比如不同报表区域、不同数据块独立填充），只需在窗口定义中增加分区规则即可，示例：

  # 按report_group列分组，每组内部独立做向前填充
  fill_window = Window.partitionBy("report_group").orderBy("row_seq").rowsBetween(Window.unboundedPreceding, Window.currentRow)
  

运行结果示例 #

执行后C列空值会被正确填充，输出格式如下：

+---+----+
|  C|   Y|
+---+----+
| BR|1995|
| BR|1997|
| BR|1999|
| BR|2001|
| BR|2003|
| BR|2005|
| NO|1995|
| NO|1997|
| NO|1999|
| NO|2001|
| NO|2003|
| NO|2005|
| RO|1995|
| RO|1997|
| RO|1999|
| RO|2001|
| RO|2003|
| RO|2005|
| AC|1995|
| AC|1997|
+---+----+

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