如何在PySpark中按组标准化DataFrame所有列？

嘿，我懂你想要复刻Pandas里那种按组对所有列做min-max标准化的操作，PySpark的API逻辑和Pandas不太一样，咱们得换个方式实现，下面是完整的解决方案：

步骤1：创建示例DataFrame #

先把你给的示例数据转换成PySpark DataFrame，方便后续测试：

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

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

# 定义列名和数据
columns = ['group', 'sensor1', 'sensor2', 'sensor3']
vals = [ 
    ("a", 0.8, 0.02, 100), 
    ("a", 0.5, 0.1, 200), 
    ("a", 1, 0.5, 50), 
    ("a", 0, 0.8, 30),
    ("b", 10, 1, 0),
    ("b", 20, 2, 3),
    ("b", 5, 4, 1) 
]

df = spark.createDataFrame(vals, columns)
df.show()

步骤2：计算每个组的min和max值 #

我们需要先按group分组，计算每个数值列的最小值和最大值，然后把这些统计量和原DataFrame关联起来：

# 获取所有需要标准化的数值列（排除group列）
num_cols = [col for col in columns if col != 'group']

# 生成每个列的min和max聚合表达式
agg_exprs = []
for col in num_cols:
    agg_exprs.append(F.min(col).alias(f"{col}_min"))
    agg_exprs.append(F.max(col).alias(f"{col}_max"))

# 计算每个组的min/max并和原表join
group_stats = df.groupBy("group").agg(*agg_exprs)
df_with_stats = df.join(group_stats, on="group", how="inner")

步骤3：执行min-max标准化 #

接下来对每个数值列应用标准化公式(x - x_min)/(x_max - x_min)，同时要处理x_max等于x_min的情况（避免除以零报错），最后填充空值为0：

norm_cols = []
for col in num_cols:
    # 计算标准化值，处理除以零的情况
    norm_col = F.when(
        F.col(f"{col}_max") != F.col(f"{col}_min"),
        (F.col(col) - F.col(f"{col}_min")) / (F.col(f"{col}_max") - F.col(f"{col}_min"))
    ).otherwise(0.0).alias(col)
    norm_cols.append(norm_col)

# 组合结果：保留group列，加上标准化后的数值列
df_norm = df_with_stats.select("group", *norm_cols).fillna(0)

查看结果 #

运行完上面的代码后，查看最终的标准化结果：

df_norm.show(truncate=False)

输出结果和你期望的一致，比如group 'a'的sensor1列，0.8对应的标准化值是(0.8-0)/(1-0)=0.8，sensor3列的200对应(200-30)/(200-30)=1，完全符合预期。

简化写法（可选） #

如果你觉得上面的循环有点繁琐，也可以用列表推导式简化：

num_cols = [col for col in columns if col != 'group']

group_stats = df.groupBy("group").agg(
    *[F.min(c).alias(f"{c}_min") for c in num_cols],
    *[F.max(c).alias(f"{c}_max") for c in num_cols]
)

df_norm = df.join(group_stats, on="group") \
    .select(
        "group",
        *[
            F.when(
                F.col(f"{c}_max") != F.col(f"{c}_min"),
                (F.col(c) - F.col(f"{c}_min")) / (F.col(f"{c}_max") - F.col(f"{c}_min"))
            ).otherwise(0.0).alias(c)
            for c in num_cols
        ]
    ).fillna(0)

这个方法的核心是先预计算每组的统计量，再通过关联表把统计量带到每一行，最后逐列计算标准化值，完美对应你Pandas代码里的groupby.transform逻辑。

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