Нормализация столбца данных фрейма pyspark ML

Question

Я пытаюсь сделать L1 нормализация значений столбца в кадре данных с использованием pyspark ML библиотека. Ниже мой код. Но это не удается сделать. Можете ли вы помочь мне выяснить, что не так с этим кодом?

from pyspark.ml.feature import Normalizer

y = range(1,10) 
data = spark.createDataFrame([[float(e), ] for e in y])
#data.select('_1').show()

normalizer = Normalizer(p=1.0, inputCol="_1", outputCol="features")
data2 = normalizer.transform(data)
data2.select("features").show()

Следующее является частью журнала ошибок.

Py4JJavaError: An error occurred while calling o857.showString.
: org.apache.spark.SparkException: Job aborted due to stage failure: 
Task 0 in stage 36.0 failed 4 times, most recent failure: Lost task 0.3
in stage 36.0 (TID 67,  XXXXX.serveraddress.com): 
org.apache.spark.SparkException: Failed to execute user defined 
function($anonfun$createTransformFunc$1: (double) => vector)

Answer 1

Normalizer используется для нормализации Vectors через Rows, а не скаляры по столбцам.

Для масштабирования L1 один или несколько скалярных столбцов вы можете попробовать:

data.select([
    (data[c] / s).alias(c) 
    for c, s in zip(data.columns, data.groupBy().sum().first())
])

## +--------------------+
## |                  _1|
## +--------------------+
## |0.022222222222222223|
## |0.044444444444444446|
## | 0.06666666666666667|
## | 0.08888888888888889|
## |  0.1111111111111111|
## | 0.13333333333333333|
## | 0.15555555555555556|
## | 0.17777777777777778|
## |                 0.2|
## +--------------------+

но остерегайтесь возможных проблем переполнения и числовой точности.

Вы можете, конечно, масштабировать Column с Pipeline API также:

from pyspark.ml.feature import SQLTransformer

stf = SQLTransformer(statement="""
    WITH norm AS (SELECT SUM({inputCol}) L1 FROM __THIS__)  
    SELECT /*+ MAPJOIN(norm) +*/ 
      __THIS__.*, __THIS__.{inputCol} / norm.L1 {outputCol} 
    FROM __THIS__  CROSS JOIN norm
    """.format(inputCol="_1", outputCol="_1_scaled"))

stf.transform(data)

## +---+--------------------+
## | _1|           _1_scaled|
## +---+--------------------+
## |1.0|0.022222222222222223|
## |2.0|0.044444444444444446|
## |3.0| 0.06666666666666667|
## |4.0| 0.08888888888888889|
## |5.0|  0.1111111111111111|
## |6.0| 0.13333333333333333|
## |7.0| 0.15555555555555556|
## |8.0| 0.17777777777777778|
## |9.0|                 0.2|
## +---+--------------------+