title: Not Serializable exception when integrating SQL and Spark Streaming
body: <p>In Addition to <a href="https://stackoverflow.com/questions/26774046/not-serializable-exception-when-integrating-spark-sql-and-spark-streaming">Not Serializable exception when integrating Spark SQL and Spark Streaming</a></p>
  
  <p>My source code </p>
  
  <pre class="lang-java prettyprint-override"><code>public static void main(String args[]) {
      SparkConf sparkConf = new SparkConf().setAppName("NumberCount");
      JavaSparkContext jc = new JavaSparkContext(sparkConf);
      JavaStreamingContext jssc = new JavaStreamingContext(jc, new Duration(2000));
      jssc.addStreamingListener(new WorkCountMonitor());
      int numThreads = Integer.parseInt(args[3]);
      Map&lt;String,Integer&gt; topicMap = new HashMap&lt;String,Integer&gt;();
      String[] topics = args[2].split(",");
      for (String topic : topics) {
          topicMap.put(topic, numThreads);
      }
      JavaPairReceiverInputDStream&lt;String,String&gt; data = KafkaUtils.createStream(jssc, args[0], args[1], topicMap);
      data.print();
  
      JavaDStream&lt;Person&gt; streamData = data.map(new Function&lt;Tuple2&lt;String, String&gt;, Person&gt;() {
              public Person call(Tuple2&lt;String,String&gt; v1) throws Exception {
                  String[] stringArray = v1._2.split(",");
                  Person Person = new Person();
                  Person.setName(stringArray[0]);
                  Person.setAge(stringArray[1]);
                  return Person;
              }
  
          });
  
  
      final JavaSQLContext sqlContext = new JavaSQLContext(jc);
      streamData.foreachRDD(new Function&lt;JavaRDD&lt;Person&gt;,Void&gt;() {
          public Void call(JavaRDD&lt;Person&gt; rdd) {
  
              JavaSchemaRDD subscriberSchema = sqlContext.applySchema(rdd, Person.class);
  
              subscriberSchema.registerAsTable("people");
              System.out.println("all data");
              JavaSchemaRDD names = sqlContext.sql("SELECT name FROM people");
              System.out.println("afterwards");
  
              List&lt;String&gt; males = new ArrayList&lt;String&gt;();
  
              males = names.map(new Function&lt;Row,String&gt;() {
                  public String call(Row row) {
                      return row.getString(0);
                  }
              }).collect();
              System.out.println("before for");
              for (String name : males) {
                  System.out.println(name);
              }
              return null;
          }
      });
      jssc.start();
      jssc.awaitTermination();
  }
  </code></pre>
  
  <p>JavaSQLContext is also declared outside the ForeachRDD loop but i am still getting NonSerializableException</p>
  
  <blockquote>
    <p>14/12/23 23:49:38 ERROR JobScheduler: Error running job streaming job 1419378578000 ms.1
    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:1435)
            at org.apache.spark.rdd.RDD.map(RDD.scala:271)
            at org.apache.spark.api.java.JavaRDDLike$class.map(JavaRDDLike.scala:78)
            at org.apache.spark.sql.api.java.JavaSchemaRDD.map(JavaSchemaRDD.scala:42)
            at com.basic.spark.NumberCount$2.call(NumberCount.java:79)
            at com.basic.spark.NumberCount$2.call(NumberCount.java:67)
            at org.apache.spark.streaming.api.java.JavaDStreamLike$$anonfun$foreachRDD$1.apply(JavaDStreamLike.scala:274)
            at org.apache.spark.streaming.api.java.JavaDStreamLike$$anonfun$foreachRDD$1.apply(JavaDStreamLike.scala:274)
            at org.apache.spark.streaming.dstream.DStream$$anonfun$foreachRDD$1.apply(DStream.scala:529)
            at org.apache.spark.streaming.dstream.DStream$$anonfun$foreachRDD$1.apply(DStream.scala:529)
            at org.apache.spark.streaming.dstream.ForEachDStream$$anonfun$1.apply$mcV$sp(ForEachDStream.scala:42)
            at org.apache.spark.streaming.dstream.ForEachDStream$$anonfun$1.apply(ForEachDStream.scala:40)
            at org.apache.spark.streaming.dstream.ForEachDStream$$anonfun$1.apply(ForEachDStream.scala:40)
            at scala.util.Try$.apply(Try.scala:161)
            at org.apache.spark.streaming.scheduler.Job.run(Job.scala:32)
            at org.apache.spark.streaming.scheduler.JobScheduler$JobHandler.run(JobScheduler.scala:171)
            at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)
            at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)
            at java.lang.Thread.run(Thread.java:724)
    Caused by: java.io.NotSerializableException: org.apache.spark.sql.api.java.JavaSQLContext
            at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1181)
            at java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1541)
            at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1506)
            at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1429)
            at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1175)
            at java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1541)
            at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1506)
            at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1429)
            at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1175)
            at java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1541)
            at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1506)
            at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1429)
            at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1175)
            at java.io.ObjectOutputStream.writeObject(ObjectOutputStream.java:347)
            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)
            ... 20 more</p>
  </blockquote>
  
  <p>I appreciate if you have any suggestion.</p>