title: How to read records in JSON format from Kafka using Structured Streaming?
body: <p>I am trying to use <a href="http://spark.apache.org/docs/latest/structured-streaming-programming-guide.html" rel="noreferrer">structured streaming approach</a> using Spark-Streaming based on DataFrame/Dataset API to load a stream of data from Kafka.</p>
  
  <p>I use:</p>
  
  <ul>
  <li>Spark 2.10</li>
  <li>Kafka 0.10</li>
  <li>spark-sql-kafka-0-10</li>
  </ul>
  
  <p>Spark Kafka DataSource has defined underlying schema: </p>
  
  <pre><code>|key|value|topic|partition|offset|timestamp|timestampType|
  </code></pre>
  
  <p>My data come in json format and they are stored in the <em>value</em> column. I am looking for a way how to extract underlying schema from value column and update received dataframe to columns stored in <em>value</em>? I tried the approach below but it does not work:</p>
  
  <pre><code> val columns = Array("column1", "column2") // column names
   val rawKafkaDF = sparkSession.sqlContext.readStream
    .format("kafka")
    .option("kafka.bootstrap.servers","localhost:9092")
    .option("subscribe",topic)
    .load()
    val columnsToSelect = columns.map( x =&gt; new Column("value." + x))
    val kafkaDF = rawKafkaDF.select(columnsToSelect:_*)
  
    // some analytics using stream dataframe kafkaDF
  
    val query = kafkaDF.writeStream.format("console").start()
    query.awaitTermination()
  </code></pre>
  
  <p>Here I am getting Exception <code>org.apache.spark.sql.AnalysisException: Can't extract value from value#337;</code> because in time of creation of the stream, values inside are not known...</p>
  
  <p>Do you have any suggestions?</p>