title: Passing a map with struct-type key into a Spark UDF
body: <p>I want to write an Spark 1.6 UDF which takes the following map:</p>
  
  <pre><code>case class MyRow(mapping: Map[(Int, Int), Double])
  
  val data = Seq(
    MyRow(Map((1, 1) -&gt; 1.0))
  )
  val df = sc.parallelize(data).toDF()
  
  df.printSchema()
  
  root
   |-- mapping: map (nullable = true)
   |    |-- key: struct
   |    |-- value: double (valueContainsNull = false)
   |    |    |-- _1: integer (nullable = false)
   |    |    |-- _2: integer (nullable = false)
  </code></pre>
  
  <p>(As a side-note: I find the above output strange as the Type of the key is printed below the type of the value, why is that?)</p>
  
  <p>Now I define my UDF as:</p>
  
  <pre><code>val myUDF = udf((inputMapping: Map[(Int,Int), Double]) =&gt;
    inputMapping.map { case ((i1, i2), value) =&gt; ((i1 + i2), value) }
  )
  
  df
    .withColumn("udfResult", myUDF($"mapping"))
    .show()
  </code></pre>
  
  <p>But this gives me:</p>
  
  <pre><code>java.lang.ClassCastException: org.apache.spark.sql.catalyst.expressions.GenericRowWithSchema cannot be cast to scala.Tuple2
  </code></pre>
  
  <p>So I tried to replace <code>(Int,Int)</code> with a custom <code>case class</code>, because this is how I normally do it if I want to pass a <code>struct</code> to an UDF:</p>
  
  <pre><code>case class MyTuple2(i1: Int, i2: Int)
  val myUDF = udf((inputMapping: Map[MyTuple2, Double]) =&gt; 
    inputMapping.map { case (MyTuple2(i1, i2), value) =&gt; ((i1 + i2), value) }
  )
  </code></pre>
  
  <p>This strangely gives :</p>
  
  <pre><code>org.apache.spark.sql.AnalysisException: cannot resolve 'UDF(mapping)' due to data type mismatch: argument 1 requires map&lt;struct&lt;i1:int,i2:int&gt;,double&gt; type, however, 'mapping' is of map&lt;struct&lt;_1:int,_2:int&gt;,double&gt; type.
  </code></pre>
  
  <p>I don't understand the above exception as the types match.</p>
  
  <p>The only (ugly) solution I've found is passing a <code>org.apache.spark.sql.Row</code> and then "extract" the elements of the struct:</p>
  
  <pre><code>val myUDF = udf((inputMapping: Map[Row, Double]) =&gt; inputMapping
    .map { case (key, value) =&gt; ((key.getInt(0), key.getInt(1)), value) } // extract Row into Tuple2
    .map { case ((i1, i2), value) =&gt; ((i1 + i2), value) }
  )
  </code></pre>