如何在UDF中执行spark.sql？解决空指针异常问题

问题描述 #

尝试在UDF（用户自定义函数）中执行spark.sql时，遇到java.lang.NullPointerException异常，相关Scala代码及报错信息如下：

代码示例 #

import org.apache.spark.sql.functions.udf
import org.apache.spark.sql.{Column, DataFrame, SparkSession}

// 定义UDF
def myUdf(spark: SparkSession): UserDefinedFunction = udf((col1: String, col2: String) => {
  // 执行SQL查询
  val result = spark.sql(s"SELECT 'Hello World!' as text")

  // 返回结果字符串
  result.toString()
})

// 在DataFrame转换中使用UDF
def transform(df: DataFrame, col1: Column, col2: Column): DataFrame = {
  df.withColumn("result", myUdf(spark)(col1, col2))
}

val res = transform(df, col("salary"), col("gender"))
res.show()

报错信息 #

22/12/07 11:10:32 ERROR Executor: Exception in task 0.0 in stage 1679.0 (TID 19329)
org.apache.spark.SparkException: Failed to execute user defined function($read$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$$$82b5b23cea489b2712a1db46c77e458$$$$w$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$Lambda$4802/591483562: (string, string) => string)
        at org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage1.processNext(Unknown Source)
        at org.apache.spark.sql.execution.BufferedRowIterator.hasNext(BufferedRowIterator.java:43)
        at org.apache.spark.sql.execution.WholeStageCodegenExec$$anon$1.hasNext(WholeStageCodegenExec.scala:755)
        at org.apache.spark.sql.execution.SparkPlan.$anonfun$getByteArrayRdd$1(SparkPlan.scala:345)
        at org.apache.spark.rdd.RDD.$anonfun$mapPartitionsInternal$2(RDD.scala:898)
        at org.apache.spark.rdd.RDD.$anonfun$mapPartitionsInternal$2$adapted(RDD.scala:898)
        at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52)
        at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:373)
        at org.apache.spark.rdd.RDD.iterator(RDD.scala:337)
        at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:90)
        at org.apache.spark.scheduler.Task.run(Task.scala:131)
        at org.apache.spark.executor.Executor$TaskRunner.$anonfun$run$3(Executor.scala:497)
        at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1439)
        at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:500)
        at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
        at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
        at java.lang.Thread.run(Thread.java:750)
Caused by: java.lang.NullPointerException
        at org.apache.spark.sql.SparkSession.sessionState$lzycompute(SparkSession.scala:154)
        at org.apache.spark.sql.SparkSession.sessionState(SparkSession.scala:152)
        at org.apache.spark.sql.SparkSession.$anonfun$sql$2(SparkSession.scala:616)
        at org.apache.spark.sql.catalyst.QueryPlanningTracker.measurePhase(QueryPlanningTracker.scala:111)
        at org.apache.spark.sql.SparkSession.$anonfun$sql$1(SparkSession.scala:616)
        at org.apache.spark.sql.SparkSession.withActive(SparkSession.scala:775)
        at org.apache.spark.sql.SparkSession.sql(SparkSession.scala:613)

问题原因 #

1. SparkSession无法序列化到Executor：Spark UDF的逻辑会分发到Executor节点执行，而SparkSession是Driver端的核心对象，无法被序列化传递到Executor。当UDF在Executor中调用spark.sql时，传入的SparkSession实例已变为null，直接触发空指针异常。
2. UDF设计定位不符：UDF的初衷是针对单条数据做轻量、快速的转换操作，不适合在其中执行分布式SQL查询——这种操作会触发大量分布式任务，严重拖慢性能，甚至导致集群资源耗尽。

解决方案 #

方案1：使用DataFrame关联操作替代UDF（推荐） #

如果需求是基于原数据关联其他查询结果，优先使用Spark原生的join或常量列操作，这能充分利用Spark的优化器，保证性能。

示例（固定查询结果场景）：

import org.apache.spark.sql.functions.{col, lit}
import org.apache.spark.sql.{Column, DataFrame, SparkSession}

// 提前在Driver端执行SQL获取结果
val staticResult = spark.sql("SELECT 'Hello World!' as text").first().getString(0)

// 直接给原DataFrame添加常量列
def transform(df: DataFrame, col1: Column, col2: Column): DataFrame = {
  df.withColumn("result", lit(staticResult))
}

val res = transform(df, col("salary"), col("gender"))
res.show()

如果需要根据原数据的字段动态关联其他表数据，直接使用join：

// 假设需要关联的表已经注册为临时视图
spark.sql("CREATE TEMP VIEW other_table AS SELECT id, value FROM some_source")

// 通过join关联，替代UDF内的SQL查询
val res = df.join(spark.table("other_table"), df("id") === spark.table("other_table")("id"), "left")

方案2：使用mapPartitions执行分区级查询（仅特殊场景使用） #

如果确实需要根据每条数据的参数动态执行SQL，可使用mapPartitions在分区级别执行逻辑。每个分区内可以获取有效的SparkSession，但需注意控制查询频率，避免性能问题。

示例：

import org.apache.spark.sql.{Row, SparkSession}

def transform(df: DataFrame): DataFrame = {
  df.rdd.mapPartitions { iter =>
    // 在分区内获取SparkSession（Executor端可用）
    val spark = SparkSession.builder().getOrCreate()
    iter.map { row =>
      val col1 = row.getAs[String]("salary")
      val col2 = row.getAs[String]("gender")
      // 根据字段动态构造SQL
      val resultDf = spark.sql(s"SELECT 'Hello $col1 $col2!' as text")
      val result = resultDf.first().getString(0)
      // 返回原字段加查询结果
      (col1, col2, result)
    }
  }.toDF("salary", "gender", "result")
}

val res = transform(df)
res.show()

注意事项 #

• 除非万不得已，不要在UDF或分区操作中执行SQL，这会绕过Spark的优化机制，引发严重性能问题。
• 若必须在Executor端执行SQL，尽量批量处理分区内的数据，减少查询次数。
• Executor端的SparkSession通过getOrCreate()获取，无需手动传递Driver端的实例。

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