Блог

Я пытаюсь создать пользовательский трансформатор в Spark 2.4.0. Сохранение его работает нормально. Однако, когда я пытаюсь загрузить его, я получаю следующую ошибку:

 java.lang.NoSuchMethodException: TestTransformer.<init>(java.lang.String)
  at java.lang.Class.getConstructor0(Class.java:3082)
  at java.lang.Class.getConstructor(Class.java:1825)
  at org.apache.spark.ml.util.DefaultParamsReader.load(ReadWrite.scala:496)
  at org.apache.spark.ml.util.MLReadable$class.load(ReadWrite.scala:380)
  at TestTransformer$.load(<console>:40)
  ... 31 elided
 

Это наводит меня на мысль, что он не может найти конструктор моего трансформатора, что на самом деле не имеет смысла для меня.

MCVE:

 import org.apache.spark.sql.{Dataset, DataFrame}
import org.apache.spark.sql.types.{StructType}
import org.apache.spark.ml.Transformer
import org.apache.spark.ml.param.ParamMap
import org.apache.spark.ml.util.{DefaultParamsReadable, DefaultParamsWritable, Identifiable}

class TestTransformer(override val uid: String) extends Transformer with DefaultParamsWritable{

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

    override def transform(df: Dataset[_]): DataFrame = {
        val columns = df.columns
        df.select(columns.head, columns.tail: _*)
    }

    override def transformSchema(schema: StructType): StructType = {
        schema
    }

    override def copy(extra: ParamMap): TestTransformer = defaultCopy[TestTransformer](extra)
}

object TestTransformer extends DefaultParamsReadable[TestTransformer]{

    override def load(path: String): TestTransformer = super.load(path)

}

val transformer = new TestTransformer("test")

transformer.write.overwrite().save("test_transformer")
TestTransformer.load("test_transformer")
 

Выполнение этого (я использую записную книжку Jupyter) приводит к вышеуказанной ошибке. Я попытался скомпилировать и запустить его как файл .jar, без разницы.

Что меня озадачивает, так это то, что эквивалентный код PySpark работает нормально:

 from pyspark.sql import SparkSession, DataFrame
from pyspark.ml import Transformer
from pyspark.ml.util import DefaultParamsReadable, DefaultParamsWritable

class TestTransformer(Transformer, DefaultParamsWritable, DefaultParamsReadable):

    def transform(self, df: DataFrame) -> DataFrame:
        return df

TestTransformer().save('test_transformer')
TestTransformer.load('test_transformer')
 

Как я могу создать пользовательский искровой трансформатор, который можно сохранить и загрузить?

Я могу воспроизвести вашу проблему в spark-shell.

Пытаясь найти источник проблемы, я изучил DefaultParamsReadable и DefaultParamsReader источники, и я мог видеть, что они используют отражение Java.

https://github.com/apache/spark/blob/v2.4.0/mllib/src/main/scala/org/apache/spark/ml/util/ReadWrite.scala

строки 495-496

 val instance =
    cls.getConstructor(classOf[String]).newInstance(metadata.uid).asInstanceOf[Params]
 

Я думаю, что реплики scala и отражение Java не являются хорошими друзьями.

Если вы запустите этот фрагмент (после вашего):

 new TestTransformer().getClass.getConstructors
 

вы получите следующий результат:

 res1: Array[java.lang.reflect.Constructor[_]] = Array(public TestTransformer($iw), public TestTransformer($iw,java.lang.String))
 

Это правда! TestTransformer.<init>(java.lang.String) не существует.

Я нашел 2 обходных пути,

1. Компиляция вашего кода с помощью sbt и создание jar, а затем включение в spark-shell с :require помощью, сработали для меня (вы упомянули, что пробовали jar, хотя я не знаю, как)
2. Вставка кода в spark-shell с :paste -raw помощью, также работала нормально. Я полагаю -raw , что это мешает REPL делать махинации с вашими классами. Смотрите: https://docs.scala-lang.org/overviews/repl/overview.html

Я не уверен, как вы можете адаптировать любой из них к Jupyter, но я надеюсь, что эта информация полезна для вас.

ПРИМЕЧАНИЕ: на самом деле я использовал spark-shell в spark 2.4.1