PySpark调用Scala UDF报错：UDF类未实现任何UDF接口

错误原因分析 #

你的代码存在几个核心问题：

1. 注册方式不匹配：spark.udf.registerJavaFunction要求目标类必须实现Spark的Java UDF接口（如UDF2），但你用Scala的udf方法生成的是Scala专属的UserDefinedFunction对象，且传入的是Scala Object类，不符合接口要求。
2. Scala代码语法错误：
   • case class未指定类名，语法不合法；
   • UDF参数名重复（两个参数都叫param），编译会失败。

解决方案 #

方案一：实现Java UDF接口（推荐用于跨语言调用） #

修改Scala代码，实现对应参数个数的Java UDF接口：

package com.spark.udfexample

import org.apache.spark.sql.api.java.UDF2
import org.apache.spark.sql.Row
import java.util.List
import org.apache.spark.sql.types.{StringType, StructType}

// 实现UDF2接口，对应2个输入参数，返回值类型可根据实际需求调整
class Udf1 extends UDF2[String, List[Row], String] {
  override def call(inputStr: String, inputRows: List[Row]): String = {
    // 这里编写你的UDF业务逻辑
    // 示例：返回输入字符串和行列表的长度
    s"Input string: $inputStr, Rows count: ${inputRows.size()}"
  }
}

object test {
  // 修正case class的定义，添加类名
  case class MyCaseClass(a: Option[String], b: Option[String])
  val foo: StructType = new StructType().add("a", StringType).add("b", StringType)
}

打包Jar后，在PySpark中注册并使用：

# 注册UDF
spark.udf.registerJavaFunction("udf1", "com.spark.udfexample.Udf1")

# 测试调用
spark.sql("SELECT udf1('test_input', array(named_struct('a', 'val1', 'b', 'val2')))").show()

方案二：在Scala中注册UDF，PySpark直接使用 #

修改Scala代码，在Object中实现UDF注册逻辑：

package com.spark.udfexample

import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.functions.udf
import org.apache.spark.sql.Row
import org.apache.spark.sql.types.{StringType, StructType}

object UdfRegistrator {
  case class MyCaseClass(a: Option[String], b: Option[String])
  val foo: StructType = new StructType().add("a", StringType).add("b", StringType)

  def registerUdfs(spark: SparkSession): Unit = {
    import spark.implicits._
    // 修正参数名重复问题，定义UDF
    val udf1 = udf((inputStr: String, inputRows: Seq[Row]) => {
      // 编写你的业务逻辑
      s"Param1: $inputStr, Param2 size: ${inputRows.size}"
    })
    // 将UDF注册到Spark Catalog
    spark.udf.register("udf1", udf1)
  }
}

在PySpark中调用Scala的注册方法后直接使用：

from pyspark.sql import SparkSession

spark = SparkSession.builder.appName("ScalaUDFTest").getOrCreate()
# 调用Scala Object的注册方法
spark.sparkContext._jvm.com.spark.udfexample.UdfRegistrator.registerUdfs(spark._jsparkSession)

# 直接使用已注册的UDF
spark.sql("SELECT udf1('hello', array(named_struct('a', 'x', 'b', 'y')))").show()

关键注意事项 #

• 若使用registerJavaFunction，必须确保类实现了对应参数数量的UDFn接口（如UDF0无参数、UDF1一个参数，最多到UDF22）；
• Scala和Java集合需要兼容，比如Scala的Seq对应Java的List，在UDF中可通过scala.collection.JavaConverters转换；
• 打包Scala代码时需确保依赖的Spark版本与PySpark使用的版本一致。

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