如何在循环中覆盖 RDD答案

【问题标题】：How to overwrite a RDD in a loop如何在循环中覆盖 RDD
【发布时间】：2017-01-17 04:32:41
【问题描述】：

我对 Spark 和 Scala 非常陌生，我正在实现一个迭代算法来操作一个大图。假设在 for 循环中，我们有两个 RDD（rdd1 和 rdd2）并且它们的值被更新。例如：

for (i <- 0 to 5){
   val rdd1 = rdd2.some Transformations
   rdd2 = rdd1
}

所以基本上，在迭代 i+1 期间，rdd1 的值是根据它在迭代 i 时的值计算的。我知道 RDD 是不可变的，所以我不能真正为它们重新分配任何东西，但我只是想知道，我的想法是否可以实现。如果是这样，怎么做？非常感谢任何帮助。

谢谢，

更新：当我尝试这段代码时：

var size2 = freqSubGraphs.join(groupedNeighbours).map(y => extendFunc(y))

for(i <- 0 to 5){
    var size2 = size2.map(y=> readyForExpandFunc(y))
}
size2.collect()

它给了我这个错误：“递归变量 size2 需要类型” 不知道是什么意思

【问题讨论】：

第二个错误是因为您在循环中使用了var size2 = size2...。你不需要你已经声明的var。
非常感谢您的评论。即使删除 var 也没有帮助我，我仍然得到了错误。我使用了 foldRight，它帮助我修复了错误。
很高兴您的问题以另一种方式得到解决，但var 仍然是您发布的代码的问题。

标签： scala apache-spark

【解决方案1】：

为了完整起见，如果您希望您的代码更纯正地道，您可以使用foldRight 来避免使用可变的var：

val zeroRdd = freqSubGraphs.join(groupedNeighbours).map(y => extendFunc(y))
val size2 = (0 to 5).foldRight(zeroRdd) {
  (_, rdd) => rdd.map(y => readyForExpandFunc(y))
}

【讨论】：

【解决方案2】：

只需打开一个 spark-shell 并尝试一下：

scala> var rdd1 = sc.parallelize(List(1,2,3,4,5))
rdd1: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[0] at parallelize at <console>:24

scala> for( i <- 0 to 5 ) { rdd1 = rdd1.map( _ + 1 ) }

scala> rdd1.collect()
res1: Array[Int] = Array(7, 8, 9, 10, 11)

如您所见，它有效。

【讨论】：

【解决方案3】：

您访问RDD 上的数据的方式取决于它的结构。如果你想对单项中的数据进行一些计算，可以直接使用map：

val intRDD = spark.sparkContext.parallelize(Seq(1,2,3,4,5))
val multipliedBy10 = intRDD.map(myInteger=>myInteger*10)
print(multipliedBy10.collect.toList) // output: List(10, 20, 30, 40, 50)

如果您的RDD 包含多个值（即：一个元组），您可以这样做：

val tupleRDD = spark.sparkContext.parallelize(Seq(('A', 1), ('B', 2), ('C', 3)))
val concatTuple = tupleRDD.map(tuple=>tuple._1 + "-" + tuple._2)
print(concatTuple.collect.toList) // output: List(A-1, B-2, C-3)

如果您还需要来自另一个 RDD 的数据来进行计算，我建议您先加入 RDD 的数据

【讨论】：