title: Serialization issues in Spark Streaming
body: <p>I'm quite confused about how Spark works with the data under the hood. For example, when I run a streaming job and apply <code>foreachRDD</code>, the behaviour depends on whether a variable is captured from the outer scope or initialised inside.</p>
  
  <pre><code>val sparkConf = new SparkConf()
  dStream.foreachRDD(rdd =&gt; {
      val spark = SparkSession.builder.config(sparkConf).getOrCreate()
      ...
  })
  </code></pre>
  
  <p>In this case, I get an exception:</p>
  
  <blockquote>
    <p>java.io.NotSerializableException: Object of org.apache.spark.streaming.kafka.DirectKafkaInputDStream$DirectKafkaInputDStreamCheckpointData is being serialized possibly as a part of closure of an RDD operation. This is because  the DStream object is being referred to from within the closure.  Please rewrite the RDD operation inside this DStream to avoid this.  This has been enforced to avoid bloating of Spark tasks  with unnecessary objects.</p>
  </blockquote>
  
  <p>But if I move <code>sparkConf</code> inside, everything seems to be fine:</p>
  
  <pre><code>dStream.foreachRDD(rdd =&gt; {
      val sparkConf = rdd.sparkContext.getConf
      val spark = SparkSession.builder.config(sparkConf).getOrCreate()
      ...
  })
  </code></pre>
  
  <p>This looks quite odd to me because I thought that <code>foreachRDD</code> runs on the driver node, so I didn't expect any difference.</p>
  
  <p>Now, if I move both the SQL session and the config outside <code>foreachRDD</code>, it works fine again:</p>
  
  <pre><code>val sparkConf = new SparkConf()
  val spark = SparkSession.builder.config(sparkConf).getOrCreate()
  dStream.foreachRDD(rdd =&gt; {
      val df = spark.read.json(rdd)
      ...
  })
  </code></pre>
  
  <p>A <a href="http://spark.apache.org/docs/latest/streaming-programming-guide.html#dataframe-and-sql-operations" rel="noreferrer">snippet</a> in Spark documentation suggests the previous version (where both config and SQL context are created inside <code>foreachRDD</code>), which seems less efficient to me: why create them for every batch if they could be created just once?</p>
  
  <p>Could someone explain why the exception is thrown and what is the proper way to create the SQL context?</p>