Spark Scala:任务不可序列化错误

Question

我正在使用带有Scala插件和spark库的IntelliJ社区版.我还在学习Spark并且正在使用Scala工作表.

我写了下面的代码,删除字符串中的标点符号:

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:org.apache.spark.util上的org.apache.spark.util.ClosureCleaner $ .ensureSerializable(/home/ubuntu/src/main/scala/Test.sc:294)中的任务不可序列化.ClosureCleaner $ .org $ apache $ spark $ util $ ClosureCleaner $$ clean(/home/ubuntu/src/main/scala/Test.sc:284)org.apache.spark.util.ClosureCleaner $ .clean(/ home /ubuntu/src/main/scala/Test.sc:104)在org.apache.spark的org.apache.spark.SparkContext.clean(/home/ubuntu/src/main/scala/Test.sc:2090). rdd.RDD $$ anonfun $ map $ 1.apply(/home/ubuntu/src/main/scala/Test.sc:366)org.apache.spark.rdd.RDD $$ anonfun $ map $ 1.apply(/ home /ubuntu/src/main/scala/Test.sc:365)在org.apache.spark.rdd.RDDOperationScope $ .withScope(/home/ubuntu/src/main/scala/Test.sc:147)的#worksheet# .#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,名称:$ 外部,类型:A $ A21 $ A $ A21) - org.apache.spark.serializer.SerializationDebugger $ .improveException(SerializationDebugger.scala:对象(A $ A21 $ A $ A21 $$ anonfun $ words $ 1). 40)org.apache.spark.serializer.JavaSerializationStream.writeObject(JavaSerializer.scala:46)atg.apache.spark.serializer.JavaSerializerInstance.serialize(JavaSerializer.scala:100)at org.apache.spark.util.ClosureCleaner $ .ensureSerializable(ClosureCleaner.scala:295)atg.apache.spark.util.ClosureCleaner $ .org $ apache $ spark $ util $ ClosureCleaner $$ clean(ClosureCleaner.scala:288)at org.apache.spark.util. closureCleaner $ .clean(ClosureCleaner.scala:108)org.apache.spark.SparkContext.clean(SparkContext.scala:2094)at org.apache.spark.rdd.RDD $$ anonfun $ map $ 1.apply(RDD.scala) :370)org.apache.spark.rdd.RDD $$ anonfun $ map $ 1.apply(RDD.scala:369)atg.apache.spark.rdd.RDDOperationScope $ .withScope(RDDOperationScope.scala:151)at org org.ap上的.apache.spark.rdd.RDDOperationScope $ .withScope(RDDOperationScope.scala:112)ache.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)A $ A21 $ A $ A21.words(Test.sc:27)A $ A21 $ A $ A21.get $$实例$$字(Test.sc:27)A $ A21 $.在Sun.reflect的sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)的sun.reflect.NativeMethodAccessorImpl.invoke0(本地方法)的A $ A21.main(Test.sc)处的main(Test.sc:73) .delegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)位于org.jetbrains.plugins.scala.worksheet.MyWorksheetRunner.main的java.lang.reflect.Method.invoke(Method.java:498)(MyWorksheetRunner.java:22)

Answer 1

正如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
  }
}

Answer 2

正如@TGaweda所建议的那样，Spark的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)

这些选项当然Regex 可以工作，因为您应该确认可序列化。

在第二点但很重要的一点上，构造a Regex的代价很高，因此为了性能起见，请将其排除在转换之外（可能是广播）。