title: Serialization Exception on spark
body: <p>I meet a very strange problem on Spark about serialization.
  The code is as below:</p>
  
  <pre><code>class PLSA(val sc : SparkContext, val numOfTopics : Int) extends Serializable
  {
      def infer(document: RDD[Document]): RDD[DocumentParameter] = {
        val docs = documents.map(doc =&gt; DocumentParameter(doc, numOfTopics))
        docs
      }
  }
  </code></pre>
  
  <p>where Document is defined as:</p>
  
  <pre><code>class Document(val tokens: SparseVector[Int]) extends Serializable
  </code></pre>
  
  <p>and DocumentParameter is:</p>
  
  <pre><code>class DocumentParameter(val document: Document, val theta: Array[Float]) extends Serializable
  
  object DocumentParameter extends Serializable
  {
    def apply(document: Document, numOfTopics: Int) = new DocumentParameter(document, 
      Array.ofDim[Float](numOfTopics))
  }
  </code></pre>
  
  <p>SparseVectoris a serializable class in <code>breeze.linalg.SparseVector</code>.</p>
  
  <p>This is a simple map procedure, and all the classes are serializable, but I get this exception:</p>
  
  <pre><code>org.apache.spark.SparkException: Task not serializable
  </code></pre>
  
  <p>But when I remove the <code>numOfTopics</code> parameter, that is:</p>
  
  <pre><code>object DocumentParameter extends Serializable
  {
    def apply(document: Document) = new DocumentParameter(document, 
      Array.ofDim[Float](10))
  }
  </code></pre>
  
  <p>and call it like this:</p>
  
  <pre><code>val docs = documents.map(DocumentParameter.apply)
  </code></pre>
  
  <p>and it seems OK. </p>
  
  <p>Is type Int not serializable? But I do see that some code is written like that. </p>
  
  <p>I am not sure how to fix this bug.</p>
  
  <p><strong>#UPDATED#</strong>:</p>
  
  <p>Thank you @samthebest. I will add more details about it.</p>
  
  <pre><code>stack trace:
  org.apache.spark.SparkException: Task not serializable
      at org.apache.spark.util.ClosureCleaner$.ensureSerializable(ClosureCleaner.scala:166)
      at org.apache.spark.util.ClosureCleaner$.clean(ClosureCleaner.scala:158)
      at org.apache.spark.SparkContext.clean(SparkContext.scala:1242)
      at org.apache.spark.rdd.RDD.map(RDD.scala:270)
      at com.topicmodel.PLSA.infer(PLSA.scala:13)
      at $iwC$$iwC$$iwC$$iwC$$iwC$$iwC.&lt;init&gt;(&lt;console&gt;:30)
      at $iwC$$iwC$$iwC$$iwC$$iwC.&lt;init&gt;(&lt;console&gt;:35)
      at $iwC$$iwC$$iwC$$iwC.&lt;init&gt;(&lt;console&gt;:37)
      at $iwC$$iwC$$iwC.&lt;init&gt;(&lt;console&gt;:39)
      at $iwC$$iwC.&lt;init&gt;(&lt;console&gt;:41)
      at $iwC.&lt;init&gt;(&lt;console&gt;:43)
      at &lt;init&gt;(&lt;console&gt;:45)
      at .&lt;init&gt;(&lt;console&gt;:49)
      at .&lt;clinit&gt;(&lt;console&gt;)
      at .&lt;init&gt;(&lt;console&gt;:7)
      at .&lt;clinit&gt;(&lt;console&gt;)
      at $print(&lt;console&gt;)
      at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
      at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
      at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
      at java.lang.reflect.Method.invoke(Method.java:483)
      at org.apache.spark.repl.SparkIMain$ReadEvalPrint.call(SparkIMain.scala:789)
      at org.apache.spark.repl.SparkIMain$Request.loadAndRun(SparkIMain.scala:1062)
      at org.apache.spark.repl.SparkIMain.loadAndRunReq$1(SparkIMain.scala:615)
      at org.apache.spark.repl.SparkIMain.interpret(SparkIMain.scala:646)
      at org.apache.spark.repl.SparkIMain.interpret(SparkIMain.scala:610)
      at org.apache.spark.repl.SparkILoop.reallyInterpret$1(SparkILoop.scala:814)
      at org.apache.spark.repl.SparkILoop.interpretStartingWith(SparkILoop.scala:859)
      at org.apache.spark.repl.SparkILoop.command(SparkILoop.scala:771)
      at org.apache.spark.repl.SparkILoop.processLine$1(SparkILoop.scala:616)
      at org.apache.spark.repl.SparkILoop.innerLoop$1(SparkILoop.scala:624)
      at org.apache.spark.repl.SparkILoop.loop(SparkILoop.scala:629)
      at org.apache.spark.repl.SparkILoop$$anonfun$process$1.apply$mcZ$sp(SparkILoop.scala:954)
      at org.apache.spark.repl.SparkILoop$$anonfun$process$1.apply(SparkILoop.scala:902)
      at org.apache.spark.repl.SparkILoop$$anonfun$process$1.apply(SparkILoop.scala:902)
      at scala.tools.nsc.util.ScalaClassLoader$.savingContextLoader(ScalaClassLoader.scala:135)
      at org.apache.spark.repl.SparkILoop.process(SparkILoop.scala:902)
      at org.apache.spark.repl.SparkILoop.process(SparkILoop.scala:997)
      at org.apache.spark.repl.Main$.main(Main.scala:31)
      at org.apache.spark.repl.Main.main(Main.scala)
      at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
      at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
      at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
      at java.lang.reflect.Method.invoke(Method.java:483)
      at org.apache.spark.deploy.SparkSubmit$.launch(SparkSubmit.scala:328)
      at org.apache.spark.deploy.SparkSubmit$.main(SparkSubmit.scala:75)
      at org.apache.spark.deploy.SparkSubmit.main(SparkSubmit.scala)
  Caused by: java.io.NotSerializableException: org.apache.spark.SparkContext
      at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1184)
      at java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1548)
      at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1509)
      at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1432)
      at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1178)
      at java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1548)
      at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1509)
      at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1432)
      at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1178)
      at java.io.ObjectOutputStream.writeObject(ObjectOutputStream.java:348)
      at org.apache.spark.serializer.JavaSerializationStream.writeObject(JavaSerializer.scala:42)
      at org.apache.spark.serializer.JavaSerializerInstance.serialize(JavaSerializer.scala:73)
      at org.apache.spark.util.ClosureCleaner$.ensureSerializable(ClosureCleaner.scala:164)
      ... 46 more
  </code></pre>
  
  <p>As the stack trace gives the general information of exception, I removed it.</p>
  
  <p>I run the code in the spark-shell.</p>
  
  <pre><code>// suppose I have get RDD[Document] for docs
  val numOfTopics = 100
  val plsa = new PLSA(sc, numOfTopics)
  val docPara = plsa.infer(docs)
  </code></pre>
  
  <p>Could you give me some tutorials or tips on serializable?</p>