From f2ced743b2aa147ed2b43221b4f52d502b85c8c6 Mon Sep 17 00:00:00 2001 From: hushan Date: Thu, 21 Apr 2016 21:43:30 +0800 Subject: [PATCH 1/2] fix --- .../main/scala/org/apache/spark/graphx/impl/EdgeRDDImpl.scala | 4 ---- .../main/scala/org/apache/spark/graphx/impl/GraphImpl.scala | 3 ++- .../scala/org/apache/spark/graphx/impl/VertexRDDImpl.scala | 4 ---- 3 files changed, 2 insertions(+), 9 deletions(-) diff --git a/graphx/src/main/scala/org/apache/spark/graphx/impl/EdgeRDDImpl.scala b/graphx/src/main/scala/org/apache/spark/graphx/impl/EdgeRDDImpl.scala index 98e082cc44e1a..b22a6f792d1b5 100644 --- a/graphx/src/main/scala/org/apache/spark/graphx/impl/EdgeRDDImpl.scala +++ b/graphx/src/main/scala/org/apache/spark/graphx/impl/EdgeRDDImpl.scala @@ -75,10 +75,6 @@ class EdgeRDDImpl[ED: ClassTag, VD: ClassTag] private[graphx] ( partitionsRDD.checkpoint() } - override def isCheckpointed: Boolean = { - firstParent[(PartitionID, EdgePartition[ED, VD])].isCheckpointed - } - override def getCheckpointFile: Option[String] = { partitionsRDD.getCheckpointFile } diff --git a/graphx/src/main/scala/org/apache/spark/graphx/impl/GraphImpl.scala b/graphx/src/main/scala/org/apache/spark/graphx/impl/GraphImpl.scala index e18831382d4d5..5a535c1cbaccc 100644 --- a/graphx/src/main/scala/org/apache/spark/graphx/impl/GraphImpl.scala +++ b/graphx/src/main/scala/org/apache/spark/graphx/impl/GraphImpl.scala @@ -68,7 +68,8 @@ class GraphImpl[VD: ClassTag, ED: ClassTag] protected ( } override def isCheckpointed: Boolean = { - vertices.isCheckpointed && replicatedVertexView.edges.isCheckpointed + vertices.partitionsRDD.isCheckpointed && + replicatedVertexView.edges.partitionsRDD.isCheckpointed } override def getCheckpointFiles: Seq[String] = { diff --git a/graphx/src/main/scala/org/apache/spark/graphx/impl/VertexRDDImpl.scala b/graphx/src/main/scala/org/apache/spark/graphx/impl/VertexRDDImpl.scala index d314522de9916..2c82b6800b2fb 100644 --- a/graphx/src/main/scala/org/apache/spark/graphx/impl/VertexRDDImpl.scala +++ b/graphx/src/main/scala/org/apache/spark/graphx/impl/VertexRDDImpl.scala @@ -75,10 +75,6 @@ class VertexRDDImpl[VD] private[graphx] ( partitionsRDD.checkpoint() } - override def isCheckpointed: Boolean = { - firstParent[ShippableVertexPartition[VD]].isCheckpointed - } - override def getCheckpointFile: Option[String] = { partitionsRDD.getCheckpointFile } From a106758dd4492ccae851c48d010e6eb9fb26b849 Mon Sep 17 00:00:00 2001 From: hushan Date: Fri, 22 Apr 2016 10:37:33 +0800 Subject: [PATCH 2/2] Add testcase --- graphx/src/test/scala/org/apache/spark/graphx/GraphSuite.scala | 2 ++ 1 file changed, 2 insertions(+) diff --git a/graphx/src/test/scala/org/apache/spark/graphx/GraphSuite.scala b/graphx/src/test/scala/org/apache/spark/graphx/GraphSuite.scala index 96aa262a395c8..35687ddc92faa 100644 --- a/graphx/src/test/scala/org/apache/spark/graphx/GraphSuite.scala +++ b/graphx/src/test/scala/org/apache/spark/graphx/GraphSuite.scala @@ -350,6 +350,8 @@ class GraphSuite extends SparkFunSuite with LocalSparkContext { assert(verticesDependencies.forall(_.rdd.isInstanceOf[CheckpointRDD[_]])) assert(graph.isCheckpointed) assert(graph.getCheckpointFiles.size === 2) + assert(!graph.edges.isCheckpointed) + assert(!graph.vertices.isCheckpointed) } }