Spark中的cache()会改变RDD的状态还是创建一个新的RDD？

这个问题是我之前问题的后续问题如果我在 Spark 中缓存同一个 RDD 两次会发生什么 https://stackoverflow.com/questions/36195105/what-happens-if-i-cache-the-same-rdd-twice-in-spark.

打电话时cache()在 RDD 上，RDD 的状态是否发生变化（返回的 RDD 只是this为了便于使用）或者创建一个新的 RDD 来包装现有的 RDD？

下面的代码会发生什么：

// Init
JavaRDD<String> a = ... // some initialise and calculation functions.
JavaRDD<String> b = a.cache();
JavaRDD<String> c = b.cache();

// Case 1, will 'a' be calculated twice in this case 
// because it's before the cache layer:
a.saveAsTextFile(somePath);
a.saveAsTextFile(somePath);

// Case 2, will the data of the calculation of 'a' 
// be cached in the memory twice in this case
// (once as 'b' and once as 'c'):
c.saveAsTextFile(somePath);

当在 RDD 上调用 cache() 时，RDD 的状态是否改变（以及 返回的 RDD 只是为了方便使用）或者创建一个新的 RDD 包裹现有的

相同RDD被返回 https://github.com/apache/spark/blob/master/core/src/main/scala/org/apache/spark/rdd/RDD.scala#L165:

/**
 * Mark this RDD for persisting using the specified level.
 *
 * @param newLevel the target storage level
 * @param allowOverride whether to override any existing level with the new one
 */
  private def persist(newLevel: StorageLevel, allowOverride: Boolean): this.type = {
  // TODO: Handle changes of StorageLevel
  if (storageLevel != StorageLevel.NONE && newLevel != storageLevel && !allowOverride) {
    throw new UnsupportedOperationException(
      "Cannot change storage level of an RDD after it was already assigned a level")
}
  // If this is the first time this RDD is marked for persisting, register it
  // with the SparkContext for cleanups and accounting. Do this only once.
  if (storageLevel == StorageLevel.NONE) {
    sc.cleaner.foreach(_.registerRDDForCleanup(this))
    sc.persistRDD(this)
  }
  storageLevel = newLevel
  this
}

缓存不会对上述 RDD 造成任何副作用。如果它已经被标记为持久化，则什么也不会发生。如果不是，唯一的副作用是将其注册到SparkContext，其中副作用不在RDD本身，而是上下文。

Edit:

看着JavaRDD.cache https://github.com/apache/spark/blob/master/core/src/main/scala/org/apache/spark/api/java/JavaRDD.scala#L38，看来底层调用会导致另一个的分配JavaRDD:

/** Persist this RDD with the default storage level (`MEMORY_ONLY`). */
def cache(): JavaRDD[T] = wrapRDD(rdd.cache())

Where wrapRDD calls JavaRDD.fromRDD:

object JavaRDD {

  implicit def fromRDD[T: ClassTag](rdd: RDD[T]): JavaRDD[T] = new JavaRDD[T](rdd)
  implicit def toRDD[T](rdd: JavaRDD[T]): RDD[T] = rdd.rdd
}

这将导致分配一个新的JavaRDD。也就是说，内部实例RDD[T]将保持不变。