title: NullPointerException in SQLContext.read() Spark
body: <p>I am trying to read JSON records, produced by Kafka, in Spark using SQLContext.read(). Every time NullPointerException appears. </p>
  
  <pre><code>    SparkConf conf = new SparkConf()
         .setAppName("kafka-sandbox")
          .setMaster("local[*]");
      JavaSparkContext sc = new JavaSparkContext(conf);
      JavaStreamingContext ssc = new JavaStreamingContext(sc, new Duration(2000));
  
      Set&lt;String&gt; topics = Collections.singleton(topicString);
      Map&lt;String, String&gt; kafkaParams = new HashMap&lt;&gt;();
      kafkaParams.put("metadata.broker.list", servers);
  
      JavaPairInputDStream&lt;String, String&gt; directKafkaStream = KafkaUtils.createDirectStream(
              ssc, String.class, String.class, StringDecoder.class, StringDecoder.class, 
              kafkaParams, topics);
      SQLContext sqlContext = new org.apache.spark.sql.SQLContext(sc);
  
      directKafkaStream
          .map(message -&gt; message._2)
          .foreachRDD(rdd -&gt; {
              rdd.foreach(record -&gt; {
                  Dataset&lt;Row&gt; ds = sqlContext.read().json(rdd);
              });
           });
      ssc.start();
      ssc.awaitTermination();
  </code></pre>
  
  <p>Here is a log:</p>
  
  <pre><code>java.lang.NullPointerException
      at org.apache.spark.sql.SparkSession.sessionState$lzycompute(SparkSession.scala:112)
      at org.apache.spark.sql.SparkSession.sessionState(SparkSession.scala:110)
      at org.apache.spark.sql.DataFrameReader.&lt;init&gt;(DataFrameReader.scala:535)
      at org.apache.spark.sql.SparkSession.read(SparkSession.scala:595)
      at org.apache.spark.sql.SQLContext.read(SQLContext.scala:504)
      at SparkJSONConsumer$1.lambda$2(SparkJSONConsumer.java:73)
      at SparkJSONConsumer$1$$Lambda$8/1821075039.call(Unknown Source)
      at org.apache.spark.api.java.JavaRDDLike$$anonfun$foreach$1.apply(JavaRDDLike.scala:350)
      at org.apache.spark.api.java.JavaRDDLike$$anonfun$foreach$1.apply(JavaRDDLike.scala:350)
      at scala.collection.Iterator$class.foreach(Iterator.scala:893)
      at scala.collection.AbstractIterator.foreach(Iterator.scala:1336)
      at org.apache.spark.rdd.RDD$$anonfun$foreach$1$$anonfun$apply$27.apply(RDD.scala:875)
      at org.apache.spark.rdd.RDD$$anonfun$foreach$1$$anonfun$apply$27.apply(RDD.scala:875)
      at org.apache.spark.SparkContext$$anonfun$runJob$5.apply(SparkContext.scala:1897)
      at org.apache.spark.SparkContext$$anonfun$runJob$5.apply(SparkContext.scala:1897)
      at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:70)
      at org.apache.spark.scheduler.Task.run(Task.scala:85)
      at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:274)
      at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
      at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
      at java.lang.Thread.run(Thread.java:745)
  </code></pre>
  
  <p>I assume the problem is due to foreachRDD clause, but can't figure it out. So any suggestions would be great.</p>
  
  <p>Also, I am using sqlContext, because after I plan to serialize records in avro format ("com.databricks.spark.avro"). If there is a way to serialize a string, containing JSON structure, to avro format without defining the schema, you are very welcome to share it!</p>
  
  <p>Thanks in advance.</p>