title: Spark Java Streaming - Null Pointer Exception while creating a Spark Java Dataset
body: <p>I am trying to Process a query on Presto using JDBC and pass the resultset back to Spark for creating a temp table on it. My result set is in List</p>
  
  <p>I am getting query in form of json Msg from kafka producer. so have created kafka consumer in spark to get the message and do further processing. </p>
  
  <p>Below is my main function :</p>
  
  <pre><code>public static void main(String[] args) throws InterruptedException {
  
      SparkConf conf = new SparkConf();
       conf.setAppName("Wordcount Background");
       conf.setMaster("local");
  
  
      //SparkContext sc = SparkContext.getOrCreate(conf);
       SparkSession spark = 
  SparkSession.builder().config(conf).getOrCreate();
  
       JavaSparkContext sc = new 
  JavaSparkContext(spark.sparkContext());
       JavaStreamingContext ssc = new JavaStreamingContext(sc, 
  Durations.seconds(5));
       SQLContext sqc = new SQLContext(sc);
  
       Set&lt;String&gt; topics = Collections.singleton("TestTopic");
          Map&lt;String, String&gt; kafkaParams = new HashMap&lt;&gt;();
          kafkaParams.put("metadata.broker.list", "172.20.3.189:9092");
  
          JavaPairInputDStream&lt;String, String&gt; directKafkaStream = 
  KafkaUtils.createDirectStream(ssc,
                  String.class, String.class, StringDecoder.class, 
  StringDecoder.class, kafkaParams, topics);
  
          directKafkaStream.foreachRDD(rdd -&gt; {
              //System.out.println("--- New RDD with " + 
  rdd.partitions().size()
                    //  + " partitions and " + rdd.count() + " 
  records");
              rdd.foreach(record -&gt; {
  
                  SparkkafkaJson sk = new SparkkafkaJson();
  
                  Dataset&lt;String&gt; dfrdd = 
  spark.createDataset(sk.process_query(record._2), Encoders.STRING());
                  System.out.print(dfrdd);
                  //Dataset&lt;Row&gt; df = spark.read().json(dfrdd);
                  //df.show();
  
              });
          });
  
          ssc.start();
          ssc.awaitTermination();
  
  }
  </code></pre>
  
  <p>Following is the process_query Method which returns the resultset to the main function :</p>
  
  <pre><code>public List&lt;String&gt; process_query(String queryjson) {
      String resstr="";
      String columnValue="";
      List&lt;String&gt; jsonList = new ArrayList&lt;&gt;();
      //List&lt;String&gt; list=new ArrayList&lt;String&gt;();
      try {
  
          Class.forName(JDBC_DRIVER);
          //Open a connection
          conn = DriverManager.getConnection(DB_URL, USER, PASS);
          //Execute a query
          stmt = conn.createStatement();
          String sql = process_json(queryjson);
          ResultSet res = stmt.executeQuery(sql);
          ResultSetMetaData rsmd = res.getMetaData();
          int columnsNumber = rsmd.getColumnCount();
          //Extract data from result set
          while (res.next()) {
            //System.out.println(res.getString(""));
              Gson userGson=new GsonBuilder().create();
              JsonObject params = new JsonObject();
              for (int i = 1; i &lt;= columnsNumber; i++) {
  
                  String ColName = rsmd.getColumnName(i);
                  String ColVal = res.getString(i);
  
  
                  params.addProperty(ColName, ColVal);
  
              }
  
              resstr = userGson.toJson(params);
              jsonList.add(resstr);
          }
  
  
          //Clean-up environment
          res.close();
          stmt.close();
          conn.close();
        } catch (SQLException se) {
          //Handle errors for JDBC
          se.printStackTrace();
        } catch (Exception e) {
          //Handle errors for Class.forName
          e.printStackTrace();
        } finally {
          //finally block used to close resources
          try {
            if (stmt != null) stmt.close();
          } catch (SQLException sqlException) {
            sqlException.printStackTrace();
          }
          try {
            if (conn != null) conn.close();
          } catch (SQLException se) {
            se.printStackTrace();
          }
        }
      return jsonList;
  }
  </code></pre>
  
  <p><em>But still I get this error output</em></p>
  
  <pre><code>    2019-05-30 13:17:41 INFO  ContextCleaner:54 - Cleaned accumulator 42
      2019-05-30 13:17:41 INFO  ContextCleaner:54 - Cleaned accumulator 109
      2019-05-30 13:17:43 INFO  CodeGenerator:54 - Code generated in 216.222798 
      ms
      2019-05-30 13:17:43 ERROR Executor:91 - Exception in task 1.0 in stage 9.0 
      (TID 19)
      java.lang.NullPointerException
          at 
  org.apache.spark.sql.SparkSession.sessionState$lzycompute(SparkSession.scala:143)
          at org.apache.spark.sql.SparkSession.sessionState(SparkSession.scala:141)
          at org.apache.spark.sql.Dataset.&lt;init&gt;(Dataset.scala:183)
          at org.apache.spark.sql.Dataset$.apply(Dataset.scala:65)
          at org.apache.spark.sql.SparkSession.createDataset(SparkSession.scala:474)
          at org.apache.spark.sql.SparkSession.createDataset(SparkSession.scala:511)
          at SparkkafkaJson.SparkkafkaJson.lambda$1(SparkkafkaJson.java:213)
          at org.apache.spark.api.java.JavaRDDLike$$anonfun$foreach$1.apply(JavaRDDLike.scala:351)
          at org.apache.spark.api.java.JavaRDDLike$$anonfun$foreach$1.apply(JavaRDDLike.scala:351)
          at scala.collection.Iterator$class.foreach(Iterator.scala:891)
          at org.apache.spark.util.NextIterator.foreach(NextIterator.scala:21)
          at org.apache.spark.rdd.RDD$$anonfun$foreach$1$$anonfun$apply$27.apply(RDD.scala:927)
          at org.apache.spark.rdd.RDD$$anonfun$foreach$1$$anonfun$apply$27.apply(RDD.scala:927)
          at org.apache.spark.SparkContext$$anonfun$runJob$5.apply(SparkContext.scala:2101)
          at org.apache.spark.SparkContext$$anonfun$runJob$5.apply(SparkContext.scala:2101)
          at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:90)
          at org.apache.spark.scheduler.Task.run(Task.scala:121)
          at org.apache.spark.executor.Executor$TaskRunner$$anonfun$10.apply(Executor.scala:402)
          at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1360)
          at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:408)
          at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
          at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
          at java.lang.Thread.run(Thread.java:748)
  2019-05-30 13:17:43 WARN  TaskSetManager:66 - Lost task 1.0 in stage 9.0 (TID 19, localhost, executor driver): java.lang.NullPointerException
          at org.apache.spark.sql.SparkSession.sessionState$lzycompute(SparkSession.scala:143)
          at org.apache.spark.sql.SparkSession.sessionState(SparkSession.scala:141)
          at org.apache.spark.sql.Dataset.&lt;init&gt;(Dataset.scala:183)
          at org.apache.spark.sql.Dataset$.apply(Dataset.scala:65)
          at org.apache.spark.sql.SparkSession.createDataset(SparkSession.scala:474)
          at org.apache.spark.sql.SparkSession.createDataset(SparkSession.scala:511)
          at SparkkafkaJson.SparkkafkaJson.lambda$1(SparkkafkaJson.java:213)
          at org.apache.spark.api.java.JavaRDDLike$$anonfun$foreach$1.apply(JavaRDDLike.scala:351)
          at org.apache.spark.api.java.JavaRDDLike$$anonfun$foreach$1.apply(JavaRDDLike.scala:351)
  </code></pre>
  
  <p>Please help</p>