通过对字符串的反射定义 spark udf

Question

我正在尝试从包含 scala 函数的字符串中在 spark(2.0) 中定义一个 udf definition.Here 是片段：

val universe: scala.reflect.runtime.universe.type = scala.reflect.runtime.universe
import universe._
import scala.reflect.runtime.currentMirror
import scala.tools.reflect.ToolBox
val toolbox = currentMirror.mkToolBox()
val f = udf(toolbox.eval(toolbox.parse("(s:String) => 5")).asInstanceOf[String => Int])
sc.parallelize(Seq("1","5")).toDF.select(f(col("value"))).show

这给了我一个错误：

  Caused by: java.lang.ClassCastException: cannot assign instance of scala.collection.immutable.List$SerializationProxy to field org.apache.spark.rdd.RDD.org$apache$spark$rdd$RDD$$dependencies_ of type scala.collection.Seq in instance of org.apache.spark.rdd.MapPartitionsRDD
   at java.io.ObjectStreamClass$FieldReflector.setObjFieldValues(ObjectStreamClass.java:2133)
   at java.io.ObjectStreamClass.setObjFieldValues(ObjectStreamClass.java:1305)
   at java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:2024)
   at java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:1942)
   at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:1808)
   at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1353)
   at java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:2018)
   at java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:1942)
   at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:1808)
   at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1353)
   at java.io.ObjectInputStream.readObject(ObjectInputStream.java:373)
   at org.apache.spark.serializer.JavaDeserializationStream.readObject(JavaSerializer.scala:75)
   at org.apache.spark.serializer.JavaSerializerInstance.deserialize(JavaSerializer.scala:114)
   at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:66)
   at org.apache.spark.scheduler.Task.run(Task.scala:85)
   at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:274)
   at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
   at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
   at java.lang.Thread.run(Thread.java:745)

然而，当我将 udf 定义为：

val f = udf((s:String) => 5)

它工作得很好。这里有什么问题？最后 objective 是获取一个具有 scala 函数定义的字符串并将其用作 udf。

Answer 1

我有同样的错误，它没有显示 ClassNotFoundException，因为 JavaDeserializationStream class 正在捕获异常，根据您的环境，它失败是因为找不到 class 你正试图从你的 RDD/DataSet 执行，但它没有显示 ClassNotFoundError 。为了解决这个问题，我必须生成一个包含项目中所有 classes 的 jar（包括函数和依赖项），并将该 jar 包含在 spark 环境中

这适用于独立集群

conf.setJars ( Array ("/fullpath/yourgeneratedjar.jar", "/fullpath/otherdependencies.jar") )

这是纱线簇

conf.set("spark.yarn.jars", "/fullpath/yourgeneratedjar.jar,/fullpath/otherdependencies.jar")

Answer 2

正如 Giovanny 所观察到的，问题在于 class 加载程序不同（您可以通过在任何对象上调用 .getClass.getClassLoader 来进一步研究这一点）。然后，当工作人员尝试反序列化您的反射函数时，一切都变得一团糟。

这是一个不涉及任何 class 加载程序黑客的解决方案。这个想法是将反思步骤转移给工人。我们最终将不得不重做反射步骤，但每个工人只需一次。我认为这是非常理想的——即使您只在主节点上进行了一次反射，您也必须为每个工作人员做相当多的工作才能让他们识别该功能。

val f = udf {
  new Function1[String,Int] with Serializable {
    import scala.reflect.runtime.universe._
    import scala.reflect.runtime.currentMirror
    import scala.tools.reflect.ToolBox

    lazy val toolbox = currentMirror.mkToolBox()
    lazy val func = {
      println("reflected function") // triggered at every worker
      toolbox.eval(toolbox.parse("(s:String) => 5")).asInstanceOf[String => Int]
    }

    def apply(s: String): Int = func(s)
  }
}

然后，调用 sc.parallelize(Seq("1","5")).toDF.select(f(col("value"))).show 就可以了。

请随意注释掉 println - 这只是一种计算反射发生次数的简单方法。在 spark-shell --master 'local' 中只有一次，但在 spark-shell --master 'local[2]' 中是两次。

工作原理

UDF 会立即得到评估，但它在到达工作节点之前永远不会被使用，因此惰性值 toolbox 和 func 只会在工作节点上得到评估。此外，由于它们很懒惰，因此每个工人只对它们进行一次评估。

通过对字符串的反射定义 spark udf

Define spark udf by reflection on a String

scala

apache-spark

spark-dataframe

udf

scala-reflect

工作原理