Spark MLLib自定义Transformer加载已保存对象失败求助

错误原因分析 #

这个错误是因为Spark的DefaultParamsReader在加载持久化的Transformer时，会通过反射调用带单个String类型参数（uid）的构造函数，但在Notebook环境中，自定义类的构造函数可能因为以下原因无法被反射正确访问：

• Notebook代码块中定义的类属于局部/匿名类，类的全限定名在反射时无法被正确识别
• Notebook的类加载器上下文变化，导致反射找不到目标构造函数

解决方案 #

方案1：将自定义Transformer打包为Jar（最可靠） #

这是生产环境和Notebook环境通用的标准解决方案：

1. 在IDE中创建Scala项目，将MyTransformer定义为顶级类，并放在明确的包路径下（比如com.yourcompany.ml.transformers）：

package com.yourcompany.ml.transformers

import org.apache.spark.ml.UnaryTransformer
import org.apache.spark.ml.param.DoubleParam
import org.apache.spark.ml.util.{DefaultParamsReadable, DefaultParamsWritable, Identifiable}
import org.apache.spark.sql.types.{DataTypes, DataType}

class MyTransformer(override val uid: String)
  extends UnaryTransformer[Double, Double, MyTransformer] with DefaultParamsWritable {

  final val shift: DoubleParam = new DoubleParam(this, "shift", "Value added to input")

  def getShift: Double = $(shift)

  def setShift(value: Double): this.type = set(shift, value)

  def this() = this(Identifiable.randomUID("myT"))

  override protected def createTransformFunc: Double => Double = (input: Double) => {
    input + $(shift)
  }

  override protected def outputDataType: DataType = DataTypes.DoubleType

  override protected def validateInputType(inputType: DataType): Unit = {
    require(inputType == DataTypes.DoubleType, s"Bad input type: $inputType. Requires Double.")
  }
}

object MyTransformer extends DefaultParamsReadable[MyTransformer]

2. 编译项目生成Jar文件（比如my-transformers.jar）
3. 在Notebook中上传该Jar并添加到依赖库
4. 在Notebook中导入类并执行测试代码：

import com.yourcompany.ml.transformers.MyTransformer
import org.apache.spark.sql.functions.col

val myTransformer = new MyTransformer()
  .setShift(0.5)
  .setInputCol("input")
  .setOutputCol("output")

val data = spark.range(0, 5).toDF("input")
  .select(col("input").cast("double").as("input"))

val result = myTransformer.transform(data)
println("Transformed by adding constant value")
result.show()

val dirName = "s3://xxxx/temp/myTransformer/"
myTransformer.write.overwrite().save(dirName)
// 现在可以正常加载
val loadedTransformer = MyTransformer.load(dirName)

方案2：Notebook内自定义MLReader实现（临时方案） #

如果不想打包Jar，可以修改伴生对象的MLReader实现，手动创建实例避免反射调用构造函数：

import org.apache.spark.ml.UnaryTransformer
import org.apache.spark.ml.param.DoubleParam
import org.apache.spark.ml.util.{DefaultParamsReader, MLReader, MLReadable, Identifiable}
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.types.{DataTypes, DataType}

class MyTransformer(override val uid: String)
  extends UnaryTransformer[Double, Double, MyTransformer] with DefaultParamsWritable {

  final val shift: DoubleParam = new DoubleParam(this, "shift", "Value added to input")

  def getShift: Double = $(shift)

  def setShift(value: Double): this.type = set(shift, value)

  def this() = this(Identifiable.randomUID("myT"))

  override protected def createTransformFunc: Double => Double = (input: Double) => {
    input + $(shift)
  }

  override protected def outputDataType: DataType = DataTypes.DoubleType

  override protected def validateInputType(inputType: DataType): Unit = {
    require(inputType == DataTypes.DoubleType, s"Bad input type: $inputType. Requires Double.")
  }
}

object MyTransformer extends MLReadable[MyTransformer] {
  override def read: MLReader[MyTransformer] = new MLReader[MyTransformer] {
    private val spark = SparkSession.getActiveSession.get
    override def load(path: String): MyTransformer = {
      // 加载元数据
      val metadata = DefaultParamsReader.loadMetadata(path, spark.sparkContext)
      // 手动创建实例，跳过反射调用构造函数
      val transformer = new MyTransformer(metadata.uid)
      // 加载并设置参数
      DefaultParamsReader.getAndSetParams(transformer, metadata)
      transformer
    }
  }
}

修改后再执行加载代码，即可避免构造函数找不到的问题。

方案3：检查Spark版本兼容性 #

确保你的Spark环境版本和官方示例的版本一致，部分旧版本Spark的DefaultParamsReadable实现可能存在兼容性问题，升级到对应版本可以解决问题。

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