Как преобразовать строковый столбец в UDT Vector с плавающей запятой pyspark

Question

У меня есть фрейм данных, использующий pyspark со столбцом «features», который выглядит следующим образом:

[[-0.65467646, 0.578577578, 0.577757775], [-0.65467646, 0.578577578, 0.577757775, 0.65477645, 0.5887563773], [-0.65467646, 0.578577578, 0.577757775]]

Я хотел бы применить к нему k-means, но он выдает ошибку типа, говорящую, что это строка и следует преобразовать в вектор. Я сделал преобразование, используя UDF-функцию. Он делает что-то вроде:

Udf_vector=udf(lambda v: Vector(v), UDTVector())

, но теперь он говорит, что значения [-0,65467646, 0,578577578, 0,577757775 ... не являются плавающими. Поэтому я снова использовал udf-функцию:

Udf1 =udf(lambda x:[float(y) for y in x])**

df = df.withColumns(col("features", udf(col("features"))

Но это не сработало. Может ли кто-нибудь помочь мне с этим, я был бы очень благодарен. Это мой последний шаг перед применением модели k-средних

Answer 1

Ваш столбец функций не является типом массива. Сначала вам нужно преобразовать его в массив.

Вы можете снять квадратные скобки и разбить строку, чтобы получить массив.

df = df.withColumn("features", split(expr("rtrim(']', ltrim('[', features))"), ","))

Теперь у вас есть массив строк, где каждый элемент похож на массив, поэтому вам нужно будет также преобразовать каждый элемент в массив. Для Spark 2.4+ вы можете сделать это с помощью функции transform:

df = df.withColumn("features", expr("transform(features, x -> split(rtrim(']', ltrim('[', x)), ','))"))

Наконец, сгладьте внутренние массивы, приведите элементы строк к плавающим и конвертируйте в вектор:

from pyspark.ml.linalg import VectorUDT
from pyspark.ml.linalg import Vectors

array_to_vector = udf(lambda a: Vectors.dense(a), VectorUDT())

df = df.withColumn("features", flatten(col("features")).cast("array<float>"))\
       .withColumn("features", array_to_vector(col("features")))

df.printSchema()

#root
# |-- id: long (nullable = true)
# |-- features: vector (nullable = true)

Собираем все вместе:

df = df.withColumn("features", split(expr("rtrim(']', ltrim('[', features))"), ",")) \
       .withColumn("features", expr("""transform(features, x -> split(rtrim(']', ltrim('[', x)), ","))"""))\
       .withColumn("features", flatten(col("features")).cast("array<float>"))\
       .withColumn("features", array_to_vector(col("features")))