将pyspark DataFrame转换为Labeled Point而不丢弃到RDD

Question

我是pyspark DataFrame的新手，这真的让我很头疼。也许我没有正确理解这一点。

说我有一个数据帧

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

如果我想将其转换为LabeledPoint，我需要使用map函数下拉到RDD。

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 select语句与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

问题在于您的udf定义。返回值不是StringType(默认值(，因此需要指定它。遗憾的是，没有LabeledPointUDT，因此无法使用udf进行此类操作。

退一步来看，LabeledPoint与mllib一起使用，后者主要与RDD一起工作。为什么要在DataFrame中制作LabeledPoint？不管怎样，以后使用它会很尴尬。

如果您选择使用DataFrame+ml，您会很高兴地发现有一个VectorUDT(在Spark 2.0之前的mllib.linalg中，在2.0的ml.linalg中(可以帮助您实现udf功能。