title: SPARK 1.6.1: Task not serializable when evaluating a classifier on a DataFrame
body: <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>