PySpark DataFrame合并行填充空列的实现方案

解决方案思路 #

因为同一Car+Time组内的Val1/Val2/Val3列各自只有一个非空值，所以可以通过分组聚合的方式，对每个分组的目标列取非空值来合并行。常用的聚合函数如max()、first(ignorenulls=True)、last(ignorenulls=True)都能满足需求。

代码实现 #

首先创建测试用的DataFrame（模拟你的数据源）：

from pyspark.sql import SparkSession
from pyspark.sql.functions import max, first

spark = SparkSession.builder.appName("MergeRows").getOrCreate()

# 原数据，注意列名"Val 3"
data = [
    (1, 1, None, 1.5, None),
    (1, 1, 3.5, None, None),
    (1, 1, None, None, 3.4),
    (1, 2, 2.5, None, None),
    (1, 2, None, 6.0, None),
    (1, 2, None, None, 7.3)
]

df = spark.createDataFrame(data, ["Car", "Time", "Val1", "Val2", "Val 3"])

然后执行合并操作：

# 先把列名"Val 3"改成"Val3"，和目标格式一致
df_renamed = df.withColumnRenamed("Val 3", "Val3")

# 按Car和Time分组，聚合取非空值
merged_df = df_renamed.groupBy("Car", "Time") \
    .agg(
        max("Val1").alias("Val1"),
        max("Val2").alias("Val2"),
        max("Val3").alias("Val3")
    )

# 查看结果
merged_df.show()

说明 #

• 用max()是因为同一组内目标列只有一个非空值，最大值就是这个非空值，空值会被忽略；
• 如果更倾向于取第一个出现的非空值，可以替换max()为first("Val1", ignorenulls=True)；
• 执行后得到的结果就是你需要的合并后DataFrame。

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