Преобразование pyspark DataFrame в LabeledPoint без отбрасывания на RDD

Question

Я новичок в pyspark DataFrame, и это действительно меня почему-то вызывает. Может быть, я не понимаю этого правильно.

сказать, что у меня есть кадр

a = sqlContext.createDataFrame([[(2,3,4)],[(1,2,3)]],['things']) 

данных Если я хочу, чтобы преобразовать это в LabeledPoint, мне нужно опуститься до РДА, используя функцию карты.

from pyspark.mllib.regression import LabeledPoint 
def convert(x): 
    z = [float(y) for y in x] 
    return LabeledPoint(z[0], z[1:]) 

rdd = a.map(lambda x: convert(x['things'])) 
rdd.take(2) 
df=rdd.toDF() 

Почему я не могу использовать оператор выбора DataFrame в сочетании с UDF, как показано ниже?

from pyspark.sql.functions import udf 
def convert(x): 
    z = [float(y) for y in x] 
    return LabeledPoint(z[0], z[1:]) 

udf_convert = udf(convert) 
df = a.select(udf_convert(a['things'])) 
display(df) 

Я получаю сообщение об ошибке, которая говорит:

org.apache.spark.SparkException: Job aborted due to stage failure: Task 6 in stage 33.0 failed 1 times, most recent failure: Lost task 6.0 in stage 33.0 (TID 101, localhost): net.razorvine.pickle.PickleException: expected zero arguments for construction of ClassDict (for pyspark.mllib.linalg.DenseVector) 

Answer 1

Проблема с вашим udfdefinition. Возвращаемое значение не является StringType (по умолчанию), поэтому вам нужно указать его. К сожалению, нет LabeledPointUDT, поэтому вы не можете использовать udf для такой операции.

Принимая в один шаг назад, LabeledPoint используется с mllib, который работает в основном с RDD. Почему вы хотите сделать LabeledPoint в DataFrame? В любом случае будет неудобно использовать его позже.

Если вы решили использовать DataFrame + ml, вы будете рады узнать, что есть VectorUDT (в mllib.linalg до того Спарк 2.0, в ml.linalg с 2.0), чтобы помочь с вашими udf функциями.