Scalatest和Spark給「java.io.NotSerializableException：org.scalatest.Assertions $ AssertionsHelper」

Question

我測試時，Spark流應用程序以「com.holdenkarau.spark測試基地」和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' 的字符串值，它工作正常。

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

第一個和第二個版本有什麼不同？

在此先感謝！

Answer 1

問題不是delim（字符串）的類型，它是delim本身。

儘量不要在test()方法之外定義變量。如果您在test中定義delm，它應該可以工作。

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

現在，你可能會問爲什麼？那麼，當您從外部範圍引用delim時，Scala會嘗試將封閉對象class Test合併在一起。該對象包含對org.scalatest.Assertions$AssertionsHelper的引用，它不是Serializable（請參閱您的堆棧跟蹤）。