spark-user mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From Jay Hutfles <jayhutf...@gmail.com>
Subject Re: GraphX: ShortestPaths does not terminate on a grid graph
Date Thu, 29 Jan 2015 15:25:01 GMT
Just curious, is this set to be merged at some point?

On Thu Jan 22 2015 at 4:34:46 PM Ankur Dave <ankurdave@gmail.com> wrote:

> At 2015-01-22 02:06:37 -0800, NicolasC <Nicolas.Chleq@inria.fr> wrote:
> > I try to execute a simple program that runs the ShortestPaths algorithm
> > (org.apache.spark.graphx.lib.ShortestPaths) on a small grid graph.
> > I use Spark 1.2.0 downloaded from spark.apache.org.
> >
> > This program runs more than 2 hours when the grid size is 70x70 as
> above, and is then killed
> > by the resource manager of the cluster (Torque). After a 5-6 minutes of
> execution, the
> > Spark master UI does not even respond.
> >
> > For a grid size of 30x30, the program terminates in about 20 seconds,
> and for a grid size
> > of 50x50 it finishes in about 80 seconds. The problem appears for a grid
> size of 70x70 and
> > above.
>
> Unfortunately this problem is due to a Spark bug. In later iterations of
> iterative algorithms, the lineage maintained for fault tolerance grows long
> and causes Spark to consume increasing amounts of resources for scheduling
> and task serialization.
>
> The workaround is to checkpoint the graph periodically, which writes it to
> stable storage and interrupts the lineage chain before it grows too long.
>
> If you're able to recompile Spark, you can do this by applying the patch
> to GraphX at the end of this mail, and before running graph algorithms,
> calling
>
>     sc.setCheckpointDir("/tmp")
>
> to set the checkpoint directory as desired.
>
> Ankur
>
> === patch begins here ===
>
> 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..1fbbb87 100644
> --- a/graphx/src/main/scala/org/apache/spark/graphx/Pregel.scala
> +++ b/graphx/src/main/scala/org/apache/spark/graphx/Pregel.scala
> @@ -126,6 +126,8 @@ object Pregel extends Logging {
>      // Loop
>      var prevG: Graph[VD, ED] = null
>      var i = 0
> +    val checkpoint = g.vertices.context.getCheckpointDir.nonEmpty
> +    val checkpointFrequency = 25
>      while (activeMessages > 0 && i < maxIterations) {
>        // Receive the messages. Vertices that didn't get any messages do
> not appear in newVerts.
>        val newVerts = g.vertices.innerJoin(messages)(vprog).cache()
> @@ -139,6 +141,14 @@ object Pregel extends Logging {
>        // get to send messages. We must cache messages so it can be
> materialized on the next line,
>        // allowing us to uncache the previous iteration.
>        messages = g.mapReduceTriplets(sendMsg, mergeMsg, Some((newVerts,
> activeDirection))).cache()
> +
> +      if (checkpoint && i % checkpointFrequency == checkpointFrequency -
> 1) {
> +        logInfo("Checkpointing in iteration " + i)
> +        g.vertices.checkpoint()
> +        g.edges.checkpoint()
> +        messages.checkpoint()
> +      }
> +
>        // 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).
>
> ---------------------------------------------------------------------
> To unsubscribe, e-mail: user-unsubscribe@spark.apache.org
> For additional commands, e-mail: user-help@spark.apache.org
>
>

Mime
View raw message