Removed duplicate unpersistRDD.

Author: tdas

core/src/main/scala/org/apache/spark/ContextCleaner.scala

@@ -98,13 +98,6 @@ private[spark] class ContextCleaner(sc: SparkContext) extends Logging {
     listeners += listener
   }
 
-  /** Unpersists RDD and remove all blocks for it from memory and disk. */
-  def unpersistRDD(rddId: Int, blocking: Boolean) {
-    logDebug("Unpersisted RDD " + rddId)
-    sc.env.blockManager.master.removeRdd(rddId, blocking)
-    sc.persistentRdds.remove(rddId)
-  }
-
   /** Register an object for cleanup. */
   private def registerForCleanup(objectForCleanup: AnyRef, task: CleanupTask) {
     referenceBuffer += new WeakReferenceWithCleanupTask(objectForCleanup, task)
@@ -136,7 +129,7 @@ private[spark] class ContextCleaner(sc: SparkContext) extends Logging {
   private def doCleanupRDD(rddId: Int) {
     try {
       logDebug("Cleaning RDD " + rddId)
-      unpersistRDD(rddId, false)
+      sc.unpersistRDD(rddId, false)
       listeners.foreach(_.rddCleaned(rddId))
       logInfo("Cleaned RDD " + rddId)
     } catch {

core/src/main/scala/org/apache/spark/SparkContext.scala

@@ -756,8 +756,7 @@ class SparkContext(
   /**
    * Unpersist an RDD from memory and/or disk storage
    */
-  private[spark] def unpersistRDD(rdd: RDD[_], blocking: Boolean = true) {
-    val rddId = rdd.id
+  private[spark] def unpersistRDD(rddId: Int, blocking: Boolean = true) {
     env.blockManager.master.removeRdd(rddId, blocking)
     persistentRdds.remove(rddId)
     listenerBus.post(SparkListenerUnpersistRDD(rddId))

core/src/main/scala/org/apache/spark/rdd/RDD.scala

@@ -158,7 +158,7 @@ abstract class RDD[T: ClassTag](
    */
   def unpersist(blocking: Boolean = true): RDD[T] = {
     logInfo("Removing RDD " + id + " from persistence list")
-    sc.unpersistRDD(this, blocking)
+    sc.unpersistRDD(this.id, blocking)
     storageLevel = StorageLevel.NONE
     this
   }