Error "AttributeError: 'Py4JError' object has no attribute 'message' building DecisionTreeModel

Question

I'm following Chapter 4 from "Advanced Analytics with Spark" from O'Reilly. This book is in Scala and I'm having trouble converting this code to Python.

Scala Code

import org.apache.spark.mllib.linalg._
import org.apache.spark.mllib.regression._
val rawData = sc.textFile("hdfs:///user/ds/covtype.data")
val data = rawData.map { line =>
    val values = line.split(',').map(_.toDouble)
    val featureVector = Vectors.dense(values.init)
    val label = values.last - 1
    LabeledPoint(label, featureVector)
}
val Array(trainData, cvData, testData) =
  data.randomSplit(Array(0.8, 0.1, 0.1))
trainData.cache()
cvData.cache()
testData.cache()


import org.apache.spark.mllib.evaluation._
import org.apache.spark.mllib.tree._
import org.apache.spark.mllib.tree.model._
import org.apache.spark.rdd._

def getMetrics(model: DecisionTreeModel, data: RDD[LabeledPoint]):
    MulticlassMetrics = {
 val predictionsAndLabels = data.map(example =>
    (model.predict(example.features), example.label)
 )
 new MulticlassMetrics(predictionsAndLabels)
}
val model = DecisionTree.trainClassifier(
 trainData, 7, Map[Int,Int](), "gini", 4, 100)

val metrics = getMetrics(model, cvData) 
metrics.confusionMatrix

My Python Code

from pyspark.sql.functions import col, split
import pyspark.mllib.linalg as linal
import pyspark.mllib.regression as regre
import pyspark.mllib.evaluation as eva
import pyspark.mllib.tree as tree
import pyspark.rdd as rd

raw_data = sc.textFile("covtype.data")

def fstDecisionTree(line):
    values = list(map(float,line.split(",")))
    featureVector = linal.Vectors.dense(values[:-1])
    label = values[-1]-1
    ret=regre.LabeledPoint(label, featureVector)
    return regre.LabeledPoint(label, featureVector) 

data = raw_data.map(fstDecisionTree)
trainData,cvData,testData=data.randomSplit([0.8,0.1,0.1])
trainData.cache()
cvData.cache()
testData.cache()

def help_lam(model):
 def _help_lam(dataline):
    print(dataline)
    a=dataline.collect()
    return (model.predict(a[1]),a[0])
return _help_lam

def getMetrics(model, data):
    print(type(model),type(data))
    predictionsAndLabels= data.map(help_lam(model))
    return eva.MulticlassMetrics(predictionsAndLabels)

n_targets=7
max_depth=4
max_bin_count=100
model = tree.DecisionTree.trainClassifier(trainData, n_targets, {}, "gini", max_depth, max_bin_count)

metrics=getMetrics(model,cvData)

When I run this, I have this error in the method def _help_lam(dataline) inside of def help_lam(model) when I try to implicitly pass the map iteration in:

AttributeError: 'Py4JError' object has no attribute 'message'

Answer 1

I think the problem is in the model.predict function

From pyspark mllib/tree.py

Note: In Python, predict cannot currently be used within an RDD transformation or action. Call predict directly on the RDD instead.

What you can do is pass the feature vector directly like so

>>> rdd = sc.parallelize([[1.0], [0.0]])
>>> model.predict(rdd).collect()
[1.0, 0.0]

Edit:

An update to your getMetrics could be:

def getMetrics(model, data):
    labels = data.map(lambda d: d.label)
    features = data.map(lambda d: d.features)
    predictions = model.predict(features)
    predictionsAndLabels = predictions.zip(labels)
    return eva.MulticlassMetrics(predictionsAndLabels)