Checkpointing Pregel

[Jeffrey Picard]

Hey guys,

I’m trying to run connected components on graphs that end up running for a 
fairly large number of iterations (25-30) and take 5-6 hours. I find more than 
half the time I end up getting fetch failures and losing an executor after a 
number of iterations. Then it has to go back and recompute pieces that it lost, 
which don’t seem to be getting persisted at the same level as the graph so 
those iterations take exponentially longer and I have to kill the job because 
it’s not worth waiting for it to finish.

The approach I’m currently trying is checkpointing the vertices and edges (and 
maybe the messages?) in Pregel. What I’ve been testing with so far is the below 
patch, which seems to be working (actually I haven’t had any failures since I 
added this change, so I don’t know if I did get one if it would recompute from 
the start or not) but I’m also seeing things like 5 instances of VertexRDDs 
being persisted all at the same time and “reduce at VertexRDD.scala:111” runs 
twice each time. I was wondering if this is the proper / most efficient way of 
doing this checkpointing, and if not what would work better?

diff --git a/graphx/src/main/scala/org/apache/spark/graphx/Pregel.scala 
b/graphx/src/main/scala/org/apache/spark/graphx/Pregel.scala
index 5e55620..5be40c3 100644
--- a/graphx/src/main/scala/org/apache/spark/graphx/Pregel.scala
+++ b/graphx/src/main/scala/org/apache/spark/graphx/Pregel.scala
@@ -134,6 +134,11 @@ object Pregel extends Logging {
       g = g.outerJoinVertices(newVerts) { (vid, old, newOpt) => 
newOpt.getOrElse(old) }
       g.cache()

+      g.vertices.checkpoint()
+      g.vertices.count()
+      g.edges.checkpoint()
+      g.edges.count()
+
       val oldMessages = messages
       // Send new messages. Vertices that didn't get any messages don't appear 
in newVerts, so don't
       // get to send messages. We must cache messages so it can be 
materialized on the next line,
@@ -142,6 +147,7 @@ object Pregel extends Logging {
       // The call to count() materializes `messages`, `newVerts`, and the 
vertices of `g`. This
       // hides oldMessages (depended on by newVerts), newVerts (depended on by 
messages), and the
       // vertices of prevG (depended on by newVerts, oldMessages, and the 
vertices of g).
+         messages.checkpoint()
       activeMessages = messages.count()

       logInfo("Pregel finished iteration " + i)

Best Regards,

Jeffrey Picard

signature.asc
Description: Message signed with OpenPGP using GPGMail