title: How to query Spark StreamingContext with spark sql in zeppelin?
body: <p>I am trying to use spark sql to query the data coming from kafka using zeppelin for real time trend analysis but without success.</p>
  
  <p>here is the simple code snippets that I am running in zeppelin</p>
  
  <pre><code>//Load Dependency
  %dep 
      z.reset()
      z.addRepo("Spark Packages Repo").url("http://repo1.maven.org/maven2/")
      z.load("org.apache.spark:spark-streaming-kafka-0-8_2.11:2.0.1")
      z.load("org.apache.spark:spark-core_2.11:2.0.1")
      z.load("org.apache.spark:spark-sql_2.11:2.0.1")
      z.load("org.apache.spark:spark-streaming_2.11:2.0.1"
  
  //simple streaming 
  %spark
  import org.apache.spark.streaming.{Seconds, StreamingContext}
  import org.apache.spark.rdd.RDD
  import org.apache.spark.{SparkContext, SparkConf}
  import org.apache.spark.sql.{Row, SQLContext}
  import org.apache.spark.storage.StorageLevel
  import org.apache.spark.streaming.kafka.KafkaUtils
  import _root_.kafka.serializer.StringDecoder
  import org.apache.spark.sql.SparkSession
  
  val conf = new SparkConf()
      .setAppName("clickstream")
      .setMaster("local[*]")
      .set("spark.streaming.stopGracefullyOnShutdown", "true")
      .set("spark.driver.allowMultipleContexts","true")
  
  
  val spark = SparkSession
    .builder()
    .appName("Spark SQL basic example")
    .config(conf)
    .getOrCreate()
  
  val ssc = new StreamingContext(conf, Seconds(1))
  
  val topicsSet = Set("timer")
  val kafkaParams = Map[String, String]("metadata.broker.list" -&gt; "192.168.25.1:9091,192.168.25.1:9092,192.168.25.1:9093")
  
  val lines = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](
                                     ssc, kafkaParams, topicsSet).map(_._2)
  
  lines.window(Seconds(60)).foreachRDD{ rdd =&gt;
      val clickDF = spark.read.json(rdd) //doesn't have to be json
      clickDF.createOrReplaceTempView("testjson1")
      //olderway
      //clickDF.registerTempTable("testjson2")
      clickDF.show
  
  }
  
  lines.print()
  ssc.start()
  ssc.awaitTermination()
  </code></pre>
  
  <p>I am able to print each kafka message but when I run simple sql <code>%sql select * from testjson1 // or testjson2</code>, I get the following error</p>
  
  <pre><code>java.util.NoSuchElementException: None.get
  at scala.None$.get(Option.scala:347)
  at scala.None$.get(Option.scala:345)
  at org.apache.spark.storage.BlockInfoManager.releaseAllLocksForTask(BlockInfoManager.scala:343)
  at org.apache.spark.storage.BlockManager.releaseAllLocksForTask(BlockManager.scala:646)
  at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:281)
  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>In the <a href="https://zeppelin.apache.org/docs/0.6.2/quickstart/tutorial.html#tutorial-with-streaming-data" rel="nofollow noreferrer">this post</a> Streaming Data is being queried (with twitter example). So I am thinking it should be possible with kafka streaming. So I guess, maybe, I am doing something wrong OR missing some point? </p>
  
  <p>Any ideas, suggestions, recommendation is welcomed</p>