Ошибка десериализации Apache Spark Throwing при использовании метода take на RDD

Question

Я новичок в Spark, и я использую Scala 2.12.8 с Spark 2.4.0.Я пытаюсь использовать классификатор Random Forest в Spark MLLib.Я могу построить и обучить классификатор, и классификатор может предсказать, если я использую функцию first () в результирующем RDD.Однако, если я попытаюсь использовать функцию take (n), я получу довольно большой и уродливый след стека.Кто-нибудь знает, что я делаю не так?Ошибка возникает в строке: «.take (3)».Я знаю, что это первая эффективная операция, которую я выполняю на СДР, поэтому, если кто-нибудь сможет объяснить мне, почему он не работает и как это исправить, я был бы очень благодарен.

object ItsABreeze {
  def main(args: Array[String]): Unit = {
    val spark: SparkSession = SparkSession
      .builder()
      .appName("test")
      .getOrCreate()

    //Do stuff to file
    val data: RDD[LabeledPoint] = MLUtils.loadLibSVMFile(spark.sparkContext, "file.svm")

    // Split the data into training and test sets (30% held out for testing)
    val splits: Array[RDD[LabeledPoint]] = data.randomSplit(Array(0.7, 0.3))
    val (trainingData, testData) = (splits(0), splits(1))

    // Train a RandomForest model.
    // Empty categoricalFeaturesInfo indicates all features are continuous
    val numClasses = 4
    val categoricaFeaturesInfo = Map[Int, Int]()
    val numTrees = 3
    val featureSubsetStrategy = "auto"
    val impurity = "gini"
    val maxDepth = 5
    val maxBins = 32

    val model: RandomForestModel = RandomForest.trainClassifier(
      trainingData,
      numClasses,
      categoricaFeaturesInfo,
      numTrees,
      featureSubsetStrategy,
      impurity,
      maxDepth,
      maxBins
    )

    testData
      .map((point: LabeledPoint) => model.predict(point.features))
      .take(3)
      .foreach(println)

    spark.stop()
  }
}

Верхняя часть трассировки стека выглядит следующим образом:

java.io.IOException: unexpected exception type
    at java.io.ObjectStreamClass.throwMiscException(ObjectStreamClass.java:1736)
    at java.io.ObjectStreamClass.invokeReadResolve(ObjectStreamClass.java:1266)
    at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:2078)
    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.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.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:83)
    at org.apache.spark.scheduler.Task.run(Task.scala:121)
    at org.apache.spark.executor.Executor$TaskRunner$$anonfun$10.apply(Executor.scala:402)
    at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1360)
    at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:408)
    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)
Caused by: java.lang.reflect.InvocationTargetException
    at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
    at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
    at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
    at java.lang.reflect.Method.invoke(Method.java:498)
    at java.lang.invoke.SerializedLambda.readResolve(SerializedLambda.java:230)
    at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
    at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
    at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
    at java.lang.reflect.Method.invoke(Method.java:498)
    at java.io.ObjectStreamClass.invokeReadResolve(ObjectStreamClass.java:1260)
    ... 25 more
Caused by: java.lang.BootstrapMethodError: java.lang.NoClassDefFoundError: scala/runtime/LambdaDeserialize
    at ItsABreeze$.$deserializeLambda$(ItsABreeze.scala)
    ... 35 more
Caused by: java.lang.NoClassDefFoundError: scala/runtime/LambdaDeserialize
    ... 36 more
Caused by: java.lang.ClassNotFoundException: scala.runtime.LambdaDeserialize
    at java.net.URLClassLoader.findClass(URLClassLoader.java:382)
    at java.lang.ClassLoader.loadClass(ClassLoader.java:424)
    at java.lang.ClassLoader.loadClass(ClassLoader.java:357)

Answer 1

Код, который я пытался запустить, был слегка измененной версией классификационного примера на этой странице (из документации Spark Machine Learning Library).

Оба комментария в моем исходном вопросе были верны: я изменил версию Scala, которую я использовал, с 2.12.8 на 2.11.12, и я вернул Spark на 2.2.1, и код работал так же, как и он.было.

Для всех, кто наблюдает за этой проблемой и имеет право ответить на нее, есть следующий вопрос: Spark 2.4.0 утверждает, что имеет новую экспериментальную поддержку Scala 2.12.x.Много ли известных проблем с поддержкой 2.12.x?