title: Structured Streaming - java.util.NoSuchElementException: key not found
body: <p>We've recently started with structured streaming on azure data-bricks.
  Currently we're consuming events from event hubs and writing them on to azure datalake store as parquet.</p>
  
  <p>I'm able to write the streams onto console, but encountering error when we try writing them onto any physical storage (Blob/Azure Datalake) "java.util.NoSuchElementException: key not found:"</p>
  
  <pre><code>    val schema = new StructType()
                        .add("col1",StringType, nullable = true)
                        .add("col2", StringType, nullable = true)
                        .add("col3", StringType, nullable = true)
                        .add("col4",StringType, nullable = true)
  
  
  val messages = incomingStream.selectExpr("offset","partitionKey","cast (body as string) AS Content")
  val structuredMsg = messages.select($"offset",$"partitionKey",from_json(col("Content"),schema).alias("data"))
  
  val results = structuredMsg.
  select($"offset",$"partitionKey",current_date().as("date_1"),$"data.col1".as("col1"),$"data.col2".as("col2"),$"data.col3".as("col3"),$"data.col4".as("col4"))
  
  import scala.concurrent.duration._
  import org.apache.spark.sql.streaming.{OutputMode, Trigger}
  
  results.
    withColumn("date", $"date_1").
    writeStream.
    format("text"). // write as Parquet partitioned by date
    partitionBy("date").
    option("path", "dbfs:/mnt/datalake/XXX-databricks-mount/XXX-databricks/test").
    option("checkpointLocation", "dbfs:/checkpoint_path/").
    trigger(Trigger.ProcessingTime(60.seconds)).
    outputMode(OutputMode.Append).
    start
  
  
  
  
  java.util.NoSuchElementException: key not found: {"ehName":"test1","partitionId":1}
  at scala.collection.MapLike$class.default(MapLike.scala:228)
  at scala.collection.AbstractMap.default(Map.scala:59)
  at scala.collection.MapLike$class.apply(MapLike.scala:141)
  at scala.collection.AbstractMap.apply(Map.scala:59)
  at org.apache.spark.sql.eventhubs.EventHubsSource$$anonfun$getBatch$2.apply(EventHubsSource.scala:233)
  at org.apache.spark.sql.eventhubs.EventHubsSource$$anonfun$getBatch$2.apply(EventHubsSource.scala:231)
  at scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:234)
  at scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:234)
  at scala.collection.immutable.Map$Map2.foreach(Map.scala:137)
  at scala.collection.TraversableLike$class.map(TraversableLike.scala:234)
  at scala.collection.AbstractTraversable.map(Traversable.scala:104)
  at org.apache.spark.sql.eventhubs.EventHubsSource.getBatch(EventHubsSource.scala:231)
  at org.apache.spark.sql.execution.streaming.MicroBatchExecution$$anonfun$org$apache$spark$sql$execution$streaming$MicroBatchExecution$$runBatch$1$$anonfun$apply$9.apply(MicroBatchExecution.scala:394)
  at org.apache.spark.sql.execution.streaming.MicroBatchExecution$$anonfun$org$apache$spark$sql$execution$streaming$MicroBatchExecution$$runBatch$1$$anonfun$apply$9.apply(MicroBatchExecution.scala:390)
  at scala.collection.TraversableLike$$anonfun$flatMap$1.apply(TraversableLike.scala:241)
  at scala.collection.TraversableLike$$anonfun$flatMap$1.apply(TraversableLike.scala:241)
  at scala.collection.Iterator$class.foreach(Iterator.scala:893)
  at scala.collection.AbstractIterator.foreach(Iterator.scala:1336)
  at scala.collection.IterableLike$class.foreach(IterableLike.scala:72)
  </code></pre>