Spark udf with non column parameters

Question

I want to pass a variable and not a column to a UDF in spark.

The map is of the following format Spark dataframe to nested map

val joinUDF = udf((replacementLookup: Map[String, Double], newValue: String) => {
    replacementLookup.get(newValue) match {
      case Some(tt) => tt
      case None => 0.0
    }
  })

Should be mapped like

(columnsMap).foldLeft(df) {
    (currentDF, colName) =>
      {
        println(colName._1)
        println(colName._2)
        currentDF
          .withColumn("myColumn_" + colName._1, joinUDF(colName._2, col(colName._1)))
      }
  }

But throws

type mismatch;
[error]  found   : Map
[error]  required: org.apache.spark.sql.Column
[error]           .withColumn("myColumn_" + colName._1, joinUDF(colName._2, col(colName._1)))

score 7 · Answer 1 · answered Dec 20 '16 at 13:41

You can use currying:

import org.apache.spark.sql.functions._
val df = Seq(("a", 1), ("b", 2)).toDF("StringColumn", "IntColumn")

def joinUDF(replacementLookup: Map[String, Double]) = udf((newValue: String) => {
  replacementLookup.get(newValue) match {
    case Some(tt) => tt
    case None => 0.0
  }
})

val myMap = Map("a" -> 1.5, "b" -> 3.0)

df.select(joinUDF(myMap)($"StringColumn")).show()

Also, you can try with a broadcast variable:

import org.apache.spark.sql.functions._
val df = Seq(("a", 1), ("b", 2)).toDF("StringColumn", "IntColumn")

val myMap = Map("a" -> 1.5, "b" -> 3.0)
val broadcastedMap = sc.broadcast(myMap)

def joinUDF = udf((newValue: String) => {
  broadcastedMap.value.get(newValue) match {
    case Some(tt) => tt
    case None => 0.0
  }
})

df.select(joinUDF($"StringColumn")).show()

Raphael Roth · Accepted Answer · 2016-12-20T13:52:03.327

4

If you want to pass literals to an UDF, use org.apache.spark.sql.functions.lit

i.e. use joinUDF(lit(colName._2), col(colName._1))

But maps aren't supported, so you have to rewrite your code, e.g. by applying the Map-argument before creating the udf

val joinFunction = (replacementLookup: Map[String, Double], newValue: String) => {
   replacementLookup.get(newValue) match {
     case Some(tt) => tt
     case None => 0.0
  }
}

 (columnsMap).foldLeft(df) {
   (currentDF, colName) =>
   {
     val joinUDF = udf(joinFunction(colName._2, _:String))
     currentDF
       .withColumn("myColumn_" + colName._1, joinUDF(col(colName._1)))
   }
 }

edited Dec 20 '16 at 13:52

answered Dec 20 '16 at 13:07

Raphael Roth

26,751
15
88
145

but that yields Unsupported literal type class scala.collection.immutable.Map – Georg Heiler Dec 20 '16 at 13:08
ok, this means you cannot pass a map to an UDF :) So you have to refactor your code, e.g. you a closure – Raphael Roth Dec 20 '16 at 13:12
currently working on that. Thanks. The literal is great. – Georg Heiler Dec 20 '16 at 13:12
Your current workaround results in: `missing parameter type for expanded function `. I will pass a string. Thats fine as well. – Georg Heiler Dec 20 '16 at 13:19
@GeorgHeiler Corrected my answer – Raphael Roth Dec 20 '16 at 13:24
Are you referring to joinUDF or to join? Or did you deliberately introduce this new function? – Georg Heiler Dec 20 '16 at 13:25
That still does not work: $anonfun$1 cannot be cast to scala.Function1 – Georg Heiler Dec 20 '16 at 13:30

Spark udf with non column parameters

2 Answers2