title: How to create a Spark UDF in Java / Kotlin which returns a complex type?
body: <p>I'm trying to write an UDF which returns a complex type:</p>
  
  <pre><code>private val toPrice = UDF1&lt;String, Map&lt;String, String&gt;&gt; { s -&gt;
      val elements = s.split(" ")
      mapOf("value" to elements[0], "currency" to elements[1])
  }
  
  
  val type = DataTypes.createStructType(listOf(
          DataTypes.createStructField("value", DataTypes.StringType, false),
          DataTypes.createStructField("currency", DataTypes.StringType, false)))
  df.sqlContext().udf().register("toPrice", toPrice, type)
  </code></pre>
  
  <p>but any time I use this:</p>
  
  <pre><code>df = df.withColumn("price", callUDF("toPrice", col("price")))
  </code></pre>
  
  <p>I get a cryptic error:</p>
  
  <pre class="lang-none prettyprint-override"><code>Caused by: org.apache.spark.SparkException: Failed to execute user defined function($anonfun$28: (string) =&gt; struct&lt;value:string,currency:string&gt;)
      at org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage1.processNext(Unknown Source)
      at org.apache.spark.sql.execution.BufferedRowIterator.hasNext(BufferedRowIterator.java:43)
      at org.apache.spark.sql.execution.WholeStageCodegenExec$$anonfun$10$$anon$1.hasNext(WholeStageCodegenExec.scala:614)
      at org.apache.spark.sql.execution.SparkPlan$$anonfun$2.apply(SparkPlan.scala:253)
      at org.apache.spark.sql.execution.SparkPlan$$anonfun$2.apply(SparkPlan.scala:247)
      at org.apache.spark.rdd.RDD$$anonfun$mapPartitionsInternal$1$$anonfun$apply$25.apply(RDD.scala:830)
      at org.apache.spark.rdd.RDD$$anonfun$mapPartitionsInternal$1$$anonfun$apply$25.apply(RDD.scala:830)
      at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:38)
      at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:324)
      at org.apache.spark.rdd.RDD.iterator(RDD.scala:288)
      at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:38)
      at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:324)
      at org.apache.spark.rdd.RDD.iterator(RDD.scala:288)
      at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:87)
      at org.apache.spark.scheduler.Task.run(Task.scala:109)
      at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:345)
      at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
      at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
      at java.lang.Thread.run(Thread.java:748)
  Caused by: scala.MatchError: {value=138.0, currency=USD} (of class java.util.LinkedHashMap)
      at org.apache.spark.sql.catalyst.CatalystTypeConverters$StructConverter.toCatalystImpl(CatalystTypeConverters.scala:236)
      at org.apache.spark.sql.catalyst.CatalystTypeConverters$StructConverter.toCatalystImpl(CatalystTypeConverters.scala:231)
      at org.apache.spark.sql.catalyst.CatalystTypeConverters$CatalystTypeConverter.toCatalyst(CatalystTypeConverters.scala:103)
      at org.apache.spark.sql.catalyst.CatalystTypeConverters$$anonfun$createToCatalystConverter$2.apply(CatalystTypeConverters.scala:379)
      ... 19 more
  </code></pre>
  
  <p>I tried to use a custom data type:</p>
  
  <pre><code>class Price(val value: Double, val currency: String) : Serializable
  </code></pre>
  
  <p>with an UDF which returns that type:</p>
  
  <pre><code>private val toPrice = UDF1&lt;String, Price&gt; { s -&gt;
      val elements = s.split(" ")
      Price(elements[0].toDouble(), elements[1])
  }
  </code></pre>
  
  <p>but then I get another <code>MatchError</code> which complains for the <code>Price</code> type.</p>
  
  <p><em>How do I properly write an UDF which can return a complex type?</em></p>