Проблема с UDF для столбца Векторов в PySpark DataFrame
У меня возникли проблемы с использованием UDF для столбца Векторы в PySpark, который можно проиллюстрировать здесь:
from pyspark import SparkContext
from pyspark.sql import Row
from pyspark.sql.types import DoubleType
from pyspark.sql.functions import udf
from pyspark.mllib.linalg import Vectors
FeatureRow = Row('id', 'features')
data = sc.parallelize([(0, Vectors.dense([9.7, 1.0, -3.2])),
(1, Vectors.dense([2.25, -11.1, 123.2])),
(2, Vectors.dense([-7.2, 1.0, -3.2]))])
df = data.map(lambda r: FeatureRow(*r)).toDF()
vector_udf = udf(lambda vector: sum(vector), DoubleType())
df.withColumn('feature_sums', vector_udf(df.features)).first()
Это терпит неудачу со следующей трассировкой стека:
Py4JJavaError: An error occurred while calling z:org.apache.spark.api.python.PythonRDD.collectAndServe.
: org.apache.spark.SparkException: Job aborted due to stage failure: Task 5 in stage 31.0 failed 1 times, most recent failure: Lost task 5.0 in stage 31.0 (TID 95, localhost): org.apache.spark.api.python.PythonException: Traceback (most recent call last):
File "/Users/colin/src/spark/python/lib/pyspark.zip/pyspark/worker.py", line 111, in main
process()
File "/Users/colin/src/spark/python/lib/pyspark.zip/pyspark/worker.py", line 106, in process
serializer.dump_stream(func(split_index, iterator), outfile)
x1 File "/Users/colin/src/spark/python/lib/pyspark.zip/pyspark/serializers.py", line 263, in dump_stream
vs = list(itertools.islice(iterator, batch))
File "/Users/colin/src/spark/python/pyspark/sql/functions.py", line 469, in <lambda>
func = lambda _, it: map(lambda x: f(*x), it)
File "/Users/colin/pokitdok/spark_mapper/spark_mapper/filters.py", line 143, in <lambda>
TypeError: unsupported operand type(s) for +: 'int' and 'NoneType'
Глядя на то, что передается в UDF, кажется, что-то странное. Переданный аргумент должен быть Vector, но вместо этого ему передается кортеж Python:
(1, None, None, [9.7, 1.0, -3.2])
Разве нельзя использовать UDF для столбцов DataFrame Векторов?
РЕДАКТИРОВАТЬ
Так что в списке рассылки было указано, что этоизвестная проблема, Собираюсь принять ответ от @hyim, поскольку он делает временным обходным решением для плотных векторов.