Как загрузить собственный трансформатор в Spark 2.4

Question

Я пытаюсь создать собственный преобразователь в 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, который можно сохранить и загрузить?

Answer 1

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

Пытаясь найти источник проблемы, я посмотрел на 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]

Я думаю, что REPL-файлы и 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