Spark Scala字符串匹配自定义UDF运行空指针异常问题排查

问题复现 #

编写如下字符串匹配UDF代码：

import org.apache.spark.sql.functions.lit

val containsString  = (haystack:String, needle:String) =>{
    if (haystack.contains(needle)){
        1
    }
    else{
         0
    }
}

val containsStringUDF = udf(containsString _)

// 原代码存在括号书写笔误，已修正参数位置
val new_df = df.withColumn("nameContainsxyz", containsStringUDF($"name", lit("xyz")))

代码编译通过，但执行聚合查询时抛出错误：

new_df.groupBy("nameContainsxyz").sum().show()

报错堆栈信息如下：

Caused by: org.apache.spark.SparkException: Failed to execute user defined function($anonfun$1: (string, string) => int)
  at org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage6.agg_doAggregateWithKeys_0$(Unknown Source)
  at org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage6.processNext(Unknown Source)
  at org.apache.spark.sql.execution.BufferedRowIterator.hasNext(BufferedRowIterator.java:43)
  at org.apache.spark.sql.execution.WholeStageCodegenExec$$anonfun$15$$anon$2.hasNext(WholeStageCodegenExec.scala:655)
  at scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:409)
  at org.apache.spark.shuffle.sort.BypassMergeSortShuffleWriter.write(BypassMergeSortShuffleWriter.java:125)
  at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:99)
  at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:55)
  at org.apache.spark.scheduler.Task.run(Task.scala:121)
  at org.apache.spark.executor.Executor$TaskRunner$$anonfun$10.apply(Executor.scala:408)
  at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1405)
  at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:414)
  ... 3 more
Caused by: java.lang.NullPointerException
  at $anonfun$1.apply(<console>:41)
  at $anonfun$1.apply(<console>:40)
  ... 15 more

根因分析 #

Spark DataFrame的字符串列允许存储null值，当UDF处理的name列存在null值时，null会被传入UDF赋值给haystack参数，此时调用haystack.contains(needle)方法会直接触发空指针异常。

修复方案 #

在UDF逻辑中新增空值校验即可解决问题，修改后的UDF代码如下：

val containsString  = (haystack:String, needle:String) =>{
    // 空值场景默认返回0，可根据业务需求调整返回逻辑
    if (haystack == null || needle == null) {
        0
    } else if (haystack.contains(needle)) {
        1
    } else {
        0
    }
}

也可以直接使用Spark原生内置函数实现相同逻辑，避免自定义UDF的性能开销：

import org.apache.spark.sql.functions.{instr, when}
val new_df = df.withColumn("nameContainsxyz", when(instr($"name", "xyz") > 0, 1).otherwise(0))

内容的提问来源于stack exchange，提问作者Jin