Introduction

Spark SQL UDF (a.k.a. User-Defined Function) is a highly useful feature of Spark SQL & DataFrame that extends the Spark built-in capabilities.

Note: UDF’s are the most expensive operations hence use them only you have no choice and when essential.

What is Spark UDF?

UDF a.k.a User Defined Function, If you are coming from SQL background, UDF’s are nothing new to you as most of the traditional RDBMS databases support User Defined Functions, and Spark UDF’s are similar to these.

In Spark, you create UDF by creating a function in a language you prefer to use for Spark. For example, if you are using Spark with Scala, you create a UDF in Scala language and wrap it with udf() function or register it as udf to use it on DataFrame and SQL respectively.

Why do we need a Spark UDF?

UDF’s are used to extend the functions of the framework and re-use this function on several DataFrame. For example, if you wanted to convert the every first letter of a word in a sentence to capital case, Spark build-in features doesn't have this function, hence you can create it as UDF and reuse this as needed on many DataFrames. UDF’s are once created they can be re-used on several DataFrames and SQL expressions.

Before you create any UDF, do your research to check if the similar function you wanted is already available in Spark SQL Functions. Spark SQL provides several predefined common functions and many more new functions are added with every release. Hence, it is best to check before you reinvent the wheel.

When creating UDF’s, you need to design them very carefully otherwise you will come across performance issues.

Create a DataFrame

Before we jump into creating a UDF, first let’s create a Spark DataFrame.

import spark.implicits._

val columns = Seq("Seqno", "Quote")
val data = Seq(
  ("1", "Be the change that you wish to see in the world"),
  ("2", "Everyone thinks of changing the world, but no one thinks of changing himself."),
  ("3", "The purpose of our lives is to be happy.")
)

val df = data.toDF(columns:_*)
df.show(false)

Yields the below output.

+-----+-----------------------------------------------------------------------------+
|Seqno|Quote                                                                        |
+-----+-----------------------------------------------------------------------------+
|1    |Be the change that you wish to see in the world                              |
|2    |Everyone thinks of changing the world, but no one thinks of changing himself.|
|3    |The purpose of our lives is to be happy.                                     |
+-----+-----------------------------------------------------------------------------+

Create a Function

The first step in creating a UDF is creating a Scala function. Below snippet creates a function convertCase() which takes a string parameter and converts the first letter of every word to a capital letter. UDF’s take parameters of your choice and return a value.

val convertCase = (strQuote: String) => {
  val arr = strQuote.split(" ")
  arr.map(f => f.substring(0, 1).toUpperCase + f.substring(1, f.length)).mkString(" ")
}

Create Spark UDF to use it on DataFrame

Now convert this function convertCase() to UDF by passing the function to Spark SQL udf(), this function is available at org.apache.spark.sql.functions.udf package. Make sure you import this package before using it.

val convertUDF = udf(convertCase)

Now you can use convertUDF() on a DataFrame column. The udf() function returns org.apache.spark.sql.expressions.UserDefinedFunction.

// Using with DataFrame
df.select(col("Seqno"), convertUDF(col("Quote")).as("Quote")).show(false)

This results in the below output.

+-----+-----------------------------------------------------------------------------+
|Seqno|Quote                                                                        |
+-----+-----------------------------------------------------------------------------+
|1    |Be The Change That You Wish To See In The World                              |
|2    |Everyone Thinks Of Changing The World, But No One Thinks Of Changing Himself.|
|3    |The Purpose Of Our Lives Is To Be Happy.                                     |
+-----+-----------------------------------------------------------------------------+

Registering Spark UDF to use it on SQL

In order to use convertCase() function on Spark SQL, you need to register the function with Spark using spark.udf.register().

// Using it on SQL
spark.udf.register("convertUDF", convertCase)
df.createOrReplaceTempView("QUOTE_TABLE")
spark.sql("select Seqno, convertUDF(Quote) from QUOTE_TABLE").show(false)

This yields the same output as the previous example.

null check

UDF’s are error-prone when not designed carefully. For example, when you have a column that contains the value null on some records and not handling null inside a UDF function returns the below error.

Exception in thread "main" org.apache.spark.SparkException: Failed to execute user defined function(anonfun$1: (string) => string)
    at org.apache.spark.sql.catalyst.expressions.ScalaUDF.eval(ScalaUDF.scala:1066)
    at org.apache.spark.sql.catalyst.expressions.Alias.eval(namedExpressions.scala:152)
    at org.apache.spark.sql.catalyst.expressions.InterpretedMutableProjection.apply(Projection.scala:92)
    at org.apache.spark.sql.catalyst.optimizer.ConvertToLocalRelation$anonfun$apply$24$anonfun$applyOrElse$23.apply(Optimizer.scala:1364)
    at org.apache.spark.sql.catalyst.optimizer.ConvertToLocalRelation$anonfun$apply$24$anonfun$applyOrElse$23.apply(Optimizer.scala:1364)
    at scala.collection.TraversableLike$anonfun$map$1.apply(TraversableLike.scala:234)
    at scala.collection.TraversableLike$anonfun$map$1.apply(TraversableLike.scala:234)
    at scala.collection.immutable.List.foreach(List.scala:392)
    at scala.collection.TraversableLike$class.map(TraversableLike.scala:234)
    at scala.collection.immutable.List.map(List.scala:296)
    at org.apache.spark.sql.catalyst.optimizer.ConvertToLocalRelation$anonfun$apply$24.applyOrElse(Optimizer.scala:1364)
    at org.apache.spark.sql.catalyst.optimizer.ConvertToLocalRelation$anonfun$apply$24.applyOrElse(Optimizer.scala:1359)

It is always best practice to check for null inside a UDF function rather than checking for null outside.

Performance concern using UDF

UDF’s are a black box to Spark hence it can’t apply optimization and you will lose all the optimization Spark does on DataFrame/Dataset. When possible you should use Spark SQL built-in functions as these functions provide optimization.