Похоже, это должно сработать, но я получаю ошибки:
mu = mean(df[input])
sigma = stddev(df[input])
dft = df.withColumn(output, (df[input]-mu)/sigma)
pyspark.sql.utils.AnalysisException: "grouping expressions sequence is empty, and '`user`' is not an aggregate function. Wrap '(((CAST(`sum(response)` AS DOUBLE) - avg(`sum(response)`)) / stddev_samp(CAST(`sum(response)` AS DOUBLE))) AS `scaled`)' in windowing function(s) or wrap '`user`' in first() (or first_value) if you don't care which value you get.;;\nAggregate [user#0, sum(response)#26L, ((cast(sum(response)#26L as double) - avg(sum(response)#26L)) / stddev_samp(cast(sum(response)#26L as double))) AS scaled#46]\n+- AnalysisBarrier\n +- Aggregate [user#0], [user#0, sum(cast(response#3 as bigint)) AS sum(response)#26L]\n +- Filter item_id#1 IN (129,130,131,132,133,134,135,136,137,138)\n +- Relation[user#0,item_id#1,response_value#2,response#3,trait#4,response_timestamp#5] csv\n"
Я не уверен, что происходит с этим сообщением об ошибке.
К счастью, мне удалось найти работающий код:
summary = df.select([mean(input).alias('mu'), stddev(input).alias('sigma')])\
.collect().pop()
dft = df.withColumn(output, (df[input]-summary.mu)/summary.sigma)
@pault Не стесняйтесь предлагать ответ, который, по вашему мнению, лучше.
Использование collect()
в целом не является хорошим решением, и вы увидите, что это не будет масштабироваться по мере роста ваших данных.
Если вы не хотите использовать StandardScaler
, лучше использовать Window
для вычисления среднего и стандартного отклонения.
Заимствуя тот же пример из StandardScaler в Spark не работает должным образом:
from pyspark.sql.functions import col, mean, stddev
from pyspark.sql import Window
df = spark.createDataFrame(
np.array(range(1,10,1)).reshape(3,3).tolist(),
["int1", "int2", "int3"]
)
df.show()
#+----+----+----+
#|int1|int2|int3|
#+----+----+----+
#| 1| 2| 3|
#| 4| 5| 6|
#| 7| 8| 9|
#+----+----+----+
Предположим, вы хотите стандартизировать столбец int2
:
input_col = "int2"
output_col = "int2_scaled"
w = Window.partitionBy()
mu = mean(input_col).over(w)
sigma = stddev(input_col).over(w)
df.withColumn(output_col, (col(input_col) - mu)/(sigma)).show()
#+----+----+----+-----------+
#|int1|int2|int3|int2_scaled|
#+----+----+----+-----------+
#| 1| 2| 3| -1.0|
#| 7| 8| 9| 1.0|
#| 4| 5| 6| 0.0|
#+----+----+----+-----------+
Если вы хотите использовать стандартное отклонение генеральной совокупности, как в другом примере, замените pyspark.sql.functions.stddev
на pyspark.sql.functions.stddev_pop()
.
Хороший ответ! Спасибо.
Это работает только для векторов.