将 pyspark DataFrame 转换为 LabeledPoint 而无需丢弃到 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 definition 上。 return 值不是 StringType（默认值），因此您需要指定它。不幸的是没有 LabeledPointUDT 所以你不能使用 udf 进行这样的操作。

退一步说，LabeledPoint与mllib一起使用，主要作用于RDD。为什么要在 DataFrame 中制作 LabeledPoint？反正以后用起来就别扭了

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

将 pyspark DataFrame 转换为 LabeledPoint 而无需丢弃到 RDD

converting pyspark DataFrame to LabeledPoint without dropping to RDD

pyspark

spark-dataframe