title: Performantly flatten a struct from a dataframe read from Kafka
body: <p>We have a Spark Structured Streaming job (Spark 2.4.0) which reads Avro data from Kafka, deserializes it, and saves it to HDFS. The Avro schema has ~30 fields, some of which are maps with dozens of entries.</p>
  <p>Using the built in Spark <code>from_avro</code> function seems to work well, and gives us good performance: the code below can process ~400 records/CPU assigned to the Spark job, which is fast enough for our volume.</p>
  <pre><code>val reader = spark.readStream
  reader.format(&quot;kafka&quot;)
  reader.option(&quot;subscribe&quot;, &quot;mytopic&quot;)
  
  val fromKafka = reader.load()
  
  val output = fromKafka
    .select(from_avro(col(&quot;value&quot;), schema.toString).as(&quot;avro_cols_in_a_struct&quot;))
  
  output
  .writeStream
  .format(&quot;parquet&quot;)
  .option(&quot;path&quot;, &quot;/mypath&quot;)
  .option(&quot;checkpointLocation&quot;, &quot;/mycheckpoint&quot;)
  .option(&quot;compression&quot;, &quot;snappy&quot;)
  .outputMode(&quot;append&quot;)
  .trigger(Trigger.ProcessingTime(&quot;60 seconds&quot;))
  .start()
  </code></pre>
  <p>However, the Spark <code>from_avro</code> function returns a dataframe with a single column: a struct with all the Avro fields as subfields of that struct, when we want all the Avro fields as top level columns. When we try to flatten this struct, everything we've tried massively slows down performance: All 3 of the techniques below lower the overall run time of the above code down to ~40 records/CPU, so a 10x slowdown:</p>
  <p>Technique #1:</p>
  <pre><code>val flatten_struct = output.select($&quot;avro_cols_in_a_struct.*&quot;)
  </code></pre>
  <p>Technique #2:</p>
  <pre><code>val flatten_struct = output.select(&quot;avro_cols_in_a_struct.col1&quot;, &quot;avro_cols_in_a_struct.col2&quot;...)
  </code></pre>
  <p>Technique #3:</p>
  <pre><code>def flattenSchema(schema: StructType, prefix: String = null) : Array[Column] = {
    schema.fields.flatMap(f =&gt; {
      val colName = if (prefix == null) f.name else (prefix + &quot;.&quot; + f.name)
  
      f.dataType match {
        case st: StructType =&gt; flattenSchema(st, colName)
        case _ =&gt; Array(col(colName))
      }
    })
  }
  
  val flatten_struct = output.select(flattenSchema(output.schema):_*)
  
  </code></pre>
  <p>Is there another way to speed up the struct/flattening process? 40 records/sec per CPU is too slow for our data volume. Does anyone have any advice?</p>