PySpark / Spark SQL UDF (User Defined Functions) – Learn UDFs In 5 Mins!

In this post I will dive into topic: Spark SQL" UDF (User" Defined Functions). I will provide you the introduction and use cases for UDFs.

Introduction

What Spark SQL UDFs Are And Why They Are Useful

Spark SQL" UDFs (User" Defined Functions) are user-defined functions that may be used to execute specific actions on data within Spark" SQL" queries. Users write these functions in Scala, Java", or Python" and then register them with Spark SQL" to be utilised in Spark SQL" queries.

Brief Overview Of Spark SQL And Its Use Cases

Spark SQL is an Apache Spark" framework plugin that allows users to conduct SQL" queries on large-scale datasets. Spark SQL" makes use of Spark’s distributed processing capabilities to quickly and effectively conduct SQL" queries on huge datasets. Unfortunately, the built-in methods offered by Spark SQL" have several restrictions that can make doing sophisticated data transformations or calculations difficult.

This is where UDFs can help. UDFs may be used to implement custom functions that Spark SQL" does not have by default". Users can construct their own functions with UDFs to conduct more complicated computations, data manipulations, and transformations not achievable with built-in functions.

Think about the situation when you have a dataset" with product pricing in several currencies. You want to change the pricing to all be in the same currency, but Spark SQL" doesn’t provide a built-in method for that. In this scenario, you may develop a UDF and register it with Spark SQL" to carry out the currency conversion. The pricing might then be converted using the UDF in your SQL" queries.

Creating And Registering UDFs In Spark SQL

Scala", Java", or Python" may be used to develop UDFs, and the DataFrame" API or the SQL" API can be used to register them with Spark SQL". When registered with Spark SQL", UDFs can additionally receive input parameters and have their output types specified.

How To Create a UDF Using Scala Or Python

You must specify the function that you intend to utilise as a UDF before you can construct one in Spark SQL using Scala" or Python". A value should be returned by the function after accepting one or more input parameters. The function must also be deterministic, which means that given a constant input, it must always result in the same output.

Spark UDF Scala

def add(a: Int, b: Int): Int = {
  a + b
}

The UDF must then be created using Spark Python" UDF or Spark" Scala" UDF function. The function you generated in step 1 is sent to the udf function, which creates a new function that can be used as a UDF in Spark SQL" queries.

import org.apache.spark.sql.functions.udf

val addUDF = udf((a: Int, b: Int) => add(a, b))

Lastly, you must use the register function to register the Spark UDF with Spark SQL". The UDF can be registered as a temporary function to be used just in the current SparkSession" or as a global function to be used in all SparkSessions.

spark.udf.register("add", addUDF)

Spark UDF Python

from pyspark.sql.functions import udf

def add(a, b):
  return a + b

addUDF = udf(add)
spark.udf.register("add", addUDF)

The UDF may be used in Spark SQL" queries exactly like any other built-in function after it has been registered. For instance, you may use the add UDF in the following SQL" query:

Using UDFs in Spark SQL Queries

Please find the usage of Spark UDF in SQL" query:

SELECT add(1, 2) as result

How To Use UDFs With DataFrame API

Here is an illustration of how to utilise Spark’s UserDefinedFunction (UDF) in conjunction with the DataFrame" API:

Consider the situation where we wish to develop a Spark UDF to convert a string column" to uppercase. The UDF may be defined in Scala" as follows:

import org.apache.spark.sql.functions._
import org.apache.spark.sql.types._

val toUpperCaseUDF = udf((s: String) => s.toUpperCase)

Here, we’re going to build a Spark UDF that accepts a String column and produces an uppercase version of the string using the udf method from the org.apache.spark.sql.functions package.

Create an example DataFrame" with one string column" next.

val data = Seq(("foo"), ("bar"), ("baz"))
val df = data.toDF("col1")
df.show()

This will output:

+----+
|col1|
+----+
| foo|
| bar|
| baz|
+----+

Now, let’s apply UDF to the DataFrame":

val resultDF = df.withColumn("col1_uppercase", toUpperCaseUDF(col("col1")))
resultDF.show()

This will output:

+----+--------------+
|col1|col1_uppercase|
+----+--------------+
| foo|           FOO|
| bar|           BAR|
| baz|           BAZ|
+----+--------------+

Here, we’re adding a new column to the DataFrame" with uppercase values using the withColumn" function. The first parameter is the name of the new column, and the second argument is the UDF that was applied to the first column.

It should be noted that the syntax for creating a Spark UDF differs significantly in Python":

