Wiki xml 解析器 - org.apache.spark.SparkException：任务不可序列化答案

【问题标题】：Wiki xml parser - org.apache.spark.SparkException: Task not serializableWiki xml 解析器 - org.apache.spark.SparkException：任务不可序列化
【发布时间】：2015-04-27 20:57:49
【问题描述】：

我是 scala 和 spark 的新手，并且尝试了一些教程，这个教程来自 Advanced Analytics with Spark。以下代码应该可以工作：

import com.cloudera.datascience.common.XmlInputFormat
import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.io._
val path = "/home/petr/Downloads/wiki/wiki"
val conf = new Configuration()
conf.set(XmlInputFormat.START_TAG_KEY, "<page>")
conf.set(XmlInputFormat.END_TAG_KEY, "</page>")
val kvs = sc.newAPIHadoopFile(path, classOf[XmlInputFormat],
classOf[LongWritable], classOf[Text], conf)

val rawXmls = kvs.map(p => p._2.toString)

import edu.umd.cloud9.collection.wikipedia.language._
import edu.umd.cloud9.collection.wikipedia._

def wikiXmlToPlainText(xml: String): Option[(String, String)] = {
val page = new EnglishWikipediaPage()
WikipediaPage.readPage(page, xml)
if (page.isEmpty) None
else Some((page.getTitle, page.getContent))
}

val plainText = rawXmls.flatMap(wikiXmlToPlainText)

但它给了

scala> val plainText = rawXmls.flatMap(wikiXmlToPlainText)
org.apache.spark.SparkException: Task not serializable
at org.apache.spark.util.ClosureCleaner$.ensureSerializable(ClosureCleaner.scala:166)
at org.apache.spark.util.ClosureCleaner$.clean(ClosureCleaner.scala:158)
at org.apache.spark.SparkContext.clean(SparkContext.scala:1622)
at org.apache.spark.rdd.RDD.flatMap(RDD.scala:295)
...

在本地运行 Spark v1.3.0（我只加载了大约 21MB 的 wiki 文章，只是为了测试它）。

https://stackoverflow.com/search?q=org.apache.spark.SparkException%3A+Task+not+serializable 的所有内容都没有给我任何线索...

谢谢。

【问题讨论】：

你是如何运行这段代码的？这不是一个独立的、有效的 Scala 类。尝试使用 main 方法将其放入 object 定义中。

标签： scala serialization apache-spark

【解决方案1】：

试试

    import com.cloudera.datascience.common.XmlInputFormat
import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.io._
val path = "/home/terrapin/Downloads/enwiki-20150304-pages-articles1.xml-p000000010p000010000"
val conf = new Configuration()
conf.set(XmlInputFormat.START_TAG_KEY, "<page>")
conf.set(XmlInputFormat.END_TAG_KEY, "</page>")
val kvs = sc.newAPIHadoopFile(path, classOf[XmlInputFormat],
classOf[LongWritable], classOf[Text], conf)
val rawXmls = kvs.map(p => p._2.toString)


import edu.umd.cloud9.collection.wikipedia.language._
import edu.umd.cloud9.collection.wikipedia._


val plainText = rawXmls.flatMap{line =>
val page = new EnglishWikipediaPage()
WikipediaPage.readPage(page, line)
if (page.isEmpty) None
else Some((page.getTitle, page.getContent))
}

【讨论】：

【解决方案2】：

想到的第一个猜测是：您的所有代码都包装在定义SparkContext 的对象中。 Spark 尝试序列化此对象以将wikiXmlToPlainText 函数传输到节点。尝试用唯一的一个函数wikiXmlToPlainText创建不同的对象。

【讨论】：