Scala – How to register a function to sqlContext UDF in scala

apache-sparkapache-spark-sqlscala

I have a method called getAge(timestamp:Long) and I want to register this as a sql function.

I have

sqlContext.udf.register("getAge",getAge)

But its telling me I need arguments or use _ afterwards, I tried using _ but gives me error. How do I register it with an argument. I am new to scala so I have no idea how to do this.

Best Answer

sqlContext.udf.register("getAge",getAge)

should be:

sqlContext.udf.register("getAge",getAge _)

The underscore (must have a space in between function and underscore) turns the function into a partially applied function that can be passed in the registration.

More explanation

When we invoke a function, we have to pass in all the required parameters. If we don't, the compiler will complain.

We can however ask it for the function as a value, with which we can pass in the required parameters at a later time. How we do this is to use the underscore.

getAge means to run getAge - for example, def getAge = 10 giving us 10. We don't want the result, we want the function. Moreover, with your definition, the compiler sees that getAge requires a parameter, and complains that one wasn't given.

What we want to do here is to pass getAge as a function value. We tell Scala, we don't know the parameter yet, we want the function as a value and we'll supply it with the required parameter at a later time. So, we use getAge _.

Assuming signature for getAge is:

getAge(l: Long): Long = <function>

getAge _ becomes an anonymous function:

Long => Long = <function>

which means it needs a parameter of type Long and the result of invoking it will yield a value of type Long.

Related Solutions

Scala – How to profile methods in Scala

Do you want to do this without changing the code that you want to measure timings for? If you don't mind changing the code, then you could do something like this:

def time[R](block: => R): R = {
    val t0 = System.nanoTime()
    val result = block    // call-by-name
    val t1 = System.nanoTime()
    println("Elapsed time: " + (t1 - t0) + "ns")
    result
}

// Now wrap your method calls, for example change this...
val result = 1 to 1000 sum

// ... into this
val result = time { 1 to 1000 sum }

Python – How to use Scala UDF in PySpark

Agree with @user6910411, you have to call apply method directly on the function. So, your code will be.

UDF in Scala:

import org.apache.spark.sql.expressions.UserDefinedFunction
import org.apache.spark.sql.functions._


object ScalaPySparkUDFs {

    def testFunction1(x: Int): Int = { x * 2 }

    def getFun(): UserDefinedFunction = udf(testFunction1 _ )
}

PySpark code:

def test_udf(col):
    sc = spark.sparkContext
    _test_udf = sc._jvm.com.test.ScalaPySparkUDFs.getFun()
    return Column(_test_udf.apply(_to_seq(sc, [col], _to_java_column)))


row = Row("Value")
numbers = sc.parallelize([1,2,3,4]).map(row).toDF()
numbers.withColumn("Result", test_udf(numbers['Value']))

Best Answer

More explanation

Related Solutions

Scala – How to profile methods in Scala

Python – How to use Scala UDF in PySpark

Related Topic