title: Reading schema of streaming Dataframe in Spark Structured Streaming
body: <p>I'm new with Apache Spark Structured Streaming. I'm trying to read some events from Event Hub (in XML format) and trying to create new Spark DF from the nested XML.</p>
  <p>Im using the code example described in <a href="https://github.com/databricks/spark-xml" rel="nofollow noreferrer">https://github.com/databricks/spark-xml</a> and in batch mode is running perfectly but not in Structured Spark Streaming.</p>
  <p><strong>Code chunk of spark-xml Github library</strong></p>
  <pre><code>import com.databricks.spark.xml.functions.from_xml
  import com.databricks.spark.xml.schema_of_xml
  import spark.implicits._
  val df = ... /// DataFrame with XML in column 'payload' 
  val payloadSchema = schema_of_xml(df.select(&quot;payload&quot;).as[String])
  val parsed = df.withColumn(&quot;parsed&quot;, from_xml($&quot;payload&quot;, payloadSchema))
  </code></pre>
  <p><strong>My batch code</strong></p>
  <pre><code>val df = Seq(
    (8, &quot;&lt;AccountSetup xmlns:xsi=\&quot;test\&quot;&gt;&lt;Customers test=\&quot;a\&quot;&gt;d&lt;/Customers&gt;&lt;tag1&gt;7&lt;/tag1&gt; &lt;tag2&gt;4&lt;/tag2&gt; &lt;mode&gt;0&lt;/mode&gt; &lt;Quantity&gt;1&lt;/Quantity&gt;&lt;/AccountSetup&gt;&quot;),
    (64, &quot;&lt;AccountSetup xmlns:xsi=\&quot;test\&quot;&gt;&lt;Customers test=\&quot;a\&quot;&gt;d&lt;/Customers&gt;&lt;tag1&gt;6&lt;/tag1&gt; &lt;tag2&gt;4&lt;/tag2&gt;  &lt;mode&gt;0&lt;/mode&gt; &lt;Quantity&gt;1&lt;/Quantity&gt;&lt;/AccountSetup&gt;&quot;),
    (27, &quot;&lt;AccountSetup xmlns:xsi=\&quot;test\&quot;&gt;&lt;Customers test=\&quot;a\&quot;&gt;d&lt;/Customers&gt;&lt;tag1&gt;4&lt;/tag1&gt; &lt;tag2&gt;4&lt;/tag2&gt; &lt;mode&gt;3&lt;/mode&gt; &lt;Quantity&gt;1&lt;/Quantity&gt;&lt;/AccountSetup&gt;&quot;)
  ).toDF(&quot;number&quot;, &quot;body&quot;)
  )
  
  
  val payloadSchema = schema_of_xml(df.select(&quot;body&quot;).as[String])
  val parsed = df.withColumn(&quot;parsed&quot;, from_xml($&quot;body&quot;, payloadSchema))
  
  val final_df = parsed.select(parsed.col(&quot;parsed&quot;))
  display(final_df.select(&quot;parsed.*&quot;))
  </code></pre>
  <p><a href="https://i.stack.imgur.com/rH6QC.png" rel="nofollow noreferrer"><img src="https://i.stack.imgur.com/rH6QC.png" alt="enter image description here" /></a></p>
  <p>I was trying to do same logic for Spark Structured Streaming like the following code:</p>
  <p><strong>Structured Streaming code</strong></p>
  <pre class="lang-scala prettyprint-override"><code>import com.databricks.spark.xml.functions.from_xml
  import com.databricks.spark.xml.schema_of_xml
  import org.apache.spark.eventhubs.{ ConnectionStringBuilder, EventHubsConf, EventPosition }
  import spark.implicits._
  import org.apache.spark.sql.functions._
  import org.apache.spark.sql.types._
  
  
  val streamingInputDF = 
    spark.readStream
      .format(&quot;eventhubs&quot;)
      .options(eventHubsConf.toMap)
      .load()
  
  val payloadSchema = schema_of_xml(streamingInputDF.select(&quot;body&quot;).as[String])
  val parsed = streamingSelectDF.withColumn(&quot;parsed&quot;, from_xml($&quot;body&quot;, payloadSchema))
  val final_df = parsed.select(parsed.col(&quot;parsed&quot;))
  
  display(final_df.select(&quot;parsed.*&quot;))
  </code></pre>
  <p>In code part of  <code>val payloadSchema = schema_of_xml(streamingInputDF.select(&quot;body&quot;).as[String])</code> intstruction throws the error <code> Queries with streaming sources must be executed with writeStream.start();;</code></p>
  <p><strong>Update</strong></p>
  <p>Tried to</p>
  <pre class="lang-scala prettyprint-override"><code>
  val streamingInputDF = 
    spark.readStream
      .format(&quot;eventhubs&quot;)
      .options(eventHubsConf.toMap)
      .load()
      .select(($&quot;body&quot;).cast(&quot;string&quot;))
  
  val body_value = streamingInputDF.select(&quot;body&quot;).as[String]
  body_value.writeStream
      .format(&quot;console&quot;)
      .start()
  
  spark.streams.awaitAnyTermination()
  
  
  val payloadSchema = schema_of_xml(body_value)
  val parsed = body_value.withColumn(&quot;parsed&quot;, from_xml($&quot;body&quot;, payloadSchema))
  val final_df = parsed.select(parsed.col(&quot;parsed&quot;))
  </code></pre>
  <p>Now is not running into the error but Databricks stay in &quot;Waiting status&quot;</p>
  <p><a href="https://i.stack.imgur.com/4CkHg.png" rel="nofollow noreferrer"><img src="https://i.stack.imgur.com/4CkHg.png" alt="enter image description here" /></a>
  Thanks!!</p>