SPARK 1.6.1: Task not serializable when evaluating a classifier on a DataFrame
<p>I have a DataFrame, I map it into an RDD of () to test an SVMModel.</p>

<p>I am using Zeppelin, and Spark 1.6.1</p>

<p>Here is my code:</p>

<pre><code>val loadedSVMModel = SVMModel.load(sc, pathToSvmModel)

// Clear the default threshold.
loadedSVMModel.clearThreshold()

// Compute raw scores on the test set.
val scoreAndLabels = df.select($"features", $"label")
                       .map { case Row(features:Vector, label: Double) =&gt;
                                val score = loadedSVMModel.predict(features)
                                (score,label)
                            }

// Get evaluation metrics.
val metrics = new BinaryClassificationMetrics(scoreAndLabels)
val auROC = metrics.areaUnderROC()

println("Area under ROC = " + auROC)
</code></pre>

<p>When executing the code I have a <code>org.apache.spark.SparkException: Task not serializable;</code> and I have a hard time understanding why this is happening and how can I fix it.</p>

<ul>
<li>Is it caused by the fact that I am using Zeppelin?</li>
<li>Is it because of the original DataFrame?</li>
</ul>

<p>I have executed the <a href="http://spark.apache.org/docs/latest/mllib-linear-methods.html#linear-support-vector-machines-svms" rel="nofollow">SVM example</a> in the Spark Programming Guide, and it worked perfectly. So the reason should be related to one of the points above... I guess.</p>

<p>Here is the some relevant elements of the Exception stack:</p>

<pre><code>Caused by: java.io.NotSerializableException: org.apache.spark.sql.Column
Serialization stack:
    - object not serializable (class: org.apache.spark.sql.Column, value: (sum(CASE WHEN (domainIndex = 0) THEN sumOfScores ELSE 0),mode=Complete,isDistinct=false) AS 0#100278)
    - element of array (index: 0)
    - array (class [Lorg.apache.spark.sql.Column;, size 372)
</code></pre>

<p>I didn't post the full exception stack, because Zeppelin tend to show a very long not relevant text. please let me know if you want me to past the full exception.</p>

<p><strong>Additional information</strong></p>

<p>The feature vectors are generated using a VectorAssembler() as follow</p>

<pre><code>// Prepare vector assemble
val vecAssembler =  new VectorAssembler()
                               .setInputCols(arrayOfIndices)
                               .setOutputCol("features")


// Aggregation expressions
val exprs = arrayOfIndices
                .map(c =&gt; sum(when($"domainIndex" === c, $"sumOfScores")
                .otherwise(lit(0))).alias(c))

val df = vecAssembler
           .transform(anotherDF.groupBy($"userID", $"val")
           .agg(exprs.head, exprs.tail: _*))
           .select($"userID", $"features", $"val")
           .withColumn("label", sqlCreateLabelValue($"val"))
           .drop($"val").drop($"userID")
</code></pre>