I also think a Giraph-like model could be added, but we shouldn't remove
Spargel in favour of it!
On Tue, Nov 3, 2015 at 2:35 AM, Stephan Ewen <se...@apache.org> wrote:
> When creating the original version of Spargel I was pretty much thinking in
> GSA terms, more than in Pregel terms. There are some fundamental
> differences between Spargel and Pregel. Spargel is in between GAS and
> Pregel in some way, that is how I have always thought about it.
>
> The main reason for the form is that it fits the dataflow paradigm easier:
>
> - If one function emits the new state of the vertex and the messages, it
> has two different return types, which means you need a union type and
> filer/split type of operation on the result, which also adds overhead. In
> the current model, each function has one return type, which makes it easy.
>
> - The workset is also the feedback channel, which is materialized at the
> superstep boundaries, so keeping it small at O(vertices), rather than
> O(edges) is a win for performance.
>
> There is no reason to not add a Pregel model, but I would not kill Spargel
> for it. It will be tough to get the Pregel variant to the same efficiency.
> Unless you want to say, for efficiency, go with GSA, for convenience with
> Pregel.
>
> There are some nice things about the Spargel model. The fact that messages
> are first generated then consumes makes the generation of initial messages
> simpler in many cases, I think. It was always a bit weird to me in Pregel
> that you had to check whether you are in superstep one, in which case you
> would expect no message, and generate initial value messages.
>
>
>
> On Fri, Oct 30, 2015 at 1:28 PM, Fabian Hueske <fhue...@gmail.com> wrote:
>
> > We can of course inject an optional ReduceFunction (or GroupReduce, or
> > combinable GroupReduce) to reduce the size of the work set.
> > I suggested to remove the GroupReduce function, because it did only
> collect
> > all messages into a single record by emitting the input iterator which is
> > quite dangerous. Applying a combinable reduce function is could improve
> the
> > performance considerably.
> >
> > The good news is that it would come "for free" because the necessary
> > partitioning and sorting can be reused (given the forwardField
> annotations
> > are correctly set):
> > - The partitioning of the reduce can be reused for the join with the
> > solution set
> > - The sort of the reduce is preserved by the join with the in-memory
> > hash-table of the solution set and can be reused for the coGroup.
> >
> > Best,
> > Fabian
> >
> > 2015-10-30 18:38 GMT+01:00 Vasiliki Kalavri <vasilikikala...@gmail.com>:
> >
> > > Hi Fabian,
> > >
> > > thanks so much for looking into this so quickly :-)
> > >
> > > One update I have to make is that I tried running a few experiments
> with
> > > this on a 6-node cluster. The current implementation gets stuck at
> > > "Rebuilding Workset Properties" and never finishes a single iteration.
> > > Running the plan of one superstep without a delta iteration terminates
> > > fine. I didn't have access to the cluster today, so I couldn't debug
> this
> > > further, but I will do as soon as I have access again.
> > >
> > > The rest of my comments are inline:
> > >
> > > On 30 October 2015 at 17:53, Fabian Hueske <fhue...@gmail.com> wrote:
> > >
> > > > Hi Vasia,
> > > >
> > > > I had a look at your new implementation and have a few ideas for
> > > > improvements.
> > > > 1) Sending out the input iterator as you do in the last GroupReduce
> is
> > > > quite dangerous and does not give a benefit compared to collecting
> all
> > > > elements. Even though it is an iterator, it needs to be completely
> > > > materialized in-memory whenever the record is touched by Flink or
> user
> > > > code.
> > > > I would propose to skip the reduce step completely and handle all
> > > messages
> > > > separates and only collect them in the CoGroup function before giving
> > > them
> > > > into the VertexComputeFunction. Be careful, to only do that with
> > > > objectReuse disabled or take care to properly copy the messages. If
> you
> > > > collect the messages in the CoGroup, you don't need the GroupReduce,
> > have
> > > > smaller records and you can remove the MessageIterator class
> > completely.
> > > >
> > >
> > > I see. The idea was to expose to message combiner that user could
> > > implement if the messages are combinable, e.g. min, sum. This is a
> > common
> > > case and reduces the message load significantly. Is there a way I could
> > do
> > > something similar before the coGroup?
> > >
> > >
> > >
> > > > 2) Add this annotation to the AppendVertexState function:
> > > > @ForwardedFieldsFirst("*->f0"). This indicates that the complete
> > element
> > > of
> > > > the first input becomes the first field of the output. Since the
> input
> > is
> > > > partitioned on "f0" (it comes out of the partitioned solution set)
> the
> > > > result of ApplyVertexState will be partitioned on "f0.f0" which is
> > > > (accidentially :-D) the join key of the following coGroup function ->
> > no
> > > > partitioning :-)
> > > >
> > >
> > > Great! I totally missed that ;)
> > >
> > >
> > >
> > > > 3) Adding the two flatMap functions behind the CoGroup prevents
> > chaining
> > > > and causes therefore some serialization overhead but shouldn't be too
> > > bad.
> > > >
> > > > So in total I would make this program as follows:
> > > >
> > > > iVertices<K,VV>
> > > > iMessage<K, Message> = iVertices.map(new InitWorkSet());
> > > >
> > > > iteration = iVertices.iterateDelta(iMessages, maxIt, 0)
> > > > verticesWithMessage<Vertex, Message> = iteration.getSolutionSet()
> > > > .join(iteration.workSet())
> > > > .where(0) // solution set is local and build side
> > > > .equalTo(0) // workset is shuffled and probe side of hashjoin
> > > > superstepComp<Vertex,Tuple2<K, Message>,Bool> =
> > > > verticesWithMessage.coGroup(edgessWithValue)
> > > > .where("f0.f0") // vwm is locally forward and sorted
> > > > .equalTo(0) // edges are already partitioned and sorted (if cached
> > > > correctly)
> > > > .with(...) // The coGroup collects all messages in a collection and
> > > gives
> > > > it to the ComputeFunction
> > > > delta<Vertex> = superStepComp.flatMap(...) // partitioned when merged
> > > into
> > > > solution set
> > > > workSet<K, Message> = superStepComp.flatMap(...) // partitioned for
> > join
> > > > iteration.closeWith(delta, workSet)
> > > >
> > > > So, if I am correct, the program will
> > > > - partition the workset
> > > > - sort the vertices with messages
> > > > - partition the delta
> > > >
> > > > One observation I have is that this program requires that all
> messages
> > > fit
> > > > into memory. Was that also the case before?
> > > >
> > >
> > > I believe not. The plan has one coGroup that produces the messages
> and a
> > > following coGroup that groups by the messages "target ID" and consumes
> > > them in an iterator. That doesn't require them to fit in memory,
> right?
> > >
> > >
> > > I'm also working on a version where the graph is represented as an
> > > adjacency list, instead of two separate datasets of vertices and edges.
> > The
> > > disadvantage is that the graph has to fit in memory, but I think the
> > > advantages are many. We'll be able to support edge value updates, edge
> > > mutations and different edge access order guarantees. I'll get back to
> > this
> > > thread when I have a working prototype.
> > >
> > >
> > > >
> > > > Cheers,
> > > > Fabian
> > > >
> > >
> > > Thanks again!
> > >
> > > Cheers,
> > > -Vasia.
> > >
> > >
> > >
> > > >
> > > >
> > > > 2015-10-27 19:10 GMT+01:00 Vasiliki Kalavri <
> vasilikikala...@gmail.com
> > >:
> > > >
> > > > > @Martin: thanks for your input! If you ran into any other issues
> > that I
> > > > > didn't mention, please let us know. Obviously, even with my
> proposal,
> > > > there
> > > > > are still features we cannot support, e.g. updating edge values and
> > > graph
> > > > > mutations. We'll need to re-think the underlying iteration and/or
> > graph
> > > > > representation for those.
> > > > >
> > > > > @Fabian: thanks a lot, no rush :)
> > > > > Let me give you some more information that might make it easier to
> > > reason
> > > > > about performance:
> > > > >
> > > > > Currently, in Spargel the SolutionSet (SS) keeps the vertex state
> and
> > > the
> > > > > workset (WS) keeps the active vertices. The iteration is composed
> of
> > 2
> > > > > coGroups. The first one takes the WS and the edges and produces
> > > messages.
> > > > > The second one takes the messages and the SS and produced the new
> WS
> > > and
> > > > > the SS-delta.
> > > > >
> > > > > In my proposal, the SS has the vertex state and the WS has
> <vertexId,
> > > > > MessageIterator> pairs, i.e. the inbox of each vertex. The plan is
> > more
> > > > > complicated because compute() needs to have two iterators: over the
> > > edges
> > > > > and over the messages.
> > > > > First, I join SS and WS to get the active vertices (have received a
> > > msg)
> > > > > and their current state. Then I coGroup the result with the edges
> to
> > > > access
> > > > > the neighbors. Now the main problem is that this coGroup needs to
> > have
> > > 2
> > > > > outputs: the new messages and the new vertex value. I couldn't
> really
> > > > find
> > > > > a nice way to do this, so I'm emitting a Tuple that contains both
> > types
> > > > and
> > > > > I have a flag to separate them later with 2 flatMaps. From the
> vertex
> > > > > flatMap, I crete the SS-delta and from the messaged flatMap I
> apply a
> > > > > reduce to group the messages by vertex and send them to the new WS.
> > One
> > > > > optimization would be to expose a combiner here to reduce message
> > size.
> > > > >
> > > > > tl;dr:
> > > > > 1. 2 coGroups vs. Join + coGroup + flatMap + reduce
> > > > > 2. how can we efficiently emit 2 different types of records from a
> > > > coGroup?
> > > > > 3. does it make any difference if we group/combine the messages
> > before
> > > > > updating the workset or after?
> > > > >
> > > > > Cheers,
> > > > > -Vasia.
> > > > >
> > > > >
> > > > > On 27 October 2015 at 18:39, Fabian Hueske <fhue...@gmail.com>
> > wrote:
> > > > >
> > > > > > I'll try to have a look at the proposal from a performance point
> of
> > > > view
> > > > > in
> > > > > > the next days.
> > > > > > Please ping me, if I don't follow up this thread.
> > > > > >
> > > > > > Cheers, Fabian
> > > > > >
> > > > > > 2015-10-27 18:28 GMT+01:00 Martin Junghanns <
> > m.jungha...@mailbox.org
> > > >:
> > > > > >
> > > > > > > Hi,
> > > > > > >
> > > > > > > At our group, we also moved several algorithms from Giraph to
> > Gelly
> > > > and
> > > > > > > ran into some confusing issues (first in understanding, second
> > > during
> > > > > > > implementation) caused by the conceptional differences you
> > > described.
> > > > > > >
> > > > > > > If there are no concrete advantages (performance mainly) in the
> > > > Spargel
> > > > > > > implementation, we would be very happy to see the Gelly API be
> > > > aligned
> > > > > to
> > > > > > > Pregel-like systems.
> > > > > > >
> > > > > > > Your SSSP example speaks for itself. Straightforward, if the
> > reader
> > > > is
> > > > > > > familiar with Pregel/Giraph/...
> > > > > > >
> > > > > > > Best,
> > > > > > > Martin
> > > > > > >
> > > > > > >
> > > > > > > On 27.10.2015 17:40, Vasiliki Kalavri wrote:
> > > > > > >
> > > > > > >> Hello squirrels,
> > > > > > >>
> > > > > > >> I want to discuss with you a few concerns I have about our
> > current
> > > > > > >> vertex-centric model implementation, Spargel, now fully
> subsumed
> > > by
> > > > > > Gelly.
> > > > > > >>
> > > > > > >> Spargel is our implementation of Pregel [1], but it violates
> > some
> > > > > > >> fundamental properties of the model, as described in the paper
> > and
> > > > as
> > > > > > >> implemented in e.g. Giraph, GPS, Hama. I often find myself
> > > confused
> > > > > both
> > > > > > >> when trying to explain it to current Giraph users and when
> > porting
> > > > my
> > > > > > >> Giraph algorithms to it.
> > > > > > >>
> > > > > > >> More specifically:
> > > > > > >> - in the Pregel model, messages produced in superstep n, are
> > > > received
> > > > > in
> > > > > > >> superstep n+1. In Spargel, they are produced and consumed in
> the
> > > > same
> > > > > > >> iteration.
> > > > > > >> - in Pregel, vertices are active during a superstep, if they
> > have
> > > > > > received
> > > > > > >> a message in the previous superstep. In Spargel, a vertex is
> > > active
> > > > > > during
> > > > > > >> a superstep if it has changed its value.
> > > > > > >>
> > > > > > >> These two differences require a lot of rethinking when porting
> > > > > > >> applications
> > > > > > >> and can easily cause bugs.
> > > > > > >>
> > > > > > >> The most important problem however is that we require the user
> > to
> > > > > split
> > > > > > >> the
> > > > > > >> computation in 2 phases (2 UDFs):
> > > > > > >> - messaging: has access to the vertex state and can produce
> > > messages
> > > > > > >> - update: has access to incoming messages and can update the
> > > vertex
> > > > > > value
> > > > > > >>
> > > > > > >> Pregel/Giraph only expose one UDF to the user:
> > > > > > >> - compute: has access to both the vertex state and the
> incoming
> > > > > > messages,
> > > > > > >> can produce messages and update the vertex value.
> > > > > > >>
> > > > > > >> This might not seem like a big deal, but except from forcing
> the
> > > > user
> > > > > to
> > > > > > >> split their program logic into 2 phases, Spargel also makes
> some
> > > > > common
> > > > > > >> computation patterns non-intuitive or impossible to write. A
> > very
> > > > > simple
> > > > > > >> example is propagating a message based on its value or sender
> > ID.
> > > To
> > > > > do
> > > > > > >> this with Spargel, one has to store all the incoming messages
> in
> > > the
> > > > > > >> vertex
> > > > > > >> value (might be of different type btw) during the messaging
> > phase,
> > > > so
> > > > > > that
> > > > > > >> they can be accessed during the update phase.
> > > > > > >>
> > > > > > >> So, my first question is, when implementing Spargel, were
> other
> > > > > > >> alternatives considered and maybe rejected in favor of
> > performance
> > > > or
> > > > > > >> because of some other reason? If someone knows, I would love
> to
> > > hear
> > > > > > about
> > > > > > >> them!
> > > > > > >>
> > > > > > >> Second, I wrote a prototype implementation [2] that only
> exposes
> > > one
> > > > > > UDF,
> > > > > > >> compute(), by keeping the vertex state in the solution set and
> > the
> > > > > > >> messages
> > > > > > >> in the workset. This way all previously mentioned limitations
> go
> > > > away
> > > > > > and
> > > > > > >> the API (see "SSSPComputeFunction" in the example [3]) looks a
> > lot
> > > > > more
> > > > > > >> like Giraph (see [4]).
> > > > > > >>
> > > > > > >> I have not run any experiments yet and the prototype has some
> > ugly
> > > > > > hacks,
> > > > > > >> but if you think any of this makes sense, then I'd be willing
> to
> > > > > follow
> > > > > > up
> > > > > > >> and try to optimize it. If we see that it performs well, we
> can
> > > > > consider
> > > > > > >> either replacing Spargel or adding it as an alternative.
> > > > > > >>
> > > > > > >> Thanks for reading this long e-mail and looking forward to
> your
> > > > input!
> > > > > > >>
> > > > > > >> Cheers,
> > > > > > >> -Vasia.
> > > > > > >>
> > > > > > >> [1]: https://kowshik.github.io/JPregel/pregel_paper.pdf
> > > > > > >> [2]:
> > > > > > >>
> > > > > > >>
> > > > > >
> > > > >
> > > >
> > >
> >
> https://github.com/vasia/flink/tree/spargel-2/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/spargelnew
> > > > > > >> [3]:
> > > > > > >>
> > > > > > >>
> > > > > >
> > > > >
> > > >
> > >
> >
> https://github.com/vasia/flink/blob/spargel-2/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/spargelnew/example/SSSPCompute.java
> > > > > > >> [4]:
> > > > > > >>
> > > > > > >>
> > > > > >
> > > > >
> > > >
> > >
> >
> https://github.com/grafos-ml/okapi/blob/master/src/main/java/ml/grafos/okapi/graphs/SingleSourceShortestPaths.java
> > > > > > >>
> > > > > > >>
> > > > > >
> > > > >
> > > >
> > >
> >
>
