修改Spark RDD foreach中的集合

Question

我正在尝试在迭代RDD的元素时向地图添加元素.我没有收到任何错误,但修改没有发生.

这一切都可以直接添加或迭代其他集合:

scala> val myMap = new collection.mutable.HashMap[String,String]
myMap: scala.collection.mutable.HashMap[String,String] = Map()

scala> myMap("test1")="test1"

scala> myMap
res44: scala.collection.mutable.HashMap[String,String] = Map(test1 -> test1)

scala> List("test2", "test3").foreach(w => myMap(w) = w)

scala> myMap
res46: scala.collection.mutable.HashMap[String,String] = Map(test2 -> test2, test1 -> test1, test3 -> test3)

但是当我尝试从RDD做同样的事情时:

scala> val fromFile = sc.textFile("tests.txt")
...
scala> fromFile.take(3)
...
res48: Array[String] = Array(test4, test5, test6)

scala> fromFile.foreach(w => myMap(w) = w)
scala> myMap
res50: scala.collection.mutable.HashMap[String,String] = Map(test2 -> test2, test1 -> test1, test3 -> test3)

我已经尝试在foreach之前打印地图的内容,以确保变量是相同的,并且它打印正确:

fromFile.foreach(w => println(myMap("test1")))
...
test1
test1
test1
...

我还在foreach代码中打印了地图的修改元素,并将其打印为已修改,但是当操作完成时,地图似乎未经修改.

scala> fromFile.foreach({w => myMap(w) = w; println(myMap(w))})
...
test4
test5
test6
...
scala> myMap
res55: scala.collection.mutable.HashMap[String,String] = Map(test2 -> test2, test1 -> test1, test3 -> test3)

将RDD转换为数组(collect)也可以正常工作:

fromFile.collect.foreach(w => myMap(w) = w)
scala> myMap
res89: scala.collection.mutable.HashMap[String,String] = Map(test2 -> test2, test5 -> test5, test1 -> test1, test4 -> test4, test6 -> test6, test3 -> test3)

这是一个上下文问题吗？我是否正在访问其他地方正在修改的数据副本？

Answer 1

在Spark集群(不是单个机器)上运行时会变得更清晰.RDD现在分布在几台机器上.当你打电话时foreach,你告诉每台机器如何处理它所拥有的RDD.如果你引用任何局部变量(如myMap),它们会被序列化并发送到机器,因此他们可以使用它.但什么都没有回来.所以你的原始副本myMap不受影响.

我认为这回答了你的问题,但显然你正在努力完成某些事情,而你却无法以这种方式实现目标.随意在这里或在单独的问题中解释你想要做什么,我会尽力帮助.