PySpark技术问询：如何将DataFrame空值替换为对应行的均值

你遇到的报错是因为PySpark的Column对象不支持Python式的迭代操作，必须用PySpark内置的分布式函数实现行级计算。以下是可行的解决方案：

步骤1：导入PySpark函数库 #

from pyspark.sql import functions as F

步骤2：定义行均值计算逻辑 #

用PySpark内置表达式计算每行非空值的均值，避免低效的自定义UDF：

# 生成包含目标列的数组，过滤空值后计算均值
row_avg_expr = """
    aggregate(
        filter(array(col1, col2, col3), x -> x is not null),
        0D,
        (acc, x) -> acc + x,
        acc -> acc / size(filter(array(col1, col2, col3), x -> x is not null))
    )
"""
row_avg = F.expr(row_avg_expr)

逻辑说明：

• 将col1、col2、col3转为数组
• 过滤数组中的空值
• 用aggregate累加非空值总和，最后除以非空值数量得到均值

步骤3：替换每个列的空值 #

对每个列判断是否为空，空值替换为行均值，否则保留原值：

df_result = df.select(
    *[F.when(F.col(c).isNull(), row_avg).otherwise(F.col(c)).alias(c) for c in df.columns]
)

完整代码示例 #

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

# 初始化SparkSession
spark = SparkSession.builder.appName("ReplaceNullWithRowAvg").getOrCreate()

# 创建示例DataFrame
data = [(1, 2, 3), (4, None, 6), (7, 8, None)]
df = spark.createDataFrame(data, ["col1", "col2", "col3"])

# 定义行均值表达式
row_avg_expr = """
    aggregate(
        filter(array(col1, col2, col3), x -> x is not null),
        0D,
        (acc, x) -> acc + x,
        acc -> acc / size(filter(array(col1, col2, col3), x -> x is not null))
    )
"""
row_avg = F.expr(row_avg_expr)

# 替换空值
df_result = df.select(
    *[F.when(F.col(c).isNull(), row_avg).otherwise(F.col(c)).alias(c) for c in df.columns]
)

# 查看结果
df_result.show()

运行后输出与你期望的一致：

+----+----+----+
|col1|col2|col3|
+----+----+----+
|   1|   2|   3|
|   4|   5|   6|
|   7|   8| 7.5|
+----+----+----+

报错原因说明 #

你遇到的Column is not iterable或Invalid argument, not a string or column错误，通常是因为尝试用Python原生循环遍历Column对象，或者把Column当作普通Python可迭代对象处理。PySpark的Column是分布式计算的抽象，必须用PySpark提供的函数（如when、expr、数组函数等）操作，不能用Python原生迭代逻辑。

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