Ваш код полностью перепутан (в его текущем состоянии он даже не вызовет исключение, которое вы описали в вопросе). sqrt
должен быть размещен вне reduce
звоните:
from pyspark.sql.functions import col, sqrt
from operator import add
from functools import reduce
df = spark.createDataFrame([("_", "_", 2, 1, 2, 3)], ("_1", "_2", "mean"))
cols = df.columns[3:]
sd = sqrt(
reduce(add, ((col(x) - col("mean")) ** 2 for x in cols)) / (len(cols) - 1)
)
sd
# Column<b'SQRT((((POWER((_4 - mean), 2) + POWER((_5 - mean), 2)) + POWER((_6 - mean), 2)) / 2))'>
df.withColumn("sd", sd).show()
# +---+---+----+---+---+---+---+
# | _1| _2|mean| _4| _5| _6| sd|
# +---+---+----+---+---+---+---+
# | _| _| 2| 1| 2| 3|1.0|
# +---+---+----+---+---+---+---+