title: How to define schema of streaming dataset dynamically to write to csv?
body: <p>I have a streaming dataset, reading from kafka and trying to write to CSV</p>
  
  <pre><code>case class Event(map: Map[String,String])
  def decodeEvent(arrByte: Array[Byte]): Event = ...//some implementation
  val eventDataset: Dataset[Event] = spark
    .readStream
    .format("kafka")
    .load()
    .select("value")
    .as[Array[Byte]]
    .map(decodeEvent)
  </code></pre>
  
  <p><code>Event</code> holds <code>Map[String,String]</code> inside and to write to CSV I'll need some schema.</p>
  
  <p>Let's say all the fields are of type <code>String</code> and so I tried the example from <a href="https://github.com/apache/spark/blob/master/examples/src/main/scala/org/apache/spark/examples/sql/SparkSQLExample.scala" rel="nofollow noreferrer">spark repo</a></p>
  
  <pre><code>val columns = List("year","month","date","topic","field1","field2")
  val schema = new StructType() //Prepare schema programmatically
  columns.foreach { field =&gt; schema.add(field, "string") }
  val rowRdd = eventDataset.rdd.map { event =&gt; Row.fromSeq(
       columns.map(c =&gt; event.getOrElse(c, "")
  )}
  val df = spark.sqlContext.createDataFrame(rowRdd, schema)
  </code></pre>
  
  <p>This gives error at runtime on line "eventDataset.rdd":</p>
  
  <blockquote>
    <p>Caused by: org.apache.spark.sql.AnalysisException: Queries with
    streaming sources must be executed with writeStream.start();;</p>
  </blockquote>
  
  <p>Below doesn't work because '.map' has a List[String] not Tuple</p>
  
  <pre><code>eventDataset.map(event =&gt; columns.map(c =&gt; event.getOrElse(c,""))
  .toDF(columns:_*)
  </code></pre>
  
  <p>Is there a way to achieve this with programmatic schema and structured streaming datasets?</p>