如何在PySpark中按version分组用中位数填充price列空值？

你可以通过以下步骤实现和Pandas等价的逻辑：

1. 计算各分组的中位数：PySpark没有直接的transform方法计算分组中位数，需先通过groupBy+agg结合percentile_approx（Spark 2.0+支持），计算每个version对应的price中位数。
2. 关联原表与中位数表：将原DataFrame和分组中位数结果按version列关联。
3. 填充空值：用when函数判断price是否为空，为空则替换为分组中位数，否则保留原值。

完整代码示例 #

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, when, percentile_approx

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

# 示例DataFrame
data = [
    ("v1", 10000),
    ("v1", None),
    ("v1", 12000),
    ("v2", 15000),
    ("v2", None),
    ("v2", 14000),
    ("v2", 16000)
]
df = spark.createDataFrame(data, ["version", "price"])

# 1. 计算每个version的price中位数
median_df = df.groupBy("version") \
    .agg(percentile_approx("price", 0.5, 10000).alias("median_price"))

# 2. 关联原表与中位数表
df_with_median = df.join(median_df, on="version", how="left")

# 3. 填充空值
filled_df = df_with_median.withColumn(
    "price",
    when(col("price").isNull(), col("median_price")).otherwise(col("price"))
).drop("median_price")

# 查看结果
filled_df.show()

关键说明 #

• percentile_approx(col, 0.5, 10000)：第一个参数为目标列，第二个参数0.5表示取中位数，第三个参数是精度控制（数值越大精度越高，性能略降），适合大数据场景的近似中位数计算；若需精确中位数，可对分组内数据排序后取中间值，但会影响性能。
• PySpark是不可变数据结构，没有inplace=True操作，所有转换都会生成新的DataFrame。

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