Changed checkpointer constructor not to take initial data.

Author: jkbradley

mllib/src/main/scala/org/apache/spark/mllib/impl/PeriodicCheckpointer.scala

@@ -51,37 +51,11 @@ import org.apache.spark.storage.StorageLevel
  *  - This class removes checkpoint files once later Datasets have been checkpointed.
  *    However, references to the older Datasets will still return isCheckpointed = true.
  *
- * Example usage:
- * {{{
- *  val (data1, data2, data3, ...) = ...
- *  val cp = new PeriodicCheckpointer(data1, dir, 2)
- *  data1.count();
- *  // persisted: data1
- *  cp.update(data2)
- *  data2.count();
- *  // persisted: data1, data2
- *  // checkpointed: data2
- *  cp.update(data3)
- *  data3.count();
- *  // persisted: data1, data2, data3
- *  // checkpointed: data2
- *  cp.update(data4)
- *  data4.count();
- *  // persisted: data2, data3, data4
- *  // checkpointed: data4
- *  cp.update(data5)
- *  data5.count();
- *  // persisted: data3, data4, data5
- *  // checkpointed: data4
- * }}}
- *
- * @param currentData Initial Dataset
  * @param checkpointInterval  Datasets will be checkpointed at this interval
  * @param sc  SparkContext for the Datasets given to this checkpointer
  * @tparam T  Dataset type, such as RDD[Double]
  */
 private[mllib] abstract class PeriodicCheckpointer[T](
-    var currentData: T,
     val checkpointInterval: Int,
     val sc: SparkContext) extends Logging {
 
@@ -94,10 +68,8 @@ private[mllib] abstract class PeriodicCheckpointer[T](
   /** Number of times [[update()]] has been called */
   private var updateCount = 0
 
-  update(currentData)
-
   /**
-   * Update [[currentData]] with a new Dataset. Handle persistence and checkpointing as needed.
+   * Update with a new Dataset. Handle persistence and checkpointing as needed.
    * Since this handles persistence and checkpointing, this should be called before the Dataset
    * has been materialized.
    *
@@ -124,15 +96,13 @@ private[mllib] abstract class PeriodicCheckpointer[T](
       var canDelete = true
       while (checkpointQueue.size > 1 && canDelete) {
         // Delete the oldest checkpoint only if the next checkpoint exists.
-        if (isCheckpointed(checkpointQueue.get(1).get)) {
+        if (isCheckpointed(checkpointQueue.head)) {
           removeCheckpointFile()
         } else {
           canDelete = false
         }
       }
     }
-
-    currentData = newData
   }
 
   /** Checkpoint the Dataset */

mllib/src/main/scala/org/apache/spark/mllib/impl/PeriodicGraphCheckpointer.scala

@@ -17,6 +17,7 @@
 
 package org.apache.spark.mllib.impl
 
+import org.apache.spark.SparkContext
 import org.apache.spark.graphx.Graph
 import org.apache.spark.storage.StorageLevel
 
@@ -47,7 +48,7 @@ import org.apache.spark.storage.StorageLevel
  * Example usage:
  * {{{
  *  val (graph1, graph2, graph3, ...) = ...
- *  val cp = new PeriodicGraphCheckpointer(graph1, dir, 2)
+ *  val cp = new PeriodicGraphCheckpointer(2, sc)
  *  graph1.vertices.count(); graph1.edges.count()
  *  // persisted: graph1
  *  cp.updateGraph(graph2)
@@ -68,18 +69,16 @@ import org.apache.spark.storage.StorageLevel
  *  // checkpointed: graph4
  * }}}
  *
- * @param initGraph  Initial graph
  * @param checkpointInterval Graphs will be checkpointed at this interval
  * @tparam VD  Vertex descriptor type
  * @tparam ED  Edge descriptor type
  *
  * TODO: Move this out of MLlib?
  */
 private[mllib] class PeriodicGraphCheckpointer[VD, ED](
-    initGraph: Graph[VD, ED],
-    checkpointInterval: Int)
-  extends PeriodicCheckpointer[Graph[VD, ED]](initGraph, checkpointInterval,
-    initGraph.vertices.sparkContext) {
+    checkpointInterval: Int,
+    sc: SparkContext)
+  extends PeriodicCheckpointer[Graph[VD, ED]](checkpointInterval, sc) {
 
   override def checkpoint(data: Graph[VD, ED]): Unit = data.checkpoint()
 

mllib/src/main/scala/org/apache/spark/mllib/impl/PeriodicRDDCheckpointer.scala

@@ -17,6 +17,7 @@
 
 package org.apache.spark.mllib.impl
 
+import org.apache.spark.SparkContext
 import org.apache.spark.rdd.RDD
 import org.apache.spark.storage.StorageLevel
 
@@ -47,7 +48,7 @@ import org.apache.spark.storage.StorageLevel
  * Example usage:
  * {{{
  *  val (rdd1, rdd2, rdd3, ...) = ...
- *  val cp = new PeriodicRDDCheckpointer(rdd1, dir, 2)
+ *  val cp = new PeriodicRDDCheckpointer(2, sc)
  *  rdd1.count();
  *  // persisted: rdd1
  *  cp.update(rdd2)
@@ -68,16 +69,15 @@ import org.apache.spark.storage.StorageLevel
  *  // checkpointed: rdd4
  * }}}
  *
- * @param initRDD  Initial RDD
  * @param checkpointInterval  RDDs will be checkpointed at this interval
  * @tparam T  RDD element type
  *
  * TODO: Move this out of MLlib?
  */
 private[mllib] class PeriodicRDDCheckpointer[T](
-    initRDD: RDD[T],
-    checkpointInterval: Int)
-  extends PeriodicCheckpointer[RDD[T]](initRDD, checkpointInterval, initRDD.sparkContext) {
+    checkpointInterval: Int,
+    sc: SparkContext)
+  extends PeriodicCheckpointer[RDD[T]](checkpointInterval, sc) {
 
   override def checkpoint(data: RDD[T]): Unit = data.checkpoint()
 

mllib/src/test/scala/org/apache/spark/mllib/impl/PeriodicGraphCheckpointerSuite.scala

@@ -34,7 +34,8 @@ class PeriodicGraphCheckpointerSuite extends SparkFunSuite with MLlibTestSparkCo
     var graphsToCheck = Seq.empty[GraphToCheck]
 
     val graph1 = createGraph(sc)
-    val checkpointer = new PeriodicGraphCheckpointer(graph1, 10)
+    val checkpointer =
+      new PeriodicGraphCheckpointer[Double, Double](10, graph1.vertices.sparkContext)
     graphsToCheck = graphsToCheck :+ GraphToCheck(graph1, 1)
     checkPersistence(graphsToCheck, 1)
 
@@ -55,7 +56,8 @@ class PeriodicGraphCheckpointerSuite extends SparkFunSuite with MLlibTestSparkCo
     var graphsToCheck = Seq.empty[GraphToCheck]
     sc.setCheckpointDir(path)
     val graph1 = createGraph(sc)
-    val checkpointer = new PeriodicGraphCheckpointer(graph1, checkpointInterval)
+    val checkpointer = new PeriodicGraphCheckpointer[Double, Double](
+      checkpointInterval, graph1.vertices.sparkContext)
     graph1.edges.count()
     graph1.vertices.count()
     graphsToCheck = graphsToCheck :+ GraphToCheck(graph1, 1)

mllib/src/test/scala/org/apache/spark/mllib/impl/PeriodicRDDCheckpointerSuite.scala

@@ -34,7 +34,7 @@ class PeriodicRDDCheckpointerSuite extends SparkFunSuite with MLlibTestSparkCont
     var rddsToCheck = Seq.empty[RDDToCheck]
 
     val rdd1 = createRDD(sc)
-    val checkpointer = new PeriodicRDDCheckpointer(rdd1, 10)
+    val checkpointer = new PeriodicRDDCheckpointer[Double](10, rdd1.sparkContext)
     rddsToCheck = rddsToCheck :+ RDDToCheck(rdd1, 1)
     checkPersistence(rddsToCheck, 1)
 
@@ -55,7 +55,7 @@ class PeriodicRDDCheckpointerSuite extends SparkFunSuite with MLlibTestSparkCont
     var rddsToCheck = Seq.empty[RDDToCheck]
     sc.setCheckpointDir(path)
     val rdd1 = createRDD(sc)
-    val checkpointer = new PeriodicRDDCheckpointer(rdd1, checkpointInterval)
+    val checkpointer = new PeriodicRDDCheckpointer[Double](checkpointInterval, rdd1.sparkContext)
     rdd1.count()
     rddsToCheck = rddsToCheck :+ RDDToCheck(rdd1, 1)
     checkCheckpoint(rddsToCheck, 1, checkpointInterval)