title: PySpark structured streaming output sink as Kafka giving error
body: <p>Using Kafka 0.9.0 and Spark 2.1.0 - I am using PySpark structured streaming to compute the results and output it on Kafka topic. I am referring the Spark docs for the same 
  <a href="https://spark.apache.org/docs/latest/structured-streaming-programming-guide.html#output-modes" rel="nofollow noreferrer">https://spark.apache.org/docs/latest/structured-streaming-programming-guide.html#output-modes</a></p>
  
  <p>Now when I run the command </p>
  
  <p>(output mode complete as it is aggregating the streaming data.)</p>
  
  <pre class="lang-py prettyprint-override"><code>(mydataframe.writeStream
      .outputMode("complete")
      .format("kafka")
      .option("kafka.bootstrap.servers", "x.x.x.x:9092")
      .option("topic", "topicname")
      .option("checkpointLocation","/data/checkpoint/1")
      .start())
  </code></pre>
  
  <p>It gives me error as below</p>
  
  <pre><code> ERROR StreamExecution: Query [id = 0686130b-8668-48fa-bdb7-b79b63d82680, runId = b4b7494f-d8b8-416e-ae49-ad8498dfe8f2] terminated with error
  org.apache.spark.sql.AnalysisException: Required attribute 'value' not found;
      at org.apache.spark.sql.kafka010.KafkaWriter$$anonfun$6.apply(KafkaWriter.scala:73)
      at org.apache.spark.sql.kafka010.KafkaWriter$$anonfun$6.apply(KafkaWriter.scala:73)
      at scala.Option.getOrElse(Option.scala:121)
      at org.apache.spark.sql.kafka010.KafkaWriter$.validateQuery(KafkaWriter.scala:72)
      at org.apache.spark.sql.kafka010.KafkaWriter$.write(KafkaWriter.scala:88)
      at org.apache.spark.sql.kafka010.KafkaSink.addBatch(KafkaSink.scala:38)
      at org.apache.spark.sql.execution.streaming.StreamExecution$$anonfun$org$apache$spark$sql$execution$streaming$StreamExecution$$runBatch$1.apply$mcV$sp(StreamExecution.scala:503)
      at org.apache.spark.sql.execution.streaming.StreamExecution$$anonfun$org$apache$spark$sql$execution$streaming$StreamExecution$$runBatch$1.apply(StreamExecution.scala:503)
      at org.apache.spark.sql.execution.streaming.StreamExecution$$anonfun$org$apache$spark$sql$execution$streaming$StreamExecution$$runBatch$1.apply(StreamExecution.scala:503)
      at org.apache.spark.sql.execution.streaming.ProgressReporter$class.reportTimeTaken(ProgressReporter.scala:262)
      at org.apache.spark.sql.execution.streaming.StreamExecution.reportTimeTaken(StreamExecution.scala:46)
      at org.apache.spark.sql.execution.streaming.StreamExecution.org$apache$spark$sql$execution$streaming$StreamExecution$$runBatch(StreamExecution.scala:502)
      at org.apache.spark.sql.execution.streaming.StreamExecution$$anonfun$org$apache$spark$sql$execution$streaming$StreamExecution$$runBatches$1$$anonfun$1.apply$mcV$sp(StreamExecution.scala:255)
      at org.apache.spark.sql.execution.streaming.StreamExecution$$anonfun$org$apache$spark$sql$execution$streaming$StreamExecution$$runBatches$1$$anonfun$1.apply(StreamExecution.scala:244)
      at org.apache.spark.sql.execution.streaming.StreamExecution$$anonfun$org$apache$spark$sql$execution$streaming$StreamExecution$$runBatches$1$$anonfun$1.apply(StreamExecution.scala:244)
      at org.apache.spark.sql.execution.streaming.ProgressReporter$class.reportTimeTaken(ProgressReporter.scala:262)
      at org.apache.spark.sql.execution.streaming.StreamExecution.reportTimeTaken(StreamExecution.scala:46)
      at org.apache.spark.sql.execution.streaming.StreamExecution$$anonfun$org$apache$spark$sql$execution$streaming$StreamExecution$$runBatches$1.apply$mcZ$sp(StreamExecution.scala:244)
      at org.apache.spark.sql.execution.streaming.ProcessingTimeExecutor.execute(TriggerExecutor.scala:43)
      at org.apache.spark.sql.execution.streaming.StreamExecution.org$apache$spark$sql$execution$streaming$StreamExecution$$runBatches(StreamExecution.scala:239)
      at org.apache.spark.sql.execution.streaming.StreamExecution$$anon$1.run(StreamExecution.scala:177)**
  </code></pre>
  
  <p>Not sure what attribute value does it expects. Need help in resolving this.</p>
  
  <p>The console output sink produces the correct output on console so code seems to work fine. Only when using kafka as output sink causing this issue</p>