issue created: https://issues.apache.org/jira/browse/FLINK-3207
If anyone has any other suggestion about the renaming, let me know :) -V. On 5 January 2016 at 11:52, Aljoscha Krettek <aljos...@apache.org> wrote: > 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 > >>>>>>>>>>>>>>> > >>>>>>>>>>>>>>> > >>>>>>>>>>>>> > >>>>>>>>>>>> > >>>>>>>>>>> > >>>>>>>>>> > >>>>>>>>> > >>>>>>>> > >>>>>>> > >>>>>> > >>>>> > >>>> > >>> > >>> > >> > >
