PySpark调用Scala UDF报错：apply方法不存在的解决咨询

一、错误原因 #

你遇到的Method apply([...JListWrapper]) does not exist错误，是因为直接实例化Java UDF类后调用apply的方式不正确。Java版的Spark UDF接口（UDF1）需要通过Spark的UDF注册机制来使用，而非直接调用类的方法。

二、修正步骤 #

1. 调整Scala UDF代码 #

保持原有的UDF1实现，同时优化代码逻辑并添加异常处理：

import org.apache.spark.sql.api.java.UDF1
import com.bahmanm.persianutils.DateConverter._

class PersianToGregorianUDF extends UDF1[String, String] {
  override def call(persianDateStr: String): String = {
    try {
      // 解析8位格式的波斯日期（YYYYMMDD）
      val year = persianDateStr.substring(0, 4)
      val month = persianDateStr.substring(4, 6)
      val day = persianDateStr.substring(6, 8)
      val persianDate = SimpleDate(s"$year/$month/$day")
      // 转换为公历并返回标准字符串格式
      persianToGregorian(persianDate).toString
    } catch {
      case _: Exception => null // 处理无效日期，返回空值避免任务崩溃
    }
  }
}

// 可选单例对象，简化后续实例化
object PersianToGregorianUDF {
  def apply(): PersianToGregorianUDF = new PersianToGregorianUDF()
}

2. 确认SBT配置正确性 #

你的build.sbt和plugins.sbt配置符合要求，只需确保：

• Scala版本（2.12.8）与Spark版本（3.1.2）匹配（Spark3.1.x兼容Scala2.12系列）
• sbt-assembly插件会自动将persianutils依赖打包进Jar

重新执行打包命令：

sbt clean assembly

3. 修正PySpark调用代码 #

方式一：注册后通过SQL/DataFrame API调用

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, expr

# 初始化SparkSession，加载Scala UDF Jar
spark = SparkSession.builder \
    .appName("PersianDateConverter") \
    .master("local[*]") \
    .config("spark.jars", "./UDFLib-assembly-0.1.jar") \
    .getOrCreate()

# 注册Scala UDF到Spark环境
spark.udf.register("persian_to_gregorian", spark._jvm.PersianToGregorianUDF())

# DataFrame API调用示例
df.select(
    col("persian_date"),
    expr("persian_to_gregorian(persian_date)").alias("gregorian_date")
).show()

# SQL调用示例
df.createOrReplaceTempView("date_table")
spark.sql("""
    SELECT persian_date, persian_to_gregorian(persian_date) as gregorian_date 
    FROM date_table
""").show()

方式二：直接封装为DataFrame函数调用

from pyspark.sql.column import Column, _to_java_column

def persian_to_gregorian(data: Column) -> Column:
    # 获取Scala UDF实例，调用其核心方法call而非apply
    scala_udf = spark._jvm.PersianToGregorianUDF()
    return Column(scala_udf.call(_to_java_column(data)))

# 使用示例
df.select(
    col("persian_date"),
    persian_to_gregorian(col("persian_date")).alias("gregorian_date")
).show()

三、关键注意事项 #

• Jar加载：如果有多个依赖Jar，用逗号分隔配置spark.jars，例如.config("spark.jars", "udf.jar,other.jar")
• 日志清理：删除Scala代码中的println，分布式环境下这些输出不会同步到PySpark控制台，还会增加不必要的开销
• 版本兼容：严格保持Scala、Spark、依赖库的版本匹配，避免出现兼容性问题

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