java - 如何在 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

我可以在 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 REPL 和 Java 反射不是好 friend 。

如果您运行此代码段(在您的之后):

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，然后使用 :require 将其包含在 spark-shell 中，为我工作(你提到你试过一个 jar ，但我不知道怎么做)

2. 使用 :paste -raw 将代码粘贴到 spark-shell 中，也工作得很好。我想 -raw防止 REPL 对您的类(class)进行恶作剧。请参阅:https://docs.scala-lang.org/overviews/repl/overview.html

我不确定您如何将这些中的任何一个应用到 Jupyter，但我希望此信息对您有用。

注意:我实际上在 spark 2.4.1 中使用了 spark-shell