我有一个形式为(String,(Int,Iterable[String]))的RDD。对于RDD中的每个条目，整数值(我称之为距离(最初设置为10。Iterable[String]中的每个元素在该RDD中都有自己的条目，在该条目中它充当密钥(因此，我们在单独的RDD条目中有Iterable[String]中每个元素的距离(。我的意图是：
1.如果列表(Iterable[String](包含元素"Bethan"，我会将其距离指定为1。
2.在此之后，我通过过滤创建了一个距离为1的所有关键点的列表
3.在此之后，我将RDD转换为一个新的RDD，如果它自己列表中的任何元素的距离为1，则它的距离值将更新为2。
我有以下代码：
val disOneRdd = disRdd.map(x=> {if(x._2._2.toList.contains("Bethan")) (x._1,(1,x._2._2)) else x})
var lst = disRdd.filter(x=> x._2._1 == 1).keys.collect
val disTwoRdd = disRdd.map(x=> {
var b:Boolean = false
loop.breakable{
for (str <- x._2._2)
if (lst.contains(str)) //checks if it contains element with distance 1
b = true
loop.break
}
if (b)
(x._1,(2,x._2._2))
else
(x._1,(10,x._2._2))
})
但当我运行它时，我会得到错误"任务不可序列化"。我该怎么做，还有更好的方法吗？
编辑
输入形式的RDD：
("abc",(10,List("efg","hij","klm")))
("efg",(10,List("jhg","Beethan","abc","ert")))
("Beethan",(0,List("efg","vcx","zse")))
("vcx",(10,List("czx","Beethan","abc")))
("zse",(10,List("efg","Beethan","nbh")))
("gvf",(10,List("vcsd","fdgd")))
...
列表中包含Beethan的每个元素的距离都应该为1。每个具有"距离为1的元素"(而不是Beethan(的元素都应该具有距离2。输出的形式为：
("abc",(2,List("efg","hij","klm")))
("efg",(1,List("jhg","Beethan","abc","ert")))
("Beethan",(0,List("efg","vcx","zse")))
("vcx",(1,List("czx","Beethan","abc")))
("zse",(1,List("efg","Beethan","nbh"))
("gvf",(10,List("vcsd","fdgd")))
...
错误消息：
[error] (run-main-0) org.apache.spark.SparkException: Task not serializable
org.apache.spark.SparkException: Task not serializable
at org.apache.spark.util.ClosureCleaner$.ensureSerializable(ClosureCleaner.scala:298)
at org.apache.spark.util.ClosureCleaner$.org$apache$spark$util$ClosureCleaner$$clean(ClosureCleaner.scala:288)
at org.apache.spark.util.ClosureCleaner$.clean(ClosureCleaner.scala:108)
at org.apache.spark.SparkContext.clean(SparkContext.scala:2037)
at org.apache.spark.rdd.RDD$$anonfun$map$1.apply(RDD.scala:366)
at org.apache.spark.rdd.RDD$$anonfun$map$1.apply(RDD.scala:365)
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:358)
at org.apache.spark.rdd.RDD.map(RDD.scala:365)
at Bacon$.main(Bacon.scala:86)
at Bacon.main(Bacon.scala)
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:498)
Caused by: java.io.NotSerializableException: scala.util.control.Breaks
Serialization stack:
- object not serializable (class: scala.util.control.Breaks, value: scala.util.control.Breaks@78426203)
- field (class: Bacon$$anonfun$15, name: loop$1, type: class scala.util.control.Breaks)
- object (class Bacon$$anonfun$15, <function1>)
val disOneRdd = disRdd.map(x=> {if(x._2._2.toList.contains("Bethan")) (x._1,(1,x._2._2)) else x})
var lst = disRdd.filter(x=> x._2._1 == 1).keys.collect
val disTwoRdd = disRdd.map(x=> {
var b:Boolean = x._._2.filter(y => lst.contains(y)).size() > 0
if (b)
(x._1,(2,x._2._2))
else
(x._1,(10,x._2._2))
})
或
import scala.util.control.Breaks._
val disOneRdd = disRdd.map(x=> {if(x._2._2.toList.contains("Bethan")) (x._1,(1,x._2._2)) else x})
var lst = disRdd.filter(x=> x._2._1 == 1).keys.collect
val disTwoRdd = disRdd.map(x=> {
var b:Boolean = false
breakable{
for (str <- x._2._2)
if (lst.contains(str)) //checks if it contains element with distance 1
b = true
break
}
if (b)
(x._1,(2,x._2._2))
else
(x._1,(10,x._2._2))
})
这两个版本都适用于我。问题出在无法序列化的loop.breakable上。老实说，我不知道这个构造的行为是否发生了变化，但在将loop.breakable替换为breakable之后，它就起作用了——也许API发生了一些变化。带过滤器的版本可能较慢，但避免了breakable 的问题
尽管有主要问题，lst应该是broadcasted变量-然而，我并没有把broadcastd变量放在这里，以提供尽可能简单的答案