I'm fresh with scala and udf, now I would like to write a udf which accept 3 parameters from a dataframe columns(one of them is array), for..loop current array, parse and return a case class which will be used afterwards. here's a my code roughly:

case class NewFeatures(dd: Boolean, zz: String)    
val resultUdf = udf((arrays: Option[Row], jsonData: String, placement: Int) => {
      for (item <- arrays) {
        val aa = item.getAs[Long]("aa")
        val bb = item.getAs[Long]("bb")
        breakable {
          if (aa <= 0 || bb <= 0) break
        }
        val cc = item.getAs[Long]("cc")
        val dd = cc > 0

        val jsonData = item.getAs[String]("json_data")
        val jsonDataObject = JSON.parseFull(jsonData).asInstanceOf[Map[String, Any]]
        var zz = jsonDataObject.getOrElse("zz", "").toString
        NewFeatures(dd, zz)

      }
      

    })

when I run it, it will get exception:

java.lang.UnsupportedOperationException: Schema for type Unit is not supported

how should I modify above udf

CodePudding user response：

First of all, try better naming for your variables, for instance in your case, "arrays" is of type Option[Row]. Here, for (item <- arrays) {...} is basically a .map method, using map on Options, you should provide a function, that uses Row and returns a value of some type (~= signature: def map[V](f: Row => V): Option[V], what you want in your case: def map(f: Row => NewFeatures): Option[NewFeature]). While you're breaking out of this map in some circumstances, so there's no assurance for the compiler that the function inside map method would always return an instance of NewFeatures. So it is Unit (it only returns on some cases, and not all). What you want to do could be enhanced in something similar to this:

val funcName: (Option[Row], String, Int) => Option[NewFeatures] = 
  (rowOpt, jsonData, placement) => rowOpt.filter(
    /* your break condition */
  ).map { row => // if passes the filter predicate => 
  // fetch data from row, create new instance
}