Scala 测试 SparkException：任务不可序列化

Question

我是 Scala 和 Spark 的新手。写了一个简单的测试课，一整天都在这个问题上。 请找到以下代码

A.scala

class A(key :String) extends  Serializable {
     val this.key:String=key

     def getKey(): String = { return this.key}
}

B.Scala

class B(key :String) extends  Serializable {
     val this.key:String=key
     def getKey(): String = { return this.key}
 }

Test.scala

import com.holdenkarau.spark.testing.{RDDComparisons, SharedSparkContext}
import org.scalatest.FunSuite
import org.scalatest.BeforeAndAfter

class Test extends FunSuite with SharedSparkContext with RDDComparisons with BeforeAndAfter with Serializable {

  //comment this
  private[this] val b1 = new B("test1")

  test("Test RDD") {

    val a1 = new A("test1")
    val a2 = new A("test2")

    val expected= sc.parallelize(Seq(a1,a2))
    println(b1.getKey())
     //val b1 = new B("test1")
    //val key1 :String = b1.getKey()
    expected.foreach{ a =>
      //if(a.getKey().equalsIgnoreCase(key1))
        if(a.getKey().equalsIgnoreCase(b1.getKey()))
          print("hi")
    }
  }
}

这段代码抛出异常

Task not serializable
org.apache.spark.SparkException: Task not serializable
    at org.apache.spark.util.ClosureCleaner$.ensureSerializable(ClosureCleaner.scala:403)
    at org.apache.spark.util.ClosureCleaner$.org$apache$spark$util$ClosureCleaner$$clean(ClosureCleaner.scala:393)
    at org.apache.spark.util.ClosureCleaner$.clean(ClosureCleaner.scala:162)
    at org.apache.spark.SparkContext.clean(SparkContext.scala:2326)
    at org.apache.spark.rdd.RDD$$anonfun$foreach$1.apply(RDD.scala:926)
    at org.apache.spark.rdd.RDD$$anonfun$foreach$1.apply(RDD.scala:925)
    at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:151)
    at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:112)
    at org.apache.spark.rdd.RDD.withScope(RDD.scala:363)
    at org.apache.spark.rdd.RDD.foreach(RDD.scala:925)
    at com.adgear.adata.hhid.Test$$anonfun$1.apply$mcV$sp(Test.scala:19)
    at com.adgear.adata.hhid.Test$$anonfun$1.apply(Test.scala:11)
    at com.adgear.adata.hhid.Test$$anonfun$1.apply(Test.scala:11)
    at org.scalatest.OutcomeOf$class.outcomeOf(OutcomeOf.scala:85)
    at org.scalatest.OutcomeOf$.outcomeOf(OutcomeOf.scala:104)
    at org.scalatest.Transformer.apply(Transformer.scala:22)
    at org.scalatest.Transformer.apply(Transformer.scala:20)
    at org.scalatest.FunSuiteLike$$anon$1.apply(FunSuiteLike.scala:186)
    at org.scalatest.TestSuite$class.withFixture(TestSuite.scala:196)
    at org.scalatest.FunSuite.withFixture(FunSuite.scala:1560)
    at org.scalatest.FunSuiteLike$class.invokeWithFixture$1(FunSuiteLike.scala:183)
    at org.scalatest.FunSuiteLike$$anonfun$runTest$1.apply(FunSuiteLike.scala:196)
    at org.scalatest.FunSuiteLike$$anonfun$runTest$1.apply(FunSuiteLike.scala:196)
    at org.scalatest.SuperEngine.runTestImpl(Engine.scala:289)
    at org.scalatest.FunSuiteLike$class.runTest(FunSuiteLike.scala:196)
    at org.scalatest.FunSuite.runTest(FunSuite.scala:1560)
    at org.scalatest.FunSuiteLike$$anonfun$runTests$1.apply(FunSuiteLike.scala:229)
    at org.scalatest.FunSuiteLike$$anonfun$runTests$1.apply(FunSuiteLike.scala:229)
    at org.scalatest.SuperEngine$$anonfun$traverseSubNodes$1$1.apply(Engine.scala:396)
    at org.scalatest.SuperEngine$$anonfun$traverseSubNodes$1$1.apply(Engine.scala:384)
    at scala.collection.immutable.List.foreach(List.scala:392)
    at org.scalatest.SuperEngine.traverseSubNodes$1(Engine.scala:384)
    at org.scalatest.SuperEngine.org$scalatest$SuperEngine$$runTestsInBranch(Engine.scala:379)
    at org.scalatest.SuperEngine.runTestsImpl(Engine.scala:461)
    at org.scalatest.FunSuiteLike$class.runTests(FunSuiteLike.scala:229)
    at org.scalatest.FunSuite.runTests(FunSuite.scala:1560)
    at org.scalatest.Suite$class.run(Suite.scala:1147)
    at org.scalatest.FunSuite.org$scalatest$FunSuiteLike$$super$run(FunSuite.scala:1560)
    at org.scalatest.FunSuiteLike$$anonfun$run$1.apply(FunSuiteLike.scala:233)
    at org.scalatest.FunSuiteLike$$anonfun$run$1.apply(FunSuiteLike.scala:233)
    at org.scalatest.SuperEngine.runImpl(Engine.scala:521)
    at org.scalatest.FunSuiteLike$class.run(FunSuiteLike.scala:233)
    at com.adgear.adata.hhid.Test.org$scalatest$BeforeAndAfterAll$$super$run(Test.scala:7)
    at org.scalatest.BeforeAndAfterAll$class.liftedTree1$1(BeforeAndAfterAll.scala:213)
    at org.scalatest.BeforeAndAfterAll$class.run(BeforeAndAfterAll.scala:210)
    at com.adgear.adata.hhid.Test.run(Test.scala:7)

当我注释掉 b1 的类级别声明并在测试方法本身中使用声明时，“if(a.getKey().equalsIgnoreCase(b1.getKey()))”就可以了。如果我保留类级别 b1 定义，那么“if(a.getKey().equalsIgnoreCase(b1.getKey()))”会抛出异常。为了解决这个问题，我必须使用“//val key1 :String = b1.getKey()" and "//if(a.getKey().equalsIgnoreCase(key1))”然后它才能工作。

正如人们所看到的，A、B 和 Test 都实现了 Serializable，但我仍然得到这个异常。是什么导致了这个问题？

谢谢

Answer 1

将一个类声明为 Serializable 并不意味着它可以被序列化，除非它的所有字段也都是 Serializable。

由于您的 Test 类扩展了 Funsuite，它将有一个不可序列化的“assertionsHelper”字段。因此，当您在“forEach”方法中引用“b1”字段时，Spark 将尝试序列化 Test 实例及其所有字段（包括 assertionsHelper）。

如果您想避免这种情况，您必须在其他地方定义 b1（在测试方法范围或伴随对象中），或者在将 b1 包含在 forEach 函数中之前将其取消引用到新变量中：

val b1_ref = b1
expected.foreach { a =>
  if (a.getKey().equalsIgnoreCase(b1_ref.getKey()))
    print("hi")
}

PS：当您遇到序列化异常时，您通常可以访问日志中的“序列化堆栈”，它会准确告诉您是哪个对象导致了错误