title: Merging rows into a single struct column in spark scala has efficiency problems, how do we do it better?
body: <p>I am trying to speed up and limit the cost of taking several columns and their values and inserting them into a map in the same row. This is a requirement because we have a legacy system that is reading from this job and it isn't yet ready to be refactored. There is also another map with some data that needs to be combined with this. </p>
  
  <p>Currently we have a few solutions all of which seem to result in about the same run time on the same cluster with around 1TB of data stored in Parquet: </p>
  
  <pre><code>import org.apache.spark.sql.functions._
  import org.apache.spark.sql.types._
  import org.json4s._
  import org.json4s.jackson.JsonMethods._
  import spark.implicits._
  
  def jsonToMap(s: String, map: Map[String, String]): Map[String, String] = { 
    implicit val formats = org.json4s.DefaultFormats
      val jsonMap = if(!s.isEmpty){
        parse(s).extract[Map[String, String]]
      } else {
        Map[String, String]()
      }
      if(map != null){
        map ++ jsonMap
      } else {
        jsonMap
      }
    }
  val udfJsonToMap = udf(jsonToMap _)
  
  def addMap(key:String, value:String, map: Map[String,String]): Map[String,String] = {
    if(map == null) {
      Map(key -&gt; value)
    } else {
      map + (key -&gt; value)
    }
  }
  
  val addMapUdf = udf(addMap _)
  
  val output = raw.columns.foldLeft(raw.withColumn("allMap", typedLit(Map.empty[String, String]))) { (memoDF, colName) =&gt;
      if(colName.startsWith("columnPrefix/")){
          memoDF.withColumn("allMap", when(col(colName).isNotNull, addMapUdf(substring_index(lit(colName), "/", -1), col(colName), col("allTagsMap")) ))
      } else if(colName.equals("originalMap")){
          memoDF.withColumn("allMap", when(col(colName).isNotNull, udfJsonToMap(col(colName), col("allMap"))))
      } else {
        memoDF
      }
  }
  </code></pre>
  
  <p>takes about 1h on 9 m5.xlarge </p>
  
  <pre><code>val resourceTagColumnNames = raw.columns.filter(colName =&gt; colName.startsWith("columnPrefix/"))
  def structToMap: Row =&gt; Map[String,String] = { row =&gt;
    row.getValuesMap[String](resourceTagColumnNames)
  }
  val structToMapUdf = udf(structToMap)
  
  val experiment = raw
    .withColumn("allStruct", struct(resourceTagColumnNames.head, resourceTagColumnNames.tail:_*))
    .select("allStruct")
    .withColumn("allMap", structToMapUdf(col("allStruct")))
    .select("allMap")
  </code></pre>
  
  <p>Also runs in about 1h on the same cluster </p>
  
  <p>This code all works but it isn't fast enough it is about 10 times longer than every other transform we have right now and it is a bottle neck for us. </p>
  
  <p>Is there another way to get this result that is more efficient? </p>
  
  <p>Edit: I have also tried limiting the data by a key however because the values in the columns I am merging can change despite the key remaining the same I cannot limit the data size without risking data loss. </p>