In a recent project, I developed a metadata-driven data validation framework for Spark, utilizing both Scala and Python. After the initial excitement of creating the framework, I conducted a thorough review and discovered that the User Defined Functions (UDFs) I had crafted were prone to errors in specific situations.

To address this, I explored various methods to make the UDFs fail-safe. Let's start by examining the data, as shown below:

name,date,super-name,alien-name,sex,media-type,franchise,planet,alien,alien-planet,side-kick
peter parker,22/03/1970,spiderman,,m,comic,marvel,earth,n,none,none
clark kent,14/09/1985,superman,kal el,m,comic,dc,earth,y,krypton,
bruce wayne,12/12/2000,batman,,m,comic,dc,earth,n,,Robin
Natasha Romanoff,06/04/1982,black widow,,f,movie,marvel,earth,n,none,
Carol Susan Jane Danvers,1982-04-01,Captain Marvel,,f,comic,marvel,earth,n,none,

Next, let's read the data into a dataframe, as demonstrated below:

import org.apache.spark.sql.expressions.UserDefinedFunction
import org.apache.spark.sql.functions.{col, udf}

import spark.implicits._

val df = spark.read.format("csv").option("header", "true").option("inferSchema", "true").load("super-heroes.csv")
df.show

For this dataset, let's assume we want to verify if the superhero's name is "kal el". We'll implement this verification using a UDF.

Failsafe UDF Approach

The most straightforward method to achieve this is illustrated below:

def isAlienName(data: String): String = {
  if ( data.equalsIgnoreCase("kal el") ) {
    "yes"
  } else {
    "no"
  }
}

val isAlienNameUDF = udf(isAlienName _)

val df1 = df.withColumn("df1", isAlienNameUDF(col("alien-name")))
df1.show

When working with UDFs, it's essential to consider potential errors and develop strategies to mitigate them. For more information on writing fail-safe Scala Spark UDFs, check out this article on carsnewstoday.com.

When we leverage the isAlienNameUDF method, it operates flawlessly for all instances where the column value is not null. However, if the value of the cell passed to the UDF is null, it precipitates an exception: org.apache.spark.SparkException: Failed to execute user defined function

This arises because we are attempting to invoke the equalsIgnoreCase method on a null value.

Alternative Solution

To bypass the issue in the initial approach, we can modify the UDF as follows:

def isAlienName2(data: String): String = {
  if ( "kal el".equalsIgnoreCase(data) ) {
    "yes"
  } else {
    "no"
  }
}

val isAlienNameUDF2 = udf(isAlienName2 _)

val df2 = df.withColumn("df2", isAlienNameUDF2(col("alien-name")))
df2.show

Alternative C

Rather than incorporating null checks within the UDF or rewriting the UDF code to circumvent a NullPointerException, Spark offers a built-in method that enables null checks to be performed directly at the point of UDF execution, as illustrated below:val df4 = df.withColumn("df4", isAlienNameUDF2(when(col("alien-name").isNotNull,col("alien-name")).otherwise(lit("xyz")))) df4.show

In this scenario, we validate the column value. If the value is not null, we pass the column value to the UDF. Otherwise, we pass a default value to the UDF.

Alternative D

In alternative C, the UDF is invoked regardless of the column value. We can optimize this by rearranging the order of 'when' and 'otherwise', as follows:val df5 = df.withColumn("df5", when(col("alien-name").isNotNull, isAlienNameUDF2(col("alien-name"))).otherwise(lit("xyz"))) df5.show

In this alternative, the UDF is only invoked if the column value is not null. If the column value is null, we utilize a default value instead.

Conclusion

At this point, I am convinced that alternative D should be the preferred approach when designing a UDF.