PySpark：如何使用首个非空值与最后一个非空值实现DataFrame列的Coalesce操作

这个需求我之前也遇到过，其实拆解开来两步就能搞定——先拿到首尾的非空基准值，再根据位置判断填充就行。我给你一步步演示，用你描述的场景来举例：

1. 先创建示例DataFrame #

先模拟你提到的有首尾空值的DataFrame：

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

# 初始化SparkSession（如果还没初始化的话）
spark = SparkSession.builder.appName("FillNulls").getOrCreate()

# 创建示例数据
df = spark.createDataFrame(
    [
        (1, None),
        (2, None),
        (3, 0.6),
        (4, 1.2),
        (5, 3.0),
        (6, None),
        (7, None)
    ],
    ["id", "value"]
)

df.show()

输出的原始数据：

+---+-----+
| id|value|
+---+-----+
|  1| null|
|  2| null|
|  3|  0.6|
|  4|  1.2|
|  5|  3.0|
|  6| null|
|  7| null|
+---+-----+

2. 获取首尾非空基准值 #

首先拿到value列的第一个非空值和最后一个非空值，这里要注意用ignorenulls=True来跳过空值：

# 获取首个非空值
first_non_null = df.agg(F.first("value", ignorenulls=True)).first()[0]
# 获取最后一个非空值
last_non_null = df.agg(F.last("value", ignorenulls=True)).first()[0]

这里first_non_null会是0.6，last_non_null会是3.0。

3. 确定首尾非空值的位置 #

接下来需要知道第一个非空值出现在哪一行，最后一个非空值出现在哪一行（假设id是按顺序排列的行标识，如果你的DataFrame没有自增id，可以用monotonically_increasing_id()生成临时顺序列）：

# 获取第一个非空值的最小id（最早出现的位置）
first_pos = df.agg(F.min(F.when(F.col("value").isNotNull(), F.col("id")))).first()[0]
# 获取最后一个非空值的最大id（最晚出现的位置）
last_pos = df.agg(F.max(F.when(F.col("value").isNotNull(), F.col("id")))).first()[0]

4. 填充空值 #

最后用when...otherwise逻辑判断填充：

filled_df = df.withColumn(
    "filled_value",
    F.when(
        # 第一个非空值之前的行，用首非空值填充
        F.col("id") < first_pos,
        first_non_null
    ).when(
        # 最后一个非空值之后的行，用尾非空值填充
        F.col("id") > last_pos,
        last_non_null
    ).otherwise(
        # 中间的行保留原数据
        F.col("value")
    )
)

filled_df.show()

最终输出的结果：

+---+-----+------------+
| id|value|filled_value|
+---+-----+------------+
|  1| null|         0.6|
|  2| null|         0.6|
|  3|  0.6|         0.6|
|  4|  1.2|         1.2|
|  5|  3.0|         3.0|
|  6| null|         3.0|
|  7| null|         3.0|
+---+-----+------------+

补充：如果没有自增id怎么办？ #

如果你的DataFrame没有类似id的顺序列，可以用monotonically_increasing_id()生成一个临时的顺序列来判断位置：

df_with_temp_id = df.withColumn("temp_id", F.monotonically_increasing_id())
# 之后的步骤用temp_id代替id即可，最后记得删掉临时列

边缘情况提醒 #

如果整个列全是空值，那first_non_null和last_non_null都会是None，填充后还是空值，这种情况可以根据你的业务需求额外处理。

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