我需要持久化不断更新的 RDD 吗？答案

【问题标题】：Do I need to persist a continuously updated RDD?我需要持久化不断更新的 RDD 吗？
【发布时间】：2019-07-18 00:24:15
【问题描述】：

我正在使用一个火花程序，它需要在循环中不断更新一些 RDD：

var totalRandomPath: RDD[String] = null
for (iter <- 0 until config.numWalks) {
  var randomPath: RDD[String] = examples.map { case (nodeId, clickNode) =>
    clickNode.path.mkString("\t")
  }

  for (walkCount <- 0 until config.walkLength) {
    randomPath = edge2attr.join(randomPath.mapPartitions { iter =>
      iter.map { pathBuffer =>
        val paths: Array[String] = pathBuffer.split("\t")

        (paths.slice(paths.size - 2, paths.size).mkString(""), pathBuffer)
      }
    }).mapPartitions { iter =>
      iter.map { case (edge, (attr, pathBuffer)) =>
        try {
          if (pathBuffer != null && pathBuffer.nonEmpty && attr.dstNeighbors != null && attr.dstNeighbors.nonEmpty) {
            val nextNodeIndex: PartitionID = GraphOps.drawAlias(attr.J, attr.q)
            val nextNodeId: VertexId = attr.dstNeighbors(nextNodeIndex)
            s"$pathBuffer\t$nextNodeId"
          } else {
            pathBuffer //add
          }
        } catch {
          case e: Exception => throw new RuntimeException(e.getMessage)
        }
      }.filter(_ != null)
    }
  }

  if (totalRandomPath != null) {
    totalRandomPath = totalRandomPath.union(randomPath)
  } else {
    totalRandomPath = randomPath
  }
}

在这个程序中，RDD totalRandomPath 和 randomPath 不断更新着大量的转换操作：join 和 mapPartitions。该程序将以操作collect 结束。

那么我需要坚持那些不断更新的 RDD(totalRandomPath, randomPath) 来加快我的 spark 程序吗？
而且我注意到这个程序在单节点机器上运行速度很快，但是在三节点集群上运行就变慢了，为什么会出现这种情况？

【问题讨论】：

标签： scala apache-spark hadoop rdd

【解决方案1】：

是的，您需要保留更新的 RDD 并取消保留旧的 RDD

var totalRandomPath:RDD[String] = spark.sparkContext.parallelize(List.empty[String]).cache()   
for (iter <- 0 until config.numWalks){

    // existing logic

    val tempRDD = totalRandomPath.union(randomPath).cache()
    tempRDD foreach { _ => } //this will trigger cache operation for tempRDD immediately  
    totalRandomPath.unpersist() //unpersist old RDD which is no longer needed
    totalRandomPath = tempRDD   // point totalRandomPath to updated RDD
}

【讨论】：

谢谢。但我只是想知道 spark 可以自动优化转换，为什么我们要引入像 foreach 这样的操作？只是为了触发缓存操作？我知道当 shuffle 发生时，spark 也会将结果写入磁盘。
为什么要引入像 foreach 这样的操作？只是为了触发缓存操作？ => 是的。如果我们不做foreach，它不会为每次迭代保留RDD，并且每次迭代都会一次又一次地触发整个计算。
spark 也会在 shuffle 发生时将结果写入磁盘 => 并非所有情况都如此。仅当数据在 shuffle 期间无法放入内存时，spark 才会将数据写入磁盘。
谢谢，但我还有一个问题。为什么我们需要持久化那些中间 RDD？如果我的内存足够，spark 会将中间 RDD 写入内存，那为什么需要持久化行为呢？
正确。如果您有足够的内存，则不必坚持。当应用程序内存不足时，它将删除未明确要求持久化的 RDD(totalRandomPath)，并在下次需要该 RDD(totalRandomPath) 时重新计算。