Disable checkpoint in PeriodicCheckpointer

Author: yanboliang

mllib/src/main/scala/org/apache/spark/mllib/clustering/LDAOptimizer.scala

@@ -142,11 +142,9 @@ final class EMLDAOptimizer extends LDAOptimizer {
     this.k = k
     this.vocabSize = docs.take(1).head._2.size
     this.checkpointInterval = lda.getCheckpointInterval
-    if (this.checkpointInterval != -1) {
-      this.graphCheckpointer = new PeriodicGraphCheckpointer[TopicCounts, TokenCount](
-        checkpointInterval, graph.vertices.sparkContext)
-      this.graphCheckpointer.update(this.graph)
-    }
+    this.graphCheckpointer = new PeriodicGraphCheckpointer[TopicCounts, TokenCount](
+      checkpointInterval, graph.vertices.sparkContext)
+    this.graphCheckpointer.update(this.graph)
     this.globalTopicTotals = computeGlobalTopicTotals()
     this
   }
@@ -191,9 +189,7 @@ final class EMLDAOptimizer extends LDAOptimizer {
     // Update the vertex descriptors with the new counts.
     val newGraph = Graph(docTopicDistributions, graph.edges)
     graph = newGraph
-    if (this.checkpointInterval != -1) {
-      graphCheckpointer.update(newGraph)
-    }
+    graphCheckpointer.update(newGraph)
     globalTopicTotals = computeGlobalTopicTotals()
     this
   }
@@ -212,9 +208,7 @@ final class EMLDAOptimizer extends LDAOptimizer {
 
   override private[clustering] def getLDAModel(iterationTimes: Array[Double]): LDAModel = {
     require(graph != null, "graph is null, EMLDAOptimizer not initialized.")
-    if (this.checkpointInterval != -1) {
-      this.graphCheckpointer.deleteAllCheckpoints()
-    }
+    this.graphCheckpointer.deleteAllCheckpoints()
     // The constructor's default arguments assume gammaShape = 100 to ensure equivalence in
     // LDAModel.toLocal conversion
     new DistributedLDAModel(this.graph, this.globalTopicTotals, this.k, this.vocabSize,

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

@@ -52,7 +52,8 @@ 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.
  *
- * @param checkpointInterval  Datasets will be checkpointed at this interval
+ * @param checkpointInterval  Datasets will be checkpointed at this interval.
+ *                            If this interval was set as -1, then checkpointing will be disabled.
  * @param sc  SparkContext for the Datasets given to this checkpointer
  * @tparam T  Dataset type, such as RDD[Double]
  */
@@ -89,7 +90,8 @@ private[mllib] abstract class PeriodicCheckpointer[T](
     updateCount += 1
 
     // Handle checkpointing (after persisting)
-    if ((updateCount % checkpointInterval) == 0 && sc.getCheckpointDir.nonEmpty) {
+    if (checkpointInterval != -1 && (updateCount % checkpointInterval) == 0
+      && sc.getCheckpointDir.nonEmpty) {
       // Add new checkpoint before removing old checkpoints.
       checkpoint(newData)
       checkpointQueue.enqueue(newData)

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

@@ -69,7 +69,8 @@ import org.apache.spark.storage.StorageLevel
  *  // checkpointed: graph4
  * }}}
  *
- * @param checkpointInterval Graphs will be checkpointed at this interval
+ * @param checkpointInterval Graphs will be checkpointed at this interval.
+ *                           If this interval was set as -1, then checkpointing will be disabled.
  * @tparam VD  Vertex descriptor type
  * @tparam ED  Edge descriptor type
  *