Replace edge checkpointing with spark dataset.checkpoint

amuraru · amuraru · commit e3ee26318efc · 2018-07-18T17:46:03.000+03:00
diff --git a/src/main/scala/org/graphframes/lib/ConnectedComponents.scala b/src/main/scala/org/graphframes/lib/ConnectedComponents.scala
@@ -340,18 +340,7 @@ object ConnectedComponents extends Logging {
 
       // checkpointing
       if (shouldCheckpoint && (iteration % checkpointInterval == 0)) {
-        // TODO: remove this after DataFrame.checkpoint is implemented
-        val out = s"${checkpointDir.get}/$iteration"
-        ee.write.parquet(out)
-        // may hit S3 eventually consistent issue
-        ee = sqlContext.read.parquet(out)
-
-        // remove previous checkpoint
-        if (iteration > checkpointInterval) {
-          val path = new Path(s"${checkpointDir.get}/${iteration - checkpointInterval}")
-          path.getFileSystem(sc.hadoopConfiguration).delete(path, true)
-        }
-
+        ee = ee.checkpoint()
         System.gc() // hint Spark to clean shuffle directories
       }
 