from pyspark.sql.functions import udf
from pyspark.sql.types import StringType

to_upper_case_udf = udf(lambda s: s.upper(), StringType())

The remaining code is identical to that in the Scala" example.

Spark UDF Performance Considerations

There are a few performance factors to bear in mind while utilising user-defined functions (UDFs) with Spark SQL":

1. Serialization and Deserialization Overhead: When dealing with complicated types, UDFs might add extra serialisation and deserialization complexity. To eliminate this cost when working with complicated types, it is preferable to utilise Spark’s built-in functions rather than UDFs.
2. Memory Overhead: Moreover, UDFs may cause memory cost, particularly if they generate new objects or arrays. Use objects and arrays as much as you can to save memory usage.
3. Vectorization: Performance is improved by Spark’s built-in functions being vectorized, which allows them to operate on whole arrays or columns at once. UDFs, on the other hand, work with a single element at a time and are not vectorized. Use vectorized functions rather than UDFs if performance is an issue.
4. Broadcast Variables: Instead of loading the data for each job, you should think about utilising broadcast variables if your Spark UDF has to access any other data or resources. Each node has a cache of broadcast variables that may be accessible by all jobs, minimising the need for data transfers between nodes.
5. Code Generation: For some UDFs, Spark can provide efficient bytecode, improving performance. The spark.sql.codegen.wholeStage configuration parameter can be set to true to allow code generation. You should benchmark your UDFs to determine if this option improves speed, although not all UDFs can be code-generated.
6. Using SQL" Functions: Use Spark’s built-in SQL" functions wherever you can instead of UDFs since they are more performant and have higher levels of optimization. UDFs should generally only be used when performing specialised logic that cannot be described using built-in functions.
7. Data Partitioning: Consider splitting the data if your UDF uses a huge DataFrame" to increase speed. Due to partitioning, Spark can analyse data in parallel while minimising the cost associated with data shuffles.

Tips For Optimizing UDF Performance in Spark SQL

Here are some pointers for improving Spark SQL’s UDF performance:

1. Use Simple Types: In general, basic type UDFs (such as String, Double, and Int) work more quickly than complicated type UDFs (e.g., StructType, ArrayType). Use basic types whenever you can in your Spark UDFs to boost speed.
2. Avoid Object Instantiation: Performance can be slowed down and memory overhead introduced by UDFs that create new objects or arrays. Use objects and arrays as much as you can to save memory usage.
3. Use Immutable Data Structures: In Spark, immutable data structures (such as List and Map) often perform more quickly than changeable ones (such as ArrayBuffer and HashMap). Consider utilising an immutable data structure if your UDF has to generate a data structure in order to increase speed.
4. Use Broadcast Variables: Instead of loading the data for each job, think about utilising broadcast variables if your Spark UDF has to access other data or services. Each node has a cache of broadcast variables that may be accessible by all jobs, minimising the need for data transfers between nodes.
5. Use Vectorized UDFs: UDFs may be vectorized starting with Spark 2.3, which allows them to work on full arrays or columns at once and improve speed. Use the Python" pandas udf API or the Scala" VectorizedUDF API to enable vectorization for your UDF.
6. Use Code Generation: Certain UDFs can have their bytecode generated by Spark in an optimal manner, improving performance. You may turn on code generation by setting the configuration option spark.sql.codegen.wholeStage to true. The performance of your UDFs should be benchmarked to determine if this option improves it because not all UDFs can be code-generated.
7. Use SQL" Built-In Functions If possible
8. Data Partitioning: Consider splitting the data if your UDF uses a huge DataFrame" to increase speed. Due to partitioning, Spark can analyse data in parallel while minimising the cost associated with data shuffles.

Summary

In essence, Spark UDFs (User-Defined Functions) let users add their own unique functions to the built-in capability of Spark. For usage in SQL" queries and DataFrame" transformations, UDFs may be registered with Spark SQL" and created in both Scala" and Python". When implementing custom logic that cannot be represented using built-in functions, UDFs are helpful, but they can cause performance overhead and should be properly tuned.

Users can use simple types, avoid object instantiation, use immutable data structures, avoid broadcast variables, think about vectorized UDFs, use code generation, use SQL" functions when appropriate, think about data partitioning, benchmark their UDFs to find performance bottlenecks, and more to improve Spark UDF performance.

Users may take full use of Spark UDFs’ capabilities to carry out complicated data transformations and analysis in a distributed, scalable, and effective way by adhering to these best practises.

Could You Please Share This Post? 
I appreciate It And Thank YOU! :)
Have A Nice Day!