Nice to hear. :D
I think you can go ahead and add the Jira. About the renaming: I also think
that it would make sense to do it.
> On 04 Jan 2016, at 19:48, Vasiliki Kalavri <vasilikikala...@gmail.com> wrote:
>
> Hello squirrels and happy new year!
>
> I'm reviving this thread to share some results and discuss next steps.
>
> Using the Either type I was able to get rid of redundant messages and
> vertex state. During the past few weeks, I have been running experiments,
> which show that the performance of this "Pregel" model has improved a lot :)
> In [1], you can see the speedup of GSA and Pregel over Spargel, for SSSP
> and Connected Components (CC), for the Livejournal (68m edges), Orkut (117m
> edges) and Wikipedia (340m edges) datasets.
>
> Regarding next steps, if no objections, I will open a Jira for adding a
> Pregel iteration abstraction to Gelly. The Gelly guide has to be updated to
> reflect the spectrum of iteration abstractions that we have discussed in
> this thread, i.e. Pregel -> Spargel (Scatter-Gather) -> GSA.
>
> I think it might also be a good idea to do some renaming. Currently, we
> call the Spargel iteration "vertex-centric", which fits better to the
> Pregel abstraction. I propose we rename the spargel iteration into
> "scatter-gather" or "signal-collect" (where it was first introduced [2]).
> Any other ideas?
>
> Thanks,
> -Vasia.
>
> [1]:
> https://drive.google.com/file/d/0BzQJrI2eGlyYRTRjMkp1d3R6eVE/view?usp=sharing
> [2]: http://link.springer.com/chapter/10.1007/978-3-642-17746-0_48
>
> On 11 November 2015 at 11:05, Stephan Ewen <se...@apache.org> wrote:
>
>> See: https://issues.apache.org/jira/browse/FLINK-3002
>>
>> On Wed, Nov 11, 2015 at 10:54 AM, Stephan Ewen <se...@apache.org> wrote:
>>
>>> "Either" an "Optional" types are quite useful.
>>>
>>> Let's add them to the core Java API.
>>>
>>> On Wed, Nov 11, 2015 at 10:00 AM, Vasiliki Kalavri <
>>> vasilikikala...@gmail.com> wrote:
>>>
>>>> Thanks Fabian! I'll try that :)
>>>>
>>>> On 10 November 2015 at 22:31, Fabian Hueske <fhue...@gmail.com> wrote:
>>>>
>>>>> You could implement a Java Either type (similar to Scala's Either)
>> that
>>>>> either has a Message or the VertexState and a corresponding
>>>> TypeInformation
>>>>> and TypeSerializer that serializes a byte flag to indicate which both
>>>> types
>>>>> is used.
>>>>> It might actually make sense, to add a generic Either type to the Java
>>>> API
>>>>> in general (similar to the Java Tuples with resemble the Scala
>> Tuples).
>>>>>
>>>>> Cheers, Fabian
>>>>>
>>>>> 2015-11-10 22:16 GMT+01:00 Vasiliki Kalavri <
>> vasilikikala...@gmail.com
>>>>> :
>>>>>
>>>>>> Hi,
>>>>>>
>>>>>> after running a few experiments, I can confirm that putting the
>>>> combiner
>>>>>> after the flatMap is indeed more efficient.
>>>>>>
>>>>>> I ran SSSP and Connected Components with Spargel, GSA, and the
>> Pregel
>>>>> model
>>>>>> and the results are the following:
>>>>>>
>>>>>> - for SSSP, Spargel is always the slowest, GSA is a ~1.2x faster and
>>>>> Pregel
>>>>>> is ~1.1x faster without combiner, ~1.3x faster with combiner.
>>>>>> - for Connected Components, Spargel and GSA perform similarly, while
>>>>> Pregel
>>>>>> is 1.4-1.6x slower.
>>>>>>
>>>>>> To start with, this is much better than I expected :)
>>>>>> However, there is a main shortcoming in my current implementation
>> that
>>>>>> negatively impacts performance:
>>>>>> Since the compute function coGroup needs to output both new vertex
>>>> values
>>>>>> and new messages, I emit a wrapping tuple that contains both vertex
>>>> state
>>>>>> and messages and then filter them out based on a boolean field. The
>>>>> problem
>>>>>> is that since I cannot emit null fields, I emit a dummy message for
>>>> each
>>>>>> new vertex state and a dummy vertex state for each new message. That
>>>>>> essentially means that the intermediate messages result is double in
>>>>> size,
>>>>>> if say the vertex values are of the same type as the messages (can
>> be
>>>>> worse
>>>>>> if the vertex values are more complex).
>>>>>> So my question is, is there a way to avoid this redundancy, by
>> either
>>>>>> emitting null fields or by creating an operator that could emit 2
>>>>> different
>>>>>> types of tuples?
>>>>>>
>>>>>> Thanks!
>>>>>> -Vasia.
>>>>>>
>>>>>> On 9 November 2015 at 15:20, Fabian Hueske <fhue...@gmail.com>
>> wrote:
>>>>>>
>>>>>>> Hi Vasia,
>>>>>>>
>>>>>>> sorry for the late reply.
>>>>>>> I don't think there is a big difference. In both cases, the
>>>>> partitioning
>>>>>>> and sorting happens at the end of the iteration.
>>>>>>> If the groupReduce is applied before the workset is returned, the
>>>>> sorting
>>>>>>> happens on the filtered result (after the flatMap) which might be
>> a
>>>>>> little
>>>>>>> bit more efficient (depending on the ratio of messages and
>> solution
>>>> set
>>>>>>> updates). Also it does not require that the initial workset is
>>>> sorted
>>>>> for
>>>>>>> the first groupReduce.
>>>>>>>
>>>>>>> I would put it at the end.
>>>>>>>
>>>>>>> Cheers, Fabian
>>>>>>>
>>>>>>> 2015-11-05 17:19 GMT+01:00 Vasiliki Kalavri <
>>>> vasilikikala...@gmail.com
>>>>>> :
>>>>>>>
>>>>>>>> @Fabian
>>>>>>>>
>>>>>>>> Is there any advantage in putting the reducer-combiner before
>>>>> updating
>>>>>>> the
>>>>>>>> workset vs. after (i.e. right before the join with the solution
>>>> set)?
>>>>>>>>
>>>>>>>> If it helps, here are the plans of these 2 alternatives:
>>>>>>>>
>>>>>>>>
>>>>>>>>
>>>>>>>
>>>>>>
>>>>>
>>>>
>> https://drive.google.com/file/d/0BzQJrI2eGlyYcFV2RFo5dUFNXzg/view?usp=sharing
>>>>>>>>
>>>>>>>>
>>>>>>>
>>>>>>
>>>>>
>>>>
>> https://drive.google.com/file/d/0BzQJrI2eGlyYN014NXp6OEZUdGs/view?usp=sharing
>>>>>>>>
>>>>>>>> Thanks a lot for the help!
>>>>>>>>
>>>>>>>> -Vasia.
>>>>>>>>
>>>>>>>> On 30 October 2015 at 21:28, 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
>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>
>>>>>>>>>>>>>
>>>>>>>>>>>>
>>>>>>>>>>>
>>>>>>>>>>
>>>>>>>>>
>>>>>>>>
>>>>>>>
>>>>>>
>>>>>
>>>>
>>>
>>>
>>
