运行时评估函数与Spark数据集/ RDD不兼容

Question

我有一个火花应用程序。我的用例是允许用户定义一个类似于Record => Record的任意函数作为“规则”，它将应用于RDD /数据集的每个记录。

以下是代码：

    //Sample rows with Id, Name, DOB and address
    val row1 = "19283,Alan,1989-01-20,445 Mount Eden Road Mount Eden Auckland"
    val row2 = "15689,Ben,1989-01-20,445 Mount Eden Road Mount Eden Auckland"

    val record1 = new Record(
      new RecordMetadata(),
      row1,
      true
    )
    val record2 = new Record(
      new RecordMetadata(),
      row2,
      true
    )

    val inputRecsList = List(record1, record2)
    val inputRecs = spark.sparkContext.parallelize(inputRecsList)

    val rule = ScalaExpression(
      //Sample rule. A lambda (Record => Record)
      """
        | import model.Record
        | { record: Record => record }
      """.stripMargin

    val outputRecs = inputRecs.map(rule.transformation)

以下是“Record”和“RecordMetadata”以及“ScalaExpression”类的定义：

case class Record(
                   val metadata: RecordMetadata,
                   val row: String,
                   val isValidRecord: Boolean = true
                 ) extends Serializable

case class RecordMetadata() extends Serializable

case class ScalaExpression(function: Function1[Record, Record]) extends Rule {

  def transformation = function
}

object ScalaExpression{

  /**
    * @param Scala expression as a string
    * @return Evaluated result of type Function (Record => Record)
    */
  def apply(string: String) = {
    val toolbox = currentMirror.mkToolBox()
    val tree = toolbox.parse(string)
    val fn = toolbox.eval(tree).asInstanceOf[(Record => Record)] //Or Function1(Record, Record)
    new ScalaExpression(fn)
  }
}

上面的代码引发了一个神秘的异常：

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:2287)
    at java.io.ObjectStreamClass.setObjFieldValues(ObjectStreamClass.java:1417)
    at java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:2293)
    at java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:2211)
    at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:2069)
    at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1573)
    at java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:2287)
    at java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:2211)
    at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:2069)
    at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1573)
    at java.io.ObjectInputStream.readObject(ObjectInputStream.java:431)
    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:80)
    at org.apache.spark.scheduler.Task.run(Task.scala:109)
    at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:345)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
    at java.lang.Thread.run(Thread.java:748)

但是，如果直接在代码中定义规则，则代码很有效：val rule = ScalaExpression( {record: Record => record} )

如果将地图（具有运行时评估规则）应用于List而不是RDD / Dataset，则代码也可以正常工作。

已经卡住了一段时间试图让它工作。任何帮助，将不胜感激。

编辑：标记为此问题的“可能重复”是解决一个完全不同的问题。我的用例试图在运行时从用户获取规则（将一个记录转换为另一个记录的有效scala语句），并在尝试将规则应用于数据集的每个记录时导致序列化问题。

最好的祝福。

Answer 1

Spark JIRA有一个未解决的问题来解决这个问题 - SPARK-20525这个问题的原因是由于加载Spark UDF时spark类加载器不匹配。

解决方法是在解释器之后加载你的火花会话。请找到示例代码。你也可以参考我的github，例如SparkCustomTransformations

trait CustomTransformations extends Serializable {
  def execute(spark: SparkSession, df: DataFrame, udfFunctions: AnyRef*): DataFrame
}

// IMPORTANT spark session should be lazy evaluated
lazy val spark = getSparkSession

def getInterpretor: scala.tools.nsc.interpreter.IMain = {

  import scala.tools.nsc.GenericRunnerSettings
  import scala.tools.nsc.interpreter.IMain

  val cl = ClassLoader.getSystemClassLoader
  val conf = new SparkConf()
  val settings = new GenericRunnerSettings(println _)
  settings.usejavacp.value = true

  val intp = new scala.tools.nsc.interpreter.IMain(settings, new java.io.PrintWriter(System.out))
  intp.setContextClassLoader
  intp.initializeSynchronous

  intp
}

val intp = getInterpretor

val udf_str =
  """
    (str:String)=>{
      str.toLowerCase
    }
    """
val customTransStr =
  """
    |import org.apache.spark.SparkConf
    |import org.apache.spark.sql.{DataFrame, SparkSession}
    |import org.apache.spark.sql.functions._
    |
    |new CustomTransformations {
    |    override def execute(spark: SparkSession, df: DataFrame, func: AnyRef*): DataFrame = {
    |
    |      //reading your UDF
    |      val str_lower_udf = spark.udf.register("str_lower", func(0).asInstanceOf[Function1[String,String]])
    |
    |      df.createOrReplaceTempView("df")
    |      val df_with_UDF_cols = spark.sql("select a.*, str_lower(a.fakeEventTag) as customUDFCol1 from df a").withColumn("customUDFCol2", str_lower_udf(col("fakeEventTag")))
    |
    |      df_with_UDF_cols.show()
    |      df_with_UDF_cols
    |    }
    |}
  """.stripMargin

intp.interpret(udf_str)
var udf_obj = intp.eval(udf_str)

val eval = new com.twitter.util.Eval
val customTransform: CustomTransformations = eval[CustomTransformations](customTransStr)


val sampleSparkDF = getSampleSparkDF
val outputDF = customTransform.execute(spark, sampleSparkDF, udf_obj)

outputDF.printSchema()
outputDF.show()