"java.io.NotSerializableException: org.scalatest.Assertions$AssertionsHelper“

Question

我正在用"com.holdenkarau.spark-testing-base“和scalatest来测试一个星火流应用程序。

import com.holdenkarau.spark.testing.StreamingSuiteBase
import org.apache.spark.rdd.RDD
import org.scalatest.{ BeforeAndAfter, FunSuite }

class Test extends FunSuite with BeforeAndAfter with StreamingSuiteBase {

  var delim: String = ","

  before {
    System.clearProperty("spark.driver.port")
   }

  test(“This Fails“) {

    val source = scala.io.Source.fromURL(getClass.getResource(“/some_logs.csv"))
    val input = source.getLines.toList

    val rowRDDOut = Calculator.do(sc.parallelize(input))   //Returns DataFrame

    val report: RDD[String] = rowRDDOut.map(row => new String(row.getAs[String](0) + delim + row.getAs[String](1))

    source.close
  }
}

我得到了字段‘delim’的序列化异常：

org.apache.spark.SparkException: Task not serializable
[info]   at org.apache.spark.util.ClosureCleaner$.ensureSerializable(ClosureCleaner.scala:304)
[info]   at org.apache.spark.util.ClosureCleaner$.org$apache$spark$util$ClosureCleaner$$clean(ClosureCleaner.scala:294)
[info]   at org.apache.spark.util.ClosureCleaner$.clean(ClosureCleaner.scala:122)
[info]   at org.apache.spark.SparkContext.clean(SparkContext.scala:2055)
[info]   at org.apache.spark.rdd.RDD$$anonfun$map$1.apply(RDD.scala:324)
[info]   at org.apache.spark.rdd.RDD$$anonfun$map$1.apply(RDD.scala:323)
[info]   at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:150)
[info]   at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:111)
[info]   at org.apache.spark.rdd.RDD.withScope(RDD.scala:316)
[info]   at org.apache.spark.rdd.RDD.map(RDD.scala:323)
[info]   ...
[info]   Cause: java.io.NotSerializableException: org.scalatest.Assertions$AssertionsHelper
[info] Serialization stack:
[info]  - object not serializable (class: org.scalatest.Assertions$AssertionsHelper, value: org.scalatest.Assertions$AssertionsHelper@78b339fa)
[info]  - field (class: org.scalatest.FunSuite, name: assertionsHelper, type: class org.scalatest.Assertions$AssertionsHelper)

如果我将'delim‘替换为String值的位置，它可以正常工作。

val report: RDD[String] = rowRDDOut.map(row => new String(row.getAs[String](0) + “,” + row.getAs[String](1))

第一版和第二版有什么区别？

提前感谢！

Answer 1

问题不是delim (字符串)的类型，而是delim本身。

不要定义test()方法之外的变量。如果您在您的delm中定义了test，那么它应该可以工作。

test(“This Fails“) {
   val delim = ","
   ...
}

现在，你可以问为什么？那么，当您从外部范围引用delim时，Scala将尝试将封闭的对象class Test组合在一起。此对象包含对org.scalatest.Assertions$AssertionsHelper的引用，该引用是不可序列化的(请参见堆栈跟踪)。

Answer 2

我今天遇到了这个问题，即使在我将所有代码都移到测试中(如accepted answer中提到的)之后，错误仍然存在。

最后，发现我在代码中使用了错误的语法(编译器没有捕捉到)。在我的例子中，是这样的：

// Wrong
df.filter(x => x.id === y)

// Right
df.filter(x => x.id == y)