Spark Scala：任務不可序列化錯誤

Question

我正在使用IntelliJ社區版與Scala插件和火花庫。我仍然在學習Spark，並且正在使用Scala Worksheet。

我已經寫了下面的代碼刪除標點符號的字符串：

def removePunctuation(text: String): String = { 
    val punctPattern = "[^a-zA-Z0-9\\s]".r 
    punctPattern.replaceAllIn(text, "").toLowerCase 
} 

然後我讀的文本文件，並嘗試刪除標點：

val myfile = sc.textFile("/home/ubuntu/data.txt",4).map(removePunctuation) 

這給瞭如下錯誤，任何幫助將不勝感激：

org.apache.spark.SparkException：任務不可序列化 at org.apache.spark.util.ClosureCleaner $ .ensureSerializable（/home/ubuntu/src/main/scala/Test.sc:294） at org.apache.spark.util.ClosureCleaner $ .org $ apache $ spark $ util $ ClosureCleaner $$ clean（/home/ubuntu/src/main/scala/Test.sc:284） at org.apache.spark.util.ClosureCleaner $ .clean（/ home/ubuntu/src/main/scala /Test.sc:104） at org.apache.spark.SparkContext.clean（/home/ubuntu/src/main/scala/Test.sc:2090） at org.apache.spark.rdd.RDD $$ anonfun $ map $ 1.apply（/home/ubuntu/src/main/scala/Test.sc:366） at org.apache.spark.rdd.RDD $$ anonfun $ map $ 1.apply（/ home/ubuntu/src /主頁/ scala/Test.sc：365） at org.apache.spark.rdd.RDDOperationScope $ .withScope（/home/ubuntu/src/main/scala/Test.sc:147） at #worksheet＃。＃工作表＃（/ home/ubuntu/src/main/scala/Test.sc：108） 導致：java.io.NotSerializableException：A $ A21 $ A $ A21 序列化堆棧： - 對象不可序列化（類：A $ A21 $ A $ A21，值：A $ A21 $ A $ A21 @ 62db3891） - 類（類：A $ A21 $ A $ A21 $$ anonfun $ words $ 1，名稱：$ outer，類型：class A $ A21 $ A $ A21） - 對象（class A $ A21 $ A $ A21 $$ anonfun 40） 在org.apache.spark.serializer.JavaSerializationStream.writeObject（JavaSerializer.scala：：46） 在組織在org.apache.spark.serializer.SerializationDebugger $ .improveException（SerializationDebugger.scala $字$ 1，） 。 apache.spark.serializer.JavaSerializerInstance.serialize（JavaSerializer.scala：100） at org.apache.spark.util.ClosureCleaner $ .ensureSerializable（ClosureCleaner.scala：295） at org.apache.spark.util.ClosureClean在$ org.apache.spark.util.ClosureCleaner $ .clean（ClosureCleaner.scala：108） at org.apache.spark .SparkContext.clean（SparkContext.scala：2094） at org.apache.spark.rdd.RDD $$ anonfun $ map $ 1.apply（RDD.scala：370） at org.apache.spark.rdd.RDD $$ anonfun $ map $ 1.apply（RDD.scala：369） 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：362） at org.apache.spark.rdd.RDD.map（RDD.scala：369） at A $ A21 $ A $ A21.words $ lzycompute（Test.sc:27） at A $ A21 $ A $ A21.words（Te st.sc:27） at A $ A21 $ A $ A21.get $$ instance $$ words（Test.sc:27） at A $ A21 $ .main（Test.sc:73） at A $ A21 。主要（Test.sc） 在sun.reflect.NativeMethodAccessorImpl.invoke0（本機方法） 在sun.reflect.NativeMethodAccessorImpl.invoke（NativeMethodAccessorImpl.java:62） 在sun.reflect.DelegatingMethodAccessorImpl.invoke（DelegatingMethodAccessorImpl.java： 43） at java.lang.reflect.Method.invoke（Method.java:498） at org.jetbrains.plugins.scala.worksheet.MyWorksheetRunner.main（MyWorksheetRunner。Java的：22）

Answer 1

由於@TGaweda表明，星火的SerializationDebugger是識別非常有幫助「的序列化路徑從給定的對象有問題的對象處於領先地位。」堆棧跟蹤中的「序列化堆棧」之前的所有美元符號表明您的方法的容器對象是問題所在。

雖然這是最簡單的，只是拍Serializable你的容器類，我更願意利用這樣的事實Scala是一種函數式語言，並使用你的函數作爲頭等公民：

sc.textFile("/home/ubuntu/data.txt",4).map { text => 
    val punctPattern = "[^a-zA-Z0-9\\s]".r 
    punctPattern.replaceAllIn(text, "").toLowerCase 
} 

或者，如果你真的希望保持獨立：

val removePunctuation: String => String = (text: String) => { 
    val punctPattern = "[^a-zA-Z0-9\\s]".r 
    punctPattern.replaceAllIn(text, "").toLowerCase 
} 
sc.textFile("/home/ubuntu/data.txt",4).map(removePunctuation) 

當然這些選項的工作，因爲Regexis serializable，你應該確認。

在次要的但非常重要的筆記中，構建Regex是昂貴的，所以爲了性能而將其排除在轉換之外 - 可能使用broadcast。

Answer 2

閱讀堆棧跟蹤，有：

$外，類型：A類$ A21 $ A $ A21

這是一個很好的提示。你的lambda是可序列化的，但你的類不是可序列化的。

當您創建lambda表達式時，則此表達式具有對外部類的引用。在你的情況下，外部類是不可序列化的，即沒有實現Serializable或其中一個字段不是可串行化的實例

Answer 3

正如T. Gaweda已經指出的，你很可能在一個不可序列化的類中定義函數。因爲它是一個純函數，即它不依賴於封閉類的任何上下文，所以我建議你將它放入一個伴隨對象，該對象應該擴展爲Serializable。這將是Scala的相當於Java靜態方法：

object Helper extends Serializable { 
    def removePunctuation(text: String): String = { 
    val punctPattern = "[^a-zA-Z0-9\\s]".r 
    punctPattern.replaceAllIn(text, "").toLowerCase 
    } 
}