如何使用 pyspark mllib RegressionMetrics 进行真实预测答案

【问题标题】：How to use pyspark mllib RegressionMetrics with real predictions如何使用 pyspark mllib RegressionMetrics 进行真实预测
【发布时间】：2015-10-05 15:44:22
【问题描述】：

对于 pyspark 1.4，我正在尝试使用 RegressionMetrics() 进行预测由 LinearRegressionWithSGD 生成。

pyspark mllib documentations 中给出的所有 RegressionMetrics() 示例均用于“人工”预测和观察喜欢

predictionAndObservations = sc.parallelize([ (2.5, 3.0), (0.0, -0.5), (2.0, 2.0), (8.0, 7.0)])

对于这样的“人工”（使用 sc.parallelize 生成）RDD，一切正常。但是，当对以另一种方式生成的另一个 RDD 执行相同操作时，我得到了

TypeError: DoubleType can not accept object in type <type 'numpy.float64'>

下面是简短的可重现示例。

可能是什么问题？

from pyspark.mllib.regression import LabeledPoint
from pyspark.mllib.regression import LinearRegressionWithSGD, LinearRegressionModel
from pyspark.mllib.evaluation import RegressionMetrics

dataRDD = sc.parallelize([LabeledPoint(1, [1,1]), LabeledPoint(2, [2,2]), LabeledPoint(3, [3,3])])
lrModel = LinearRegressionWithSGD.train(dataRDD)
prediObserRDD = dataRDD.map(lambda p: (lrModel.predict(p.features), p.label)).cache()

让我们检查 RDD 确实是（预测，观察）对

prediObserRDD.take(4) # looks OK

现在尝试计算指标

metrics = RegressionMetrics(prediObserRDD)

它给出了以下错误

TypeError                                 Traceback (most recent call last)
<ipython-input-1-ca9ad8e9faf1> in <module>()
      7 prediObserRDD = dataRDD.map(lambda p: (lrModel.predict(p.features), p.label)).cache()
      8 prediObserRDD.take(4)
----> 9 metrics = RegressionMetrics(prediObserRDD)
     10 #metrics.explainedVariance
     11 #metrics.meanAbsoluteError

/usr/local/spark-1.4.0-bin-hadoop2.6/python/pyspark/mllib/evaluation.py in __init__(self, predictionAndObservations)
     99         df = sql_ctx.createDataFrame(predictionAndObservations, schema=StructType([
    100             StructField("prediction", DoubleType(), nullable=False),
--> 101             StructField("observation", DoubleType(), nullable=False)]))
    102         java_class = sc._jvm.org.apache.spark.mllib.evaluation.RegressionMetrics
    103         java_model = java_class(df._jdf)

/usr/local/spark-1.4.0-bin-hadoop2.6/python/pyspark/sql/context.py in createDataFrame(self, data, schema, samplingRatio)
    337 
    338         for row in rows:
--> 339             _verify_type(row, schema)
    340 
    341         # convert python objects to sql data

/usr/local/spark-1.4.0-bin-hadoop2.6/python/pyspark/sql/types.py in _verify_type(obj, dataType)
   1027                              "length of fields (%d)" % (len(obj), len(dataType.fields)))
   1028         for v, f in zip(obj, dataType.fields):
-> 1029             _verify_type(v, f.dataType)
   1030 
   1031 _cached_cls = weakref.WeakValueDictionary()

/usr/local/spark-1.4.0-bin-hadoop2.6/python/pyspark/sql/types.py in _verify_type(obj, dataType)
   1011     if type(obj) not in _acceptable_types[_type]:
   1012         raise TypeError("%s can not accept object in type %s"
-> 1013                         % (dataType, type(obj)))
   1014 
   1015     if isinstance(dataType, ArrayType):

TypeError: DoubleType can not accept object in type <type 'numpy.float64'>

BinaryClassificationMetrics 也会出现同样的问题（对于另一个数据集和分类任务）。

【问题讨论】：

@eliasah 感谢您的评论，我在 SparkProgrammingGuide 中发现 DoubleWritable 对应于 python 类型“float”。所以我将值转换为浮点数，现在一切正常。请发表您的评论作为答案，以便我接受它
顺便说一下，这几乎是 spark 的意外行为。 DenseVector，例如 IS numpy 数组。将所有 numpy.float 类型自动转换为 DoubleType 是合理的。

标签： apache-spark pyspark apache-spark-mllib

【解决方案1】：

就像错误所说的TypeError: DoubleType can not accept object in type <type 'numpy.float64'>

您正在尝试将 numpy.float64 转换为无法完成的 Double。

要解决该 TypeError，您必须将值转换为可接受的类型。

例子：

from pyspark.mllib.regression import LabeledPoint
from pyspark.mllib.regression import LinearRegressionWithSGD, LinearRegressionModel
from pyspark.mllib.evaluation import RegressionMetrics

dataRDD = sc.parallelize([LabeledPoint(1, [1,1]), LabeledPoint(2, [2,2]), LabeledPoint(3, [3,3])])
lrModel = LinearRegressionWithSGD.train(dataRDD)
prediObserRDD = dataRDD.map(lambda p: (float(lrModel.predict(p.features)), p.label)).cache()

如果你注意到了，我已经使用 Python 内置的float 函数将预测标签转换为双精度标签。

现在您可以计算指标了：

>>> metrics = RegressionMetrics(prediObserRDD)
>>> metrics.explainedVariance
1.0

【讨论】：