title: Can't Tranform Kafka Json Data in Spark Structured Streaming
body: <p>I am trying to get Kafka messages and processing it with Spark in standalone. Kafka stores data as json format. I can get Kafka messages but can not parse json data with defining schema.</p>
  
  <p>When I run the <code>bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic my_kafka_topic --from-beginning</code> command to see the kafka messages in kafka topic, it outputs as follows:</p>
  
  <pre><code>"{\"timestamp\":1553792312117,\"values\":[{\"id\":\"Simulation.Simulator.Temperature\",\"v\":21,\"q\":true,\"t\":1553792311686}]}"
  "{\"timestamp\":1553792317117,\"values\":[{\"id\":\"Simulation.Simulator.Temperature\",\"v\":22,\"q\":true,\"t\":1553792316688}]}"
  </code></pre>
  
  <p>And I can get this data succesfully with this code block in Spark:</p>
  
  <pre><code>df = spark \
  .readStream \
  .format("kafka") \
  .option("kafka.bootstrap.servers", "localhost:9092") \
  .option("subscribe", "my_kafka_topic") \
  .load() \
  .select(col("value").cast("string"))
  </code></pre>
  
  <p>The schema is like this: </p>
  
  <pre><code>df.printSchema()
  
  root
   |-- value: string (nullable = true)
  </code></pre>
  
  <p>And then writing this dataframe to console and it prints the kafka messages:</p>
  
  <pre><code>Batch: 9
  -------------------------------------------
  +--------------------+
  |               value|
  +--------------------+
  |"{\"timestamp\":1...|
  +--------------------+
  </code></pre>
  
  <p>But I want to parse json data to define schema and the code block that I've tried to do it:</p>
  
  <pre><code>schema = StructType([ StructField("timestamp", LongType(), False), StructField("values", ArrayType( StructType([ StructField("id", StringType(), True), StructField("v", IntegerType(), False), StructField("q", BooleanType(), False), StructField("t", LongType(), False) ]), True ), True) ])
  
  parsed = spark \
    .readStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", "localhost:9092") \
    .option("subscribe", "my_kafka_topic") \
    .load() \
    .select(from_json(col("value").cast("string"), schema).alias("opc"))
  </code></pre>
  
  <p>And the schema of <code>parsed</code> dataframe:</p>
  
  <pre><code>parsed.printSchema()
  root
    |-- opc: struct (nullable = true)
    |    |-- timestamp: string (nullable = true)
    |    |-- values: struct (nullable = true)
    |    |    |-- id: string (nullable = true)
    |    |    |-- v: integer (nullable = true)
    |    |    |-- q: boolean (nullable = true)
    |    |    |-- t: string (nullable = true)
  </code></pre>
  
  <p>These code blocks run without error. But when I want to write <code>parsed</code> dataframe to the console:</p>
  
  <pre><code>query = parsed\
     .writeStream\
     .format("console")\
     .start()
  
  query.awaitTermination()
  </code></pre>
  
  <p>it is writing <code>null</code> like this in console:</p>
  
  <pre><code>+----+
  | opc|
  +----+
  |null|
  +----+
  </code></pre>
  
  <p>So, it seems there is problem with parsing json data but can't figure out it. </p>
  
  <p>Can you tell me what is wrong?</p>