You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by andralungu <gi...@git.apache.org> on 2015/03/26 09:20:45 UTC
[GitHub] flink pull request: [FLINK-1523][gelly] Vertex centric iteration e...
GitHub user andralungu opened a pull request:
https://github.com/apache/flink/pull/537
[FLINK-1523][gelly] Vertex centric iteration extensions
Thank you @StephanEwen for the hint!(keeping the degrees within the vertex value)
@vasia, I am sure I left some trailing irrelevant comments/unused functions there.
This is the first draft for the Spargel extensions; it changes the usual runVertexCentricIteration(), which means that feedback is welcome.
Thanks!
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/andralungu/flink spargelExt
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/flink/pull/537.patch
To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:
This closes #537
----
commit d6ce302fbbf8994b87f8730d0a33430711c71ab8
Author: andralungu <lu...@gmail.com>
Date: 2015-03-16T13:04:23Z
[FLINK-1523][gelly] Vertex centric iteration extensions
[FLINK-1523][gelly] Removed trailing splits into vertex key and value
[FLINK-1523][gelly] Removed unused import
[FLINK-1523][gelly] Final bugfixes
----
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-1523][gelly] Vertex centric iteration e...
Posted by andralungu <gi...@git.apache.org>.
Github user andralungu commented on the pull request:
https://github.com/apache/flink/pull/537#issuecomment-101978637
Hi @vasia,
I had a look at the new branch. The changes look good, degrees are no longer exposed to the user and the current approach removes the need to subclass Vertex. :+1:
The only small remark/comment I have comes from a user perspective:
- let's say that, by mistake, I forgot to set the degrees option;
- let's also say I was too "busy" to read the manual :)
- result: I will get -1 instead of the expected number of degrees per vertex
I understand why you had to pass -1 there; it should be of the same type as the degrees. However, maybe we can come up with some way to hint users that they should not forget to set the corresponding options. Adding an extra line in the documentation might not suffice.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-1523][gelly] Vertex centric iteration e...
Posted by vasia <gi...@git.apache.org>.
Github user vasia commented on a diff in the pull request:
https://github.com/apache/flink/pull/537#discussion_r28644120
--- Diff: flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/example/IncrementalSSSPITCase.java ---
@@ -0,0 +1,84 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.graph.test.example;
+
+import com.google.common.base.Charsets;
+import com.google.common.io.Files;
+import org.apache.flink.graph.example.IncrementalSSSPExample;
+import org.apache.flink.graph.example.utils.IncrementalSSSPData;
+import org.apache.flink.test.util.MultipleProgramsTestBase;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+
+import java.io.File;
+
+@RunWith(Parameterized.class)
+public class IncrementalSSSPITCase extends MultipleProgramsTestBase {
+
+ private String verticesPath;
+
+ private String edgesPath;
+
+ private String edgesInSSSPPath;
+
+ private String resultPath;
+
+ private String expected;
+
+ @Rule
+ public TemporaryFolder tempFolder = new TemporaryFolder();
+
+ public IncrementalSSSPITCase(TestExecutionMode mode) {
+ super(mode);
+ }
+
+ @Before
+ public void before() throws Exception {
+ resultPath = tempFolder.newFile().toURI().toString();
+ File verticesFile = tempFolder.newFile();
+ Files.write(IncrementalSSSPData.VERTICES, verticesFile, Charsets.UTF_8);
+
+ File edgesFile = tempFolder.newFile();
+ Files.write(IncrementalSSSPData.EDGES, edgesFile, Charsets.UTF_8);
+
+ File edgesInSSSPFile = tempFolder.newFile();
+ Files.write(IncrementalSSSPData.EDGES_IN_SSSP, edgesInSSSPFile, Charsets.UTF_8);
+
+ verticesPath = verticesFile.toURI().toString();
+ edgesPath = edgesFile.toURI().toString();
+ edgesInSSSPPath = edgesInSSSPFile.toURI().toString();
+ }
+
+ @Test
--- End diff --
I would add a few more tests here with a bigger dataset maybe. The example algorithm is quite complex. you only check what happens if you remove an SP-edge is a very small graph. We should at least test what happens when you remove a non-SP-edge and whether the INVALIDATE messages are propagated correctly in the case of a bigger graph.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-1523][gelly] Vertex centric iteration e...
Posted by vasia <gi...@git.apache.org>.
Github user vasia commented on a diff in the pull request:
https://github.com/apache/flink/pull/537#discussion_r27256989
--- Diff: flink-staging/flink-gelly/src/main/java/org/apache/flink/graph/spargel/MessagingFunction.java ---
@@ -38,11 +41,41 @@
* @param <Message> The type of the message sent between vertices along the edges.
* @param <EdgeValue> The type of the values that are associated with the edges.
*/
-public abstract class MessagingFunction<VertexKey extends Comparable<VertexKey> & Serializable,
- VertexValue extends Serializable, Message, EdgeValue extends Serializable> implements Serializable {
+public abstract class MessagingFunction<VertexKey extends Comparable<VertexKey> & Serializable,
+ VertexValue extends Serializable, Message, EdgeValue extends Serializable> implements Serializable {
private static final long serialVersionUID = 1L;
-
+
+ // --------------------------------------------------------------------------------------------
+ // Attributes that allow vertices to access their in/out degrees and the total number of vertices
+ // inside an iteration.
+ // --------------------------------------------------------------------------------------------
+
+ private long numberOfVertices;
+
+ public long getNumberOfVertices() {
+ return numberOfVertices;
+ }
+
+ public void setNumberOfVertices(long numberOfVertices) {
--- End diff --
I guess this one shouldn't be public.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-1523][gelly] Vertex centric iteration e...
Posted by vasia <gi...@git.apache.org>.
Github user vasia commented on a diff in the pull request:
https://github.com/apache/flink/pull/537#discussion_r28643442
--- Diff: flink-staging/flink-gelly/src/main/java/org/apache/flink/graph/example/IncrementalSSSPExample.java ---
@@ -0,0 +1,304 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.graph.example;
+
+import org.apache.flink.api.common.ProgramDescription;
+import org.apache.flink.api.common.functions.FilterFunction;
+import org.apache.flink.api.common.functions.MapFunction;
+import org.apache.flink.api.java.DataSet;
+import org.apache.flink.api.java.ExecutionEnvironment;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.api.java.tuple.Tuple3;
+import org.apache.flink.graph.Edge;
+import org.apache.flink.graph.EdgeDirection;
+import org.apache.flink.graph.Graph;
+import org.apache.flink.graph.Vertex;
+import org.apache.flink.graph.example.utils.IncrementalSSSPData;
+import org.apache.flink.graph.spargel.IterationConfiguration;
+import org.apache.flink.graph.spargel.MessageIterator;
+import org.apache.flink.graph.spargel.MessagingFunction;
+import org.apache.flink.graph.spargel.VertexUpdateFunction;
+
+/**
+ * Incremental Single Sink Shortest Paths Example.
+ *
+ * The program takes as input the resulted graph after a SSSP computation,
+ * an edge to be removed and the initial graph(i.e. before SSSP was computed).
+ *
+ * - If the removed edge does not belong to the SP-graph, no computation is necessary.
+ * The edge is simply removed from the graph.
+ * - If the removed edge is an SP-edge, then all nodes, whose shortest path contains the removed edge,
--- End diff --
and what is an SP-edge
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-1523][gelly] Vertex centric iteration e...
Posted by andralungu <gi...@git.apache.org>.
Github user andralungu commented on the pull request:
https://github.com/apache/flink/pull/537#issuecomment-90728415
Hi @vasia ,
I simply assumed that if the previous examples work, the iteration also works as before. Would you like a specific test suite for that?
For the degree option not being set : the getInDegree(); getOutDegree() methods are located inside the Vertex class. Not sure that their availability can be changed by setting the option on the iteration. They are not computed when the degree option is not set, but you can still see the method. Do you have any idea how to make the degrees inaccessible?
Once this becomes clear, I can add the tests :)
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-1523][gelly] Vertex centric iteration e...
Posted by vasia <gi...@git.apache.org>.
Github user vasia commented on a diff in the pull request:
https://github.com/apache/flink/pull/537#discussion_r28643290
--- Diff: docs/gelly_guide.md ---
@@ -380,6 +380,16 @@ all aggregates globally once per superstep and makes them available in the next
* <strong>Broadcast Variables</strong>: DataSets can be added as [Broadcast Variables](programming_guide.html#broadcast-variables) to the `VertexUpdateFunction` and `MessagingFunction`, using the `addBroadcastSetForUpdateFunction()` and `addBroadcastSetForMessagingFunction()` methods, respectively.
+* <strong>Number of Vertices</strong>: Accessing the total number of vertices within the iteration. This property can be set using the `setOptNumVertices()` method.
+
--- End diff --
delete this newline
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-1523][gelly] Vertex centric iteration e...
Posted by vasia <gi...@git.apache.org>.
Github user vasia commented on a diff in the pull request:
https://github.com/apache/flink/pull/537#discussion_r28643447
--- Diff: flink-staging/flink-gelly/src/main/java/org/apache/flink/graph/example/IncrementalSSSPExample.java ---
@@ -0,0 +1,304 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.graph.example;
+
+import org.apache.flink.api.common.ProgramDescription;
+import org.apache.flink.api.common.functions.FilterFunction;
+import org.apache.flink.api.common.functions.MapFunction;
+import org.apache.flink.api.java.DataSet;
+import org.apache.flink.api.java.ExecutionEnvironment;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.api.java.tuple.Tuple3;
+import org.apache.flink.graph.Edge;
+import org.apache.flink.graph.EdgeDirection;
+import org.apache.flink.graph.Graph;
+import org.apache.flink.graph.Vertex;
+import org.apache.flink.graph.example.utils.IncrementalSSSPData;
+import org.apache.flink.graph.spargel.IterationConfiguration;
+import org.apache.flink.graph.spargel.MessageIterator;
+import org.apache.flink.graph.spargel.MessagingFunction;
+import org.apache.flink.graph.spargel.VertexUpdateFunction;
+
+/**
+ * Incremental Single Sink Shortest Paths Example.
--- End diff --
I would also add a note that this example illustrates how to use the messaging direction configuration option of the vertex-centric iterations.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-1523][gelly] Vertex centric iteration e...
Posted by vasia <gi...@git.apache.org>.
Github user vasia commented on the pull request:
https://github.com/apache/flink/pull/537#issuecomment-94167112
Hi @andralungu! Thanks for the update :-)
I left a few inline comments. Overall it looks good, I just had a little trouble following the logic in `VertexCentricIteration` with your changes. I think it could be simplified a bit, so that all configuration options are set in one place. Other that that, there are some javadocs missing, please make sure to add a description on every public method.
One concern I have is whether the `Vertex` class is the right place for the degrees fields and methods. These should be available only inside the iteration methods. In the current implementation one can use the public `setInDegree` method and then retrieve the value, regardless of whether that happens inside an iteration or whether it was set in the configuration. Could we instead create methods inside the `VertexUpdateFunction` and `MessagingFunction`? Something like `getVertexInDegree()` instead of `vertex.getInDegree()`? Or do you have a better idea?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-1523][gelly] Vertex centric iteration e...
Posted by vasia <gi...@git.apache.org>.
Github user vasia commented on the pull request:
https://github.com/apache/flink/pull/537#issuecomment-87462312
Hi @andralungu!
Regarding (2), I think you could do that inside VertexCentricIteration instead of inside a method of Graph.
Regarding (3), I agree that adding another VertexCentricIteration wouldn't be a nice solution. Why don't you create the Tuple3 (or whatever you need based on the options) vertex dataset inside `createResult()`? I don't see the need in changing VertexCentricIteration's types / output.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-1523][gelly] Vertex centric iteration e...
Posted by vasia <gi...@git.apache.org>.
Github user vasia commented on the pull request:
https://github.com/apache/flink/pull/537#issuecomment-102366422
Hey @andralungu! Thanks a lot for looking into this!
Can you please close this PR and I'll open a new one with the latest changes.
Regarding returning -1 if the option is not set, I did that so that we avoid throwing an exception and also because other options like getIterationAggregator() have a similar behavior. I really don't think it's such a big deal for one to read the javadoc when using a method :P
Of course, if you have any idea on how we could make this better, please let me know! (but let's continue the discussion in the new PR).
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-1523][gelly] Vertex centric iteration e...
Posted by vasia <gi...@git.apache.org>.
Github user vasia commented on the pull request:
https://github.com/apache/flink/pull/537#issuecomment-86727594
Hey @andralungu! Thanks for this PR. I know it took more effort than you might have thought initially :-)
I have a few comments:
- In my opinion, the degrees availability and the total number of vertices should be *optional*. Otherwise, we are introducing unnecessary overhead for the majority of cases, i.e. when these are not actually used by an algorithm.
- The edge direction option and the degrees / numOfVertices should be *separate extensions* and not tight together, i.e. we should allow a user to choose the edge direction option independently of whether they also want to access the degrees. To be consistent with the rest of the VertexCentricIteration options, we could introduce methods `setMessagingDirection()`, `setInDegreesAvailable()` and so on.
- The change should be transparent to user code, i.e. if a user has turned on the option for accessing inDegrees, they should be able to call `vertex.getInDegree()` inside the messaging/update functions, but `vertex.getValue()` should still return the vertex value. In other words, this Tuple3 you're using to store the degrees should be hidden from the public methods.
- Finally, we should add documentation for these options, including description of how they work and some illustrative examples.
Let me know what you think about the above and whether you have any questions!
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-1523][gelly] Vertex centric iteration e...
Posted by andralungu <gi...@git.apache.org>.
Github user andralungu commented on a diff in the pull request:
https://github.com/apache/flink/pull/537#discussion_r28785218
--- Diff: flink-staging/flink-gelly/src/main/java/org/apache/flink/graph/spargel/VertexCentricIteration.java ---
@@ -138,69 +146,46 @@ public void setInput(DataSet<Vertex<VertexKey, VertexValue>> inputData) {
if (this.initialVertices == null) {
--- End diff --
I made that division in order to avoid having duplicate code: the number of vertices and the direction are totally independent of the "degree" option which is why they can be set in the createResult() method. Afterwards, the code does exactly what you described in this comment: it separates the creation of a delta iteration and the creation of the messaging function plus vertex update function according to the vertex type(with degrees or not). It's not just the vertex that changes, but everything that uses its value afterwards changes too. I suggest you look a bit closer at the createResultVerticesWithDegrees and createResultSimpleVertex methods. I don't think their functionality can be simplified by just creating a simple vertex and a vertex with degrees.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-1523][gelly] Vertex centric iteration e...
Posted by vasia <gi...@git.apache.org>.
Github user vasia commented on a diff in the pull request:
https://github.com/apache/flink/pull/537#discussion_r27256701
--- Diff: flink-staging/flink-gelly/src/main/java/org/apache/flink/graph/library/LabelPropagation.java ---
@@ -68,12 +70,12 @@ public LabelPropagation(int maxIterations) {
public static final class UpdateVertexLabel<K extends Comparable<K> & Serializable>
extends VertexUpdateFunction<K, Long, Long> {
- public void updateVertex(K vertexKey, Long vertexValue,
+ public void updateVertex(Vertex<K, Tuple3<Long,Long,Long>> vertex,
MessageIterator<Long> inMessages) {
Map<Long, Long> labelsWithFrequencies = new HashMap<Long, Long>();
long maxFrequency = 1;
- long mostFrequentLabel = vertexValue;
+ long mostFrequentLabel = vertex.getValue().f0;
--- End diff --
`vertex.getValue()` should return the vertex value, not a Tuple3 from the user has to extract the vertex value :S
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-1523][gelly] Vertex centric iteration e...
Posted by vasia <gi...@git.apache.org>.
Github user vasia commented on a diff in the pull request:
https://github.com/apache/flink/pull/537#discussion_r28643926
--- Diff: flink-staging/flink-gelly/src/main/java/org/apache/flink/graph/spargel/VertexCentricIteration.java ---
@@ -138,69 +146,46 @@ public void setInput(DataSet<Vertex<VertexKey, VertexValue>> inputData) {
if (this.initialVertices == null) {
throw new IllegalStateException("The input data set has not been set.");
}
-
+
// prepare some type information
- TypeInformation<Vertex<VertexKey, VertexValue>> vertexTypes = initialVertices.getType();
TypeInformation<VertexKey> keyType = ((TupleTypeInfo<?>) initialVertices.getType()).getTypeAt(0);
TypeInformation<Tuple2<VertexKey, Message>> messageTypeInfo = new TupleTypeInfo<Tuple2<VertexKey,Message>>(keyType, messageType);
- final int[] zeroKeyPos = new int[] {0};
-
- final DeltaIteration<Vertex<VertexKey, VertexValue>, Vertex<VertexKey, VertexValue>> iteration =
- this.initialVertices.iterateDelta(this.initialVertices, this.maximumNumberOfIterations, zeroKeyPos);
+ // create a graph
+ Graph<VertexKey, VertexValue, EdgeValue> graph =
+ Graph.fromDataSet(initialVertices, edgesWithValue, ExecutionEnvironment.getExecutionEnvironment());
- // set up the iteration operator
- if (this.configuration != null) {
+ // check whether the numVertices option is set and, if so, compute the total number of vertices
+ // and set it within the messaging and update functions
- iteration.name(this.configuration.getName(
- "Vertex-centric iteration (" + updateFunction + " | " + messagingFunction + ")"));
- iteration.parallelism(this.configuration.getParallelism());
- iteration.setSolutionSetUnManaged(this.configuration.isSolutionSetUnmanagedMemory());
-
- // register all aggregators
- for (Map.Entry<String, Aggregator<?>> entry : this.configuration.getAggregators().entrySet()) {
- iteration.registerAggregator(entry.getKey(), entry.getValue());
+ if (this.configuration != null && this.configuration.isOptNumVertices()) {
+ try {
+ long numberOfVertices = graph.numberOfVertices();
+ messagingFunction.setNumberOfVertices(numberOfVertices);
+ updateFunction.setNumberOfVertices(numberOfVertices);
+ } catch (Exception e) {
+ e.printStackTrace();
}
}
- else {
- // no configuration provided; set default name
- iteration.name("Vertex-centric iteration (" + updateFunction + " | " + messagingFunction + ")");
- }
-
- // build the messaging function (co group)
- CoGroupOperator<?, ?, Tuple2<VertexKey, Message>> messages;
- MessagingUdfWithEdgeValues<VertexKey, VertexValue, Message, EdgeValue> messenger = new MessagingUdfWithEdgeValues<VertexKey, VertexValue, Message, EdgeValue>(messagingFunction, messageTypeInfo);
- messages = this.edgesWithValue.coGroup(iteration.getWorkset()).where(0).equalTo(0).with(messenger);
-
- // configure coGroup message function with name and broadcast variables
- messages = messages.name("Messaging");
- if (this.configuration != null) {
- for (Tuple2<String, DataSet<?>> e : this.configuration.getMessagingBcastVars()) {
- messages = messages.withBroadcastSet(e.f1, e.f0);
- }
+ if(this.configuration != null) {
+ messagingFunction.setDirection(this.configuration.getDirection());
+ } else {
+ messagingFunction.setDirection(EdgeDirection.OUT);
}
-
- VertexUpdateUdf<VertexKey, VertexValue, Message> updateUdf = new VertexUpdateUdf<VertexKey, VertexValue, Message>(updateFunction, vertexTypes);
-
- // build the update function (co group)
- CoGroupOperator<?, ?, Vertex<VertexKey, VertexValue>> updates =
- messages.coGroup(iteration.getSolutionSet()).where(0).equalTo(0).with(updateUdf);
-
- // configure coGroup update function with name and broadcast variables
- updates = updates.name("Vertex State Updates");
- if (this.configuration != null) {
- for (Tuple2<String, DataSet<?>> e : this.configuration.getUpdateBcastVars()) {
- updates = updates.withBroadcastSet(e.f1, e.f0);
- }
- }
+ // retrieve the direction in which the updates are made and in which the messages are sent
+ EdgeDirection messagingDirection = messagingFunction.getDirection();
- // let the operator know that we preserve the key field
- updates.withForwardedFieldsFirst("0").withForwardedFieldsSecond("0");
-
- return iteration.closeWith(updates, updates);
-
+ DataSet<Tuple2<VertexKey, Message>> messages = null;
--- End diff --
why create a null dataset and pass it as a parameter to the methods below?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-1523][gelly] Vertex centric iteration e...
Posted by andralungu <gi...@git.apache.org>.
Github user andralungu commented on the pull request:
https://github.com/apache/flink/pull/537#issuecomment-97159494
Thanks for the review, @StephanEwen!
PR updated.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-1523][gelly] Vertex centric iteration e...
Posted by vasia <gi...@git.apache.org>.
Github user vasia commented on a diff in the pull request:
https://github.com/apache/flink/pull/537#discussion_r28644140
--- Diff: flink-staging/flink-gelly/src/main/java/org/apache/flink/graph/spargel/VertexUpdateFunction.java ---
@@ -40,7 +42,37 @@
VertexValue extends Serializable, Message> implements Serializable {
private static final long serialVersionUID = 1L;
-
+
+ // --------------------------------------------------------------------------------------------
+ // Attributes that allow vertices to access their in/out degrees and the total number of vertices
+ // inside an iteration.
+ // --------------------------------------------------------------------------------------------
+
+ private long numberOfVertices = -1L;
+
+ public long getNumberOfVertices() throws Exception{
+ if (numberOfVertices == -1) {
+ throw new InaccessibleMethodException("The number of vertices option is not set");
+ }
+ return numberOfVertices;
+ }
+
+ void setNumberOfVertices(long numberOfVertices) {
+ this.numberOfVertices = numberOfVertices;
+ }
+
+ //---------------------------------------------------------------------------------------------
+
+ private boolean optDegrees;
+
+ public boolean isOptDegrees() {
--- End diff --
this shouldn't be public, right?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-1523][gelly] Vertex centric iteration e...
Posted by andralungu <gi...@git.apache.org>.
Github user andralungu commented on a diff in the pull request:
https://github.com/apache/flink/pull/537#discussion_r28764335
--- Diff: flink-staging/flink-gelly/src/main/java/org/apache/flink/graph/spargel/VertexUpdateFunction.java ---
@@ -40,7 +42,37 @@
VertexValue extends Serializable, Message> implements Serializable {
private static final long serialVersionUID = 1L;
-
+
+ // --------------------------------------------------------------------------------------------
+ // Attributes that allow vertices to access their in/out degrees and the total number of vertices
+ // inside an iteration.
+ // --------------------------------------------------------------------------------------------
+
+ private long numberOfVertices = -1L;
+
+ public long getNumberOfVertices() throws Exception{
+ if (numberOfVertices == -1) {
+ throw new InaccessibleMethodException("The number of vertices option is not set");
+ }
+ return numberOfVertices;
+ }
+
+ void setNumberOfVertices(long numberOfVertices) {
+ this.numberOfVertices = numberOfVertices;
+ }
+
+ //---------------------------------------------------------------------------------------------
+
+ private boolean optDegrees;
+
+ public boolean isOptDegrees() {
--- End diff --
Are you talking about the getters? To my knowledge, these are public. Otherwise, they would be totally unusable.
If you're talking about the default setters, I needed to see them in the entire package. But they are not public.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-1523][gelly] Vertex centric iteration e...
Posted by andralungu <gi...@git.apache.org>.
Github user andralungu commented on the pull request:
https://github.com/apache/flink/pull/537#issuecomment-87425953
Hi @vasia ,
1). I will add the documentation once we reach a consensus. No use documenting something that might change :)
2). For the public setter: I still couldn't think of a workaround. Waiting for suggestions.
3). Most of the comments were addressed in this final commit, however, when speaking about avoiding unnecessary overhead, the only way I could do that is to add another VertexCentricIteration class that operates on Tuple3 and have the user decide from the constructor which class he/she wants to use. To me that seems to introduce a ton of duplicate code which will look very bad.
Keeping in mind that I need to store the degrees in the vertex value, any suggestions on how to eliminate this overhead would be very useful.
Thanks!
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-1523][gelly] Vertex centric iteration e...
Posted by andralungu <gi...@git.apache.org>.
Github user andralungu commented on the pull request:
https://github.com/apache/flink/pull/537#issuecomment-93696531
Hi @vasia ,
The PR was updated to include your suggestions. Last test fails for some reason, but that shouldn't get in the way of reviewers :)
As you can see, I went back to suqare one: I needed to test whether the exception is thrown. I followed the degree approach. Hope it's fine!
I would like to ask you to merge this before other vertex-centric PRs because of the following reasons:
1). FCFS
2). I had my share of merge conflicts for now :)
3). I spoke to @markus-h the other day and he said there is no rush with #598 (which I think is a good contribution, by the way)
Thanks!
Andra
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-1523][gelly] Vertex centric iteration e...
Posted by vasia <gi...@git.apache.org>.
Github user vasia commented on a diff in the pull request:
https://github.com/apache/flink/pull/537#discussion_r28644051
--- Diff: flink-staging/flink-gelly/src/main/java/org/apache/flink/graph/spargel/MessagingFunction.java ---
@@ -38,25 +42,57 @@
* @param <Message> The type of the message sent between vertices along the edges.
* @param <EdgeValue> The type of the values that are associated with the edges.
*/
-public abstract class MessagingFunction<VertexKey extends Comparable<VertexKey> & Serializable,
- VertexValue extends Serializable, Message, EdgeValue extends Serializable> implements Serializable {
+public abstract class MessagingFunction<VertexKey extends Comparable<VertexKey> & Serializable,
+ VertexValue extends Serializable, Message, EdgeValue extends Serializable> implements Serializable {
private static final long serialVersionUID = 1L;
-
+
+ // --------------------------------------------------------------------------------------------
+ // Attributes that allow vertices to access their in/out degrees and the total number of vertices
+ // inside an iteration.
+ // --------------------------------------------------------------------------------------------
+
+ private long numberOfVertices = -1L;
+
+ public long getNumberOfVertices() throws Exception{
+ if (numberOfVertices == -1) {
+ throw new InaccessibleMethodException("The number of vertices option is not set");
--- End diff --
Can we make the error message a bit more informative? i.e. explain how to set the option?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-1523][gelly] Vertex centric iteration e...
Posted by vasia <gi...@git.apache.org>.
Github user vasia commented on a diff in the pull request:
https://github.com/apache/flink/pull/537#discussion_r28769333
--- Diff: flink-staging/flink-gelly/src/main/java/org/apache/flink/graph/spargel/VertexUpdateFunction.java ---
@@ -40,7 +42,37 @@
VertexValue extends Serializable, Message> implements Serializable {
private static final long serialVersionUID = 1L;
-
+
+ // --------------------------------------------------------------------------------------------
+ // Attributes that allow vertices to access their in/out degrees and the total number of vertices
+ // inside an iteration.
+ // --------------------------------------------------------------------------------------------
+
+ private long numberOfVertices = -1L;
+
+ public long getNumberOfVertices() throws Exception{
+ if (numberOfVertices == -1) {
+ throw new InaccessibleMethodException("The number of vertices option is not set");
+ }
+ return numberOfVertices;
+ }
+
+ void setNumberOfVertices(long numberOfVertices) {
+ this.numberOfVertices = numberOfVertices;
+ }
+
+ //---------------------------------------------------------------------------------------------
+
+ private boolean optDegrees;
+
+ public boolean isOptDegrees() {
--- End diff --
I'm not sure what I was referring to :confused: You can probably ignore this comment, sorry!
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-1523][gelly] Vertex centric iteration e...
Posted by vasia <gi...@git.apache.org>.
Github user vasia commented on a diff in the pull request:
https://github.com/apache/flink/pull/537#discussion_r28643683
--- Diff: flink-staging/flink-gelly/src/main/java/org/apache/flink/graph/spargel/VertexCentricIteration.java ---
@@ -24,18 +24,24 @@
import org.apache.commons.lang3.Validate;
import org.apache.flink.api.common.aggregators.Aggregator;
+import org.apache.flink.api.common.functions.FlatJoinFunction;
+import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.DataSet;
+import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.operators.DeltaIteration;
import org.apache.flink.api.common.functions.RichCoGroupFunction;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.operators.CoGroupOperator;
import org.apache.flink.api.java.operators.CustomUnaryOperation;
import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.api.java.typeutils.ResultTypeQueryable;
import org.apache.flink.api.java.typeutils.TupleTypeInfo;
import org.apache.flink.api.java.typeutils.TypeExtractor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.graph.Edge;
+import org.apache.flink.graph.EdgeDirection;
+import org.apache.flink.graph.Graph;
import org.apache.flink.graph.Vertex;
import org.apache.flink.util.Collector;
--- End diff --
Could you please also update the description referring to the `withPlainEdges` and `withValuedEdges` methods? This is no longer the case :-)
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-1523][gelly] Vertex centric iteration e...
Posted by vasia <gi...@git.apache.org>.
Github user vasia commented on a diff in the pull request:
https://github.com/apache/flink/pull/537#discussion_r28644150
--- Diff: flink-staging/flink-gelly/src/main/java/org/apache/flink/graph/spargel/MessagingFunction.java ---
@@ -38,25 +42,57 @@
* @param <Message> The type of the message sent between vertices along the edges.
* @param <EdgeValue> The type of the values that are associated with the edges.
*/
-public abstract class MessagingFunction<VertexKey extends Comparable<VertexKey> & Serializable,
- VertexValue extends Serializable, Message, EdgeValue extends Serializable> implements Serializable {
+public abstract class MessagingFunction<VertexKey extends Comparable<VertexKey> & Serializable,
+ VertexValue extends Serializable, Message, EdgeValue extends Serializable> implements Serializable {
private static final long serialVersionUID = 1L;
-
+
+ // --------------------------------------------------------------------------------------------
+ // Attributes that allow vertices to access their in/out degrees and the total number of vertices
+ // inside an iteration.
+ // --------------------------------------------------------------------------------------------
+
+ private long numberOfVertices = -1L;
+
+ public long getNumberOfVertices() throws Exception{
+ if (numberOfVertices == -1) {
+ throw new InaccessibleMethodException("The number of vertices option is not set");
+ }
+ return numberOfVertices;
+ }
+
+ void setNumberOfVertices(long numberOfVertices) {
+ this.numberOfVertices = numberOfVertices;
+ }
+
+ // --------------------------------------------------------------------------------------------
+ // Attribute that allows the user to choose the neighborhood type(in/out/all) on which to run
+ // the vertex centric iteration.
+ // --------------------------------------------------------------------------------------------
+
+ private EdgeDirection direction;
+
+ public EdgeDirection getDirection() {
+ return direction;
+ }
+
+ public void setDirection(EdgeDirection direction) {
--- End diff --
these shouldn't be public either
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-1523][gelly] Vertex centric iteration e...
Posted by andralungu <gi...@git.apache.org>.
Github user andralungu commented on the pull request:
https://github.com/apache/flink/pull/537#issuecomment-99232613
Hey @vasia ,
Everyone is acting as though I do not want to make these methods user-freindly. I do :), it's just not possible.
Keep in mind that in the VertexUpdateFunction and in the MessagingFunction you can only access the vertex in the updateVertex() and sendMessages() methods.
getDirection(), getBroadcastSet() return a single value for all the vertices. getDegree() returns one value for each of the vertices. As previously implied, you cannot access the vertex in these classes. The only very ugly solution is to give it the entire DataSet and to constantly filter it. You basically have the degree of the vertex in the Tuple3, but this information is hidden from the user.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-1523][gelly] Vertex centric iteration e...
Posted by vasia <gi...@git.apache.org>.
Github user vasia commented on a diff in the pull request:
https://github.com/apache/flink/pull/537#discussion_r27256440
--- Diff: flink-staging/flink-gelly/src/main/java/org/apache/flink/graph/Graph.java ---
@@ -1141,8 +1185,33 @@ public boolean filter(Edge<K, EV> edge) {
*/
public <M> Graph<K, VV, EV> runVertexCentricIteration(
VertexCentricIteration<K, VV, M, EV> iteration) {
- DataSet<Vertex<K, VV>> newVertices = vertices.runOperation(iteration);
- return new Graph<K, VV, EV>(newVertices, this.edges, this.context);
+
+ DataSet<Tuple2<K, Long>> inDegrees = this.inDegrees();
+ DataSet<Tuple2<K, Long>> outDegrees = this.outDegrees();
+ DataSet<Tuple3<K, Long, Long>> degress = inDegrees.join(outDegrees).where(0).equalTo(0).with(new FlatJoinFunction<Tuple2<K,Long>, Tuple2<K,Long>, Tuple3<K,Long,Long>>() {
+ @Override
+ public void join(Tuple2<K, Long> first, Tuple2<K, Long> second, Collector<Tuple3<K, Long, Long>> out) throws Exception {
+ out.collect(new Tuple3<K, Long, Long>(first.f0,first.f1,second.f1));
+ }
+ });
+
+ DataSet<Vertex<K, Tuple3<VV,Long,Long>>> verticesWithInDegrees = vertices.join(degress).where(0).equalTo(0).with(
+ new FlatJoinFunction<Vertex<K, VV>, Tuple3<K, Long, Long>, Vertex<K, Tuple3<VV, Long, Long>>>() {
+ @Override
+ public void join(Vertex<K, VV> first, Tuple3<K, Long, Long> second, Collector<Vertex<K, Tuple3<VV, Long, Long>>> out) throws Exception {
+ out.collect(new Vertex<K, Tuple3<VV, Long, Long>>(first.getId(),new Tuple3<VV, Long, Long>(first.getValue(),second.f1,second.f2)));
+ }
+ }
+ ).rebalance();
--- End diff --
why rebalance?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-1523][gelly] Vertex centric iteration e...
Posted by vasia <gi...@git.apache.org>.
Github user vasia commented on a diff in the pull request:
https://github.com/apache/flink/pull/537#discussion_r27255825
--- Diff: flink-staging/flink-gelly/src/main/java/org/apache/flink/graph/Graph.java ---
@@ -1118,17 +1118,61 @@ public boolean filter(Edge<K, EV> edge) {
}
/**
- * Create a Vertex-Centric iteration on the graph.
- *
+ * Create a Vertex-Centric iteration on the graph. inDegrees, outDegrees
+ * and the total number of vertices are set here in order to be retrieved
+ * later on within the iteration.
+ *
+ * Also, The user can define the direction in which the updates are performed
+ * and in which the messages are sent. The function below tackles the regular
+ * use-case where updates of the in-neighbors are used to calculate state
+ * and messages are sent to out-neighbors.
+ *
* @param vertexUpdateFunction the vertex update function
* @param messagingFunction the messaging function
* @param maximumNumberOfIterations maximum number of iterations to perform
* @return
*/
- public <M> VertexCentricIteration<K, VV, M, EV> createVertexCentricIteration(
+ public <M, NV extends Serializable> VertexCentricIteration<K, VV, M, EV> createVertexCentricIteration(
--- End diff --
NV is never used?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-1523][gelly] Vertex centric iteration e...
Posted by andralungu <gi...@git.apache.org>.
Github user andralungu commented on a diff in the pull request:
https://github.com/apache/flink/pull/537#discussion_r28812911
--- Diff: flink-staging/flink-gelly/src/main/java/org/apache/flink/graph/spargel/VertexCentricIteration.java ---
@@ -228,48 +213,92 @@ public void setInput(DataSet<Vertex<VertexKey, VertexValue>> inputData) {
{
return new VertexCentricIteration<VertexKey, VertexValue, Message, EdgeValue>(uf, mf, edgesWithValue, maximumNumberOfIterations);
}
-
+
+ /**
+ * Configures this vertex-centric iteration with the provided parameters.
+ *
+ * @param parameters the configuration parameters
+ */
+ public void configure(IterationConfiguration parameters) {
+ this.configuration = parameters;
+ }
+
+ /**
+ * @return the configuration parameters of this vertex-centric iteration
+ */
+ public IterationConfiguration getIterationConfiguration() {
+ return this.configuration;
+ }
+
// --------------------------------------------------------------------------------------------
// Wrapping UDFs
// --------------------------------------------------------------------------------------------
- private static final class VertexUpdateUdf<VertexKey extends Comparable<VertexKey> & Serializable,
- VertexValue extends Serializable, Message>
- extends RichCoGroupFunction<Tuple2<VertexKey, Message>, Vertex<VertexKey, VertexValue>, Vertex<VertexKey, VertexValue>>
- implements ResultTypeQueryable<Vertex<VertexKey, VertexValue>>
+ private static abstract class VertexUpdateUdf<VertexKey extends Comparable<VertexKey> & Serializable,
+ VV extends Serializable, VertexValue extends Serializable, Message>
--- End diff --
Perhaps I chose a name that is a bit misleading, here VV is replaced by the Tuple3<VertexValue, InDegree, OutDegree> when you have the degrees option set or simply VertexValue if the degrees are not set. I still need the VertexValue generic type. I can rename VertexValue to VV, EdgeValue to EV, etc and VV to something else? Something like VertexValueMessagingFunction. I can't think of a better name :)
Tell me if my intention is clearer now and how to proceed!
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-1523][gelly] Vertex centric iteration e...
Posted by vasia <gi...@git.apache.org>.
Github user vasia commented on a diff in the pull request:
https://github.com/apache/flink/pull/537#discussion_r28643395
--- Diff: docs/gelly_guide.md ---
@@ -430,6 +440,46 @@ public static final class Messenger extends MessagingFunction {...}
{% endhighlight %}
+The following example illustrates the usage of the degree, the number of vertices as well as the messaging direction options.
+
+{% highlight java %}
+
+Graph<Long, Double, Double> graph = ...
+
+// configure the iteration
+IterationConfiguration parameters = new IterationConfiguration();
+
+// set the messaging direction
+parameters.setDirection(EdgeDirection.IN);
--- End diff --
This is not used anywhere in the example. I would remove it from here and add another example only for the direction, together with the figures I mention above.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-1523][gelly] Vertex centric iteration e...
Posted by vasia <gi...@git.apache.org>.
Github user vasia commented on a diff in the pull request:
https://github.com/apache/flink/pull/537#discussion_r27256784
--- Diff: flink-staging/flink-gelly/src/main/java/org/apache/flink/graph/library/LabelPropagation.java ---
@@ -68,12 +70,12 @@ public LabelPropagation(int maxIterations) {
public static final class UpdateVertexLabel<K extends Comparable<K> & Serializable>
extends VertexUpdateFunction<K, Long, Long> {
- public void updateVertex(K vertexKey, Long vertexValue,
+ public void updateVertex(Vertex<K, Tuple3<Long,Long,Long>> vertex,
--- End diff --
as in the comment below, this should be `Vertex<K, Long>`
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-1523][gelly] Vertex centric iteration e...
Posted by vasia <gi...@git.apache.org>.
Github user vasia commented on a diff in the pull request:
https://github.com/apache/flink/pull/537#discussion_r28643920
--- Diff: flink-staging/flink-gelly/src/main/java/org/apache/flink/graph/spargel/VertexCentricIteration.java ---
@@ -138,69 +146,46 @@ public void setInput(DataSet<Vertex<VertexKey, VertexValue>> inputData) {
if (this.initialVertices == null) {
--- End diff --
I find the new logic of this method a bit confusing. Why do you set some configuration parameters here (degrees and direction) and the rest in the helper methods?
As far as I understand, you have 2 cases that matter here: the degrees option is set or not. This defines whether you create a simple vertex or you augment it with degree information, right?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-1523][gelly] Vertex centric iteration e...
Posted by vasia <gi...@git.apache.org>.
Github user vasia commented on a diff in the pull request:
https://github.com/apache/flink/pull/537#discussion_r27257390
--- Diff: flink-staging/flink-gelly/src/main/java/org/apache/flink/graph/spargel/VertexUpdateFunction.java ---
@@ -40,7 +41,22 @@
VertexValue extends Serializable, Message> implements Serializable {
private static final long serialVersionUID = 1L;
-
+
+ // --------------------------------------------------------------------------------------------
+ // Attributes that allow vertices to access their in/out degrees and the total number of vertices
+ // inside an iteration.
+ // --------------------------------------------------------------------------------------------
+
+ private long numberOfVertices;
+
+ public long getNumberOfVertices() {
+ return numberOfVertices;
+ }
+
+ public void setNumberOfVertices(long numberOfVertices) {
--- End diff --
same here
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-1523][gelly] Vertex centric iteration e...
Posted by vasia <gi...@git.apache.org>.
Github user vasia commented on a diff in the pull request:
https://github.com/apache/flink/pull/537#discussion_r28643441
--- Diff: flink-staging/flink-gelly/src/main/java/org/apache/flink/graph/example/IncrementalSSSPExample.java ---
@@ -0,0 +1,304 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.graph.example;
+
+import org.apache.flink.api.common.ProgramDescription;
+import org.apache.flink.api.common.functions.FilterFunction;
+import org.apache.flink.api.common.functions.MapFunction;
+import org.apache.flink.api.java.DataSet;
+import org.apache.flink.api.java.ExecutionEnvironment;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.api.java.tuple.Tuple3;
+import org.apache.flink.graph.Edge;
+import org.apache.flink.graph.EdgeDirection;
+import org.apache.flink.graph.Graph;
+import org.apache.flink.graph.Vertex;
+import org.apache.flink.graph.example.utils.IncrementalSSSPData;
+import org.apache.flink.graph.spargel.IterationConfiguration;
+import org.apache.flink.graph.spargel.MessageIterator;
+import org.apache.flink.graph.spargel.MessagingFunction;
+import org.apache.flink.graph.spargel.VertexUpdateFunction;
+
+/**
+ * Incremental Single Sink Shortest Paths Example.
+ *
+ * The program takes as input the resulted graph after a SSSP computation,
+ * an edge to be removed and the initial graph(i.e. before SSSP was computed).
+ *
+ * - If the removed edge does not belong to the SP-graph, no computation is necessary.
--- End diff --
maybe we should explain what is an "SP-graph"?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-1523][gelly] Vertex centric iteration e...
Posted by StephanEwen <gi...@git.apache.org>.
Github user StephanEwen commented on the pull request:
https://github.com/apache/flink/pull/537#issuecomment-96558324
Looks good in general. A few thoughts on this:
Your prior discussion involved efficiency, and Vasia suggested to not carry the degrees in all cases (as they are not needed in most cases). It seems that was not yet realized, because the Vertex class always carries the degrees.
I think we can improve the abstraction between the with-degree and without-degree case a bit. Cases where one has to throw a "not supported" can usually be improved with a good inheritance hierarchy. Does it work to make the VertexWithDegrees class a subclass of the Vertex class?
Also, as a bit of background: Vertex is a subclass of "Tuple2". Tuples are currently the fastest data types in Flink. By adding additional Fields to the Tuple, you are making it a POJO (as far as I know), which is a bit slower to serialize type.
Also: primitives are usually better than boxed types. Prefer `long` over `Long` where possible.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-1523][gelly] Vertex centric iteration e...
Posted by vasia <gi...@git.apache.org>.
Github user vasia commented on a diff in the pull request:
https://github.com/apache/flink/pull/537#discussion_r28643940
--- Diff: flink-staging/flink-gelly/src/main/java/org/apache/flink/graph/spargel/VertexCentricIteration.java ---
@@ -228,48 +213,92 @@ public void setInput(DataSet<Vertex<VertexKey, VertexValue>> inputData) {
{
return new VertexCentricIteration<VertexKey, VertexValue, Message, EdgeValue>(uf, mf, edgesWithValue, maximumNumberOfIterations);
}
-
+
+ /**
+ * Configures this vertex-centric iteration with the provided parameters.
+ *
+ * @param parameters the configuration parameters
+ */
+ public void configure(IterationConfiguration parameters) {
+ this.configuration = parameters;
+ }
+
+ /**
+ * @return the configuration parameters of this vertex-centric iteration
+ */
+ public IterationConfiguration getIterationConfiguration() {
+ return this.configuration;
+ }
+
// --------------------------------------------------------------------------------------------
// Wrapping UDFs
// --------------------------------------------------------------------------------------------
- private static final class VertexUpdateUdf<VertexKey extends Comparable<VertexKey> & Serializable,
- VertexValue extends Serializable, Message>
- extends RichCoGroupFunction<Tuple2<VertexKey, Message>, Vertex<VertexKey, VertexValue>, Vertex<VertexKey, VertexValue>>
- implements ResultTypeQueryable<Vertex<VertexKey, VertexValue>>
+ private static abstract class VertexUpdateUdf<VertexKey extends Comparable<VertexKey> & Serializable,
+ VV extends Serializable, VertexValue extends Serializable, Message>
--- End diff --
If we change VertexValue type to VV, we should be consistent and change it throughout the class. We could also simplify VertexKey to K and EdgeValue to EV to be consistent with the rest of the classes in Gelly.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-1523][gelly] Vertex centric iteration e...
Posted by vasia <gi...@git.apache.org>.
Github user vasia commented on a diff in the pull request:
https://github.com/apache/flink/pull/537#discussion_r28643459
--- Diff: flink-staging/flink-gelly/src/main/java/org/apache/flink/graph/example/IncrementalSSSPExample.java ---
@@ -0,0 +1,304 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.graph.example;
+
+import org.apache.flink.api.common.ProgramDescription;
+import org.apache.flink.api.common.functions.FilterFunction;
+import org.apache.flink.api.common.functions.MapFunction;
+import org.apache.flink.api.java.DataSet;
+import org.apache.flink.api.java.ExecutionEnvironment;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.api.java.tuple.Tuple3;
+import org.apache.flink.graph.Edge;
+import org.apache.flink.graph.EdgeDirection;
+import org.apache.flink.graph.Graph;
+import org.apache.flink.graph.Vertex;
+import org.apache.flink.graph.example.utils.IncrementalSSSPData;
+import org.apache.flink.graph.spargel.IterationConfiguration;
+import org.apache.flink.graph.spargel.MessageIterator;
+import org.apache.flink.graph.spargel.MessagingFunction;
+import org.apache.flink.graph.spargel.VertexUpdateFunction;
+
+/**
+ * Incremental Single Sink Shortest Paths Example.
+ *
+ * The program takes as input the resulted graph after a SSSP computation,
+ * an edge to be removed and the initial graph(i.e. before SSSP was computed).
+ *
+ * - If the removed edge does not belong to the SP-graph, no computation is necessary.
+ * The edge is simply removed from the graph.
+ * - If the removed edge is an SP-edge, then all nodes, whose shortest path contains the removed edge,
+ * potentially require re-computation.
+ * When the edge <u, v> is removed, v checks if it has another out-going SP-edge.
+ * If yes, no further computation is required.
+ * If v has no other out-going SP-edge, it invalidates its current value, by setting it to INF.
+ * Then, it informs all its SP-in-neighbors by sending them an INVALIDATE message.
+ * When a vertex u receives an INVALIDATE message from v, it checks whether it has another out-going SP-edge.
+ * If not, it invalidates its current value and propagates the INVALIDATE message.
+ * The propagation stops when a vertex with an alternative shortest path is reached
+ * or when we reach a vertex with no SP-in-neighbors.
+ *
+ * Usage <code>IncrementalSSSPExample <vertex path> <edge path> <edges in SSSP>
+ * <edge to be removed> <result path> <number of iterations></code><br>
+ * If no parameters are provided, the program is run with default data from
+ * {@link org.apache.flink.graph.example.utils.IncrementalSSSPData}
+ */
+@SuppressWarnings("serial")
+public class IncrementalSSSPExample implements ProgramDescription {
+
+ public static void main(String [] args) throws Exception {
+
+ if(!parseParameters(args)) {
+ return;
+ }
+
+ ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+
+ DataSet<Vertex<Long, Double>> vertices = getVerticesDataSet(env);
+
+ DataSet<Edge<Long, Double>> edges = getEdgesDataSet(env);
+
+ DataSet<Edge<Long, Double>> edgesInSSSP = getEdgesinSSSPDataSet(env);
+
+ Edge<Long, Double> edgeToBeRemoved = getEdgeToBeRemoved();
+
+ Graph<Long, Double, Double> graph = Graph.fromDataSet(vertices, edges, env);
+
+ // Assumption: all minimum weight paths are kept
+ Graph<Long, Double, Double> ssspGraph = Graph.fromDataSet(vertices, edgesInSSSP, env);
+
+ // remove the edge
+ graph.removeEdge(edgeToBeRemoved);
+
+ // configure the iteration
+ IterationConfiguration parameters = new IterationConfiguration();
+
+ if(isInSSSP(edgeToBeRemoved, edgesInSSSP)) {
+
+ parameters.setDirection(EdgeDirection.IN);
+ parameters.setOptDegrees(true);
+
+ // run the vertex centric iteration to propagate info
+ Graph<Long, Double, Double> result = ssspGraph.runVertexCentricIteration(new VertexDistanceUpdater(),
+ new InvalidateMessenger(edgeToBeRemoved), maxIterations, parameters);
+
+ DataSet<Vertex<Long, Double>> resultedVertices = result.getVertices();
+
+ // Emit results
+ if(fileOutput) {
+ resultedVertices.writeAsCsv(outputPath, "\n", ",");
+ } else {
+ resultedVertices.print();
+ }
+
+ env.execute("Incremental SSSP Example");
+ } else {
+ // print the vertices
+ if(fileOutput) {
+ vertices.writeAsCsv(outputPath, "\n", ",");
+ } else {
+ vertices.print();
+ }
+
+ env.execute("Incremental SSSP Example");
+ }
+ }
+
+ @Override
+ public String getDescription() {
+ return "Incremental Single Sink Shortest Paths Example";
+ }
+
+ // ******************************************************************************************************************
+ // IncrementalSSSP METHODS
+ // ******************************************************************************************************************
+
+ /**
+ * Function that verifies whether the edge to be removed is part of the SSSP or not.
+ * If it is, the src vertex will be invalidated.
+ *
+ * @param edgeToBeRemoved
+ * @param edgesInSSSP
+ * @return
+ */
+ private static boolean isInSSSP(final Edge<Long, Double> edgeToBeRemoved, DataSet<Edge<Long, Double>> edgesInSSSP) throws Exception {
+
+ return edgesInSSSP.filter(new FilterFunction<Edge<Long, Double>>() {
+ @Override
+ public boolean filter(Edge<Long, Double> edge) throws Exception {
+ return edge.equals(edgeToBeRemoved);
+ }
+ }).count() > 0;
+ }
+
+ public static final class VertexDistanceUpdater extends VertexUpdateFunction<Long, Double, Double> {
+
+ @Override
+ public void updateVertex(Vertex<Long, Double> vertex, MessageIterator<Double> inMessages) throws Exception {
+ if (inMessages.hasNext()) {
+ Long outDegree = vertex.getOutDegree() - 1;
+ // check if the vertex has another SP-Edge
+ if (outDegree > 0) {
+ // there is another shortest path from the source to this vertex
+ } else {
+ // set own value to infinity
+ setNewVertexValue(Double.MAX_VALUE);
+ }
+ }
+ }
+ }
+
+ public static final class InvalidateMessenger extends MessagingFunction<Long, Double, Double, Double> {
+
+ private Edge<Long, Double> edgeToBeRemoved;
+
+ public InvalidateMessenger(Edge<Long, Double> edgeToBeRemoved) {
+ this.edgeToBeRemoved = edgeToBeRemoved;
+ }
+
+ @Override
+ public void sendMessages(Vertex<Long, Double> vertex) throws Exception {
+
+
+ if(getSuperstepNumber() == 1) {
+ if(vertex.getId().equals(edgeToBeRemoved.getSource())) {
+ // activate the edge target
+ sendMessageTo(edgeToBeRemoved.getSource(), Double.MAX_VALUE);
+ }
+ }
+
+ if(getSuperstepNumber() > 1) {
+ // invalidate all edges
+ for(Edge<Long, Double> edge : getEdges()) {
+ sendMessageTo(edge.getSource(), Double.MAX_VALUE);
+ }
+ }
+ }
+ }
+
+ // ******************************************************************************************************************
+ // UTIL METHODS
+ // ******************************************************************************************************************
+
+ private static boolean fileOutput = false;
+
+ private static String verticesInputPath = null;
+
+ private static String edgesInputPath = null;
+
+ private static String edgesInSSSPInputPath = null;
+
+ private static String edgeToBeRemoved = null;
+
+ private static String outputPath = null;
+
+ private static int maxIterations = 5;
+
+ private static boolean parseParameters(String[] args) {
+ if (args.length > 0) {
+ if (args.length == 6) {
+ fileOutput = true;
+ verticesInputPath = args[0];
+ edgesInputPath = args[1];
+ edgesInSSSPInputPath = args[2];
+ edgeToBeRemoved = args[3];
+ outputPath = args[4];
+ maxIterations = Integer.parseInt(args[5]);
+ } else {
+ System.out.println("Executing IncrementalSSSP example with default parameters and built-in default data.");
+ System.out.println("Provide parameters to read input data from files.");
+ System.out.println("See the documentation for the correct format of input files.");
+ System.out.println("Usage: IncrementalSSSP <vertex path> <edge path> <edges in SSSP> <edge to be removed> <output path> <max iterations>");
+
+ return false;
+ }
+ }
+ return true;
+ }
+
+ private static DataSet<Vertex<Long, Double>> getVerticesDataSet(ExecutionEnvironment env) {
+ if (fileOutput) {
+ return env.readCsvFile(verticesInputPath)
+ .lineDelimiter("\n")
+ .types(Long.class, Double.class)
+ .map(new MapFunction<Tuple2<Long, Double>, Vertex<Long, Double>>() {
--- End diff --
we have a Tuple2ToVertex utility mapper :-)
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-1523][gelly] Vertex centric iteration e...
Posted by vasia <gi...@git.apache.org>.
Github user vasia commented on a diff in the pull request:
https://github.com/apache/flink/pull/537#discussion_r28643385
--- Diff: docs/gelly_guide.md ---
@@ -430,6 +440,46 @@ public static final class Messenger extends MessagingFunction {...}
{% endhighlight %}
+The following example illustrates the usage of the degree, the number of vertices as well as the messaging direction options.
+
+{% highlight java %}
+
+Graph<Long, Double, Double> graph = ...
+
+// configure the iteration
+IterationConfiguration parameters = new IterationConfiguration();
+
+// set the messaging direction
+parameters.setDirection(EdgeDirection.IN);
+
+// set the number of vertices option to true
+parameters.setOptNumVertices(true);
+
+// set the degree option to true
+parameters.setOptDegrees(true);
+
+// run the vertex-centric iteration, also passing the configuration parameters
+Graph<Long, Double, Double> result =
+ graph.runVertexCentricIteration(
+ VertexDistanceUpdater(), new MinDistanceMessenger(), maxIterations, parameters);
--- End diff --
A "new" is missing. Maybe we can use the shorter function names "VertexUpdater" and "Messenger" as in the previous example, otherwise the last line is split into two.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-1523][gelly] Vertex centric iteration e...
Posted by vasia <gi...@git.apache.org>.
Github user vasia commented on the pull request:
https://github.com/apache/flink/pull/537#issuecomment-92301058
Hey @andralungu!
Now that #547 is merged, the degrees, total number of vertices and direction can be added as configuration options. This should also simplify writing the tests. Take a look at `VertexCentricConfigurationITCase`. You should be able to cover some of the cases by extending it.
I hope rebasing won't be a big pain :-)
If an option is not set and the user tries to call one of the methods that isn't computed, then we could display a message to inform them that the corresponding option is not set and how to enable it.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-1523][gelly] Vertex centric iteration e...
Posted by andralungu <gi...@git.apache.org>.
Github user andralungu commented on a diff in the pull request:
https://github.com/apache/flink/pull/537#discussion_r28811435
--- Diff: flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/example/IncrementalSSSPITCase.java ---
@@ -0,0 +1,84 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.graph.test.example;
+
+import com.google.common.base.Charsets;
+import com.google.common.io.Files;
+import org.apache.flink.graph.example.IncrementalSSSPExample;
+import org.apache.flink.graph.example.utils.IncrementalSSSPData;
+import org.apache.flink.test.util.MultipleProgramsTestBase;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+
+import java.io.File;
+
+@RunWith(Parameterized.class)
+public class IncrementalSSSPITCase extends MultipleProgramsTestBase {
+
+ private String verticesPath;
+
+ private String edgesPath;
+
+ private String edgesInSSSPPath;
+
+ private String resultPath;
+
+ private String expected;
+
+ @Rule
+ public TemporaryFolder tempFolder = new TemporaryFolder();
+
+ public IncrementalSSSPITCase(TestExecutionMode mode) {
+ super(mode);
+ }
+
+ @Before
+ public void before() throws Exception {
+ resultPath = tempFolder.newFile().toURI().toString();
+ File verticesFile = tempFolder.newFile();
+ Files.write(IncrementalSSSPData.VERTICES, verticesFile, Charsets.UTF_8);
+
+ File edgesFile = tempFolder.newFile();
+ Files.write(IncrementalSSSPData.EDGES, edgesFile, Charsets.UTF_8);
+
+ File edgesInSSSPFile = tempFolder.newFile();
+ Files.write(IncrementalSSSPData.EDGES_IN_SSSP, edgesInSSSPFile, Charsets.UTF_8);
+
+ verticesPath = verticesFile.toURI().toString();
+ edgesPath = edgesFile.toURI().toString();
+ edgesInSSSPPath = edgesInSSSPFile.toURI().toString();
+ }
+
+ @Test
--- End diff --
The test for removing the non-SP-edge is doable. It's the bigger graph test case that concerns me. Of course, we can generate some randomised edges, but how do we know which of those edges are in SSSP? That can only be done if we have the actual algorithm implemented.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-1523][gelly] Vertex centric iteration e...
Posted by vasia <gi...@git.apache.org>.
Github user vasia commented on a diff in the pull request:
https://github.com/apache/flink/pull/537#discussion_r28643473
--- Diff: flink-staging/flink-gelly/src/main/java/org/apache/flink/graph/example/IncrementalSSSPExample.java ---
@@ -0,0 +1,304 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.graph.example;
+
+import org.apache.flink.api.common.ProgramDescription;
+import org.apache.flink.api.common.functions.FilterFunction;
+import org.apache.flink.api.common.functions.MapFunction;
+import org.apache.flink.api.java.DataSet;
+import org.apache.flink.api.java.ExecutionEnvironment;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.api.java.tuple.Tuple3;
+import org.apache.flink.graph.Edge;
+import org.apache.flink.graph.EdgeDirection;
+import org.apache.flink.graph.Graph;
+import org.apache.flink.graph.Vertex;
+import org.apache.flink.graph.example.utils.IncrementalSSSPData;
+import org.apache.flink.graph.spargel.IterationConfiguration;
+import org.apache.flink.graph.spargel.MessageIterator;
+import org.apache.flink.graph.spargel.MessagingFunction;
+import org.apache.flink.graph.spargel.VertexUpdateFunction;
+
+/**
+ * Incremental Single Sink Shortest Paths Example.
+ *
+ * The program takes as input the resulted graph after a SSSP computation,
+ * an edge to be removed and the initial graph(i.e. before SSSP was computed).
+ *
+ * - If the removed edge does not belong to the SP-graph, no computation is necessary.
+ * The edge is simply removed from the graph.
+ * - If the removed edge is an SP-edge, then all nodes, whose shortest path contains the removed edge,
+ * potentially require re-computation.
+ * When the edge <u, v> is removed, v checks if it has another out-going SP-edge.
+ * If yes, no further computation is required.
+ * If v has no other out-going SP-edge, it invalidates its current value, by setting it to INF.
+ * Then, it informs all its SP-in-neighbors by sending them an INVALIDATE message.
+ * When a vertex u receives an INVALIDATE message from v, it checks whether it has another out-going SP-edge.
+ * If not, it invalidates its current value and propagates the INVALIDATE message.
+ * The propagation stops when a vertex with an alternative shortest path is reached
+ * or when we reach a vertex with no SP-in-neighbors.
+ *
+ * Usage <code>IncrementalSSSPExample <vertex path> <edge path> <edges in SSSP>
+ * <edge to be removed> <result path> <number of iterations></code><br>
+ * If no parameters are provided, the program is run with default data from
+ * {@link org.apache.flink.graph.example.utils.IncrementalSSSPData}
+ */
+@SuppressWarnings("serial")
+public class IncrementalSSSPExample implements ProgramDescription {
+
+ public static void main(String [] args) throws Exception {
+
+ if(!parseParameters(args)) {
+ return;
+ }
+
+ ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+
+ DataSet<Vertex<Long, Double>> vertices = getVerticesDataSet(env);
+
+ DataSet<Edge<Long, Double>> edges = getEdgesDataSet(env);
+
+ DataSet<Edge<Long, Double>> edgesInSSSP = getEdgesinSSSPDataSet(env);
+
+ Edge<Long, Double> edgeToBeRemoved = getEdgeToBeRemoved();
+
+ Graph<Long, Double, Double> graph = Graph.fromDataSet(vertices, edges, env);
+
+ // Assumption: all minimum weight paths are kept
+ Graph<Long, Double, Double> ssspGraph = Graph.fromDataSet(vertices, edgesInSSSP, env);
+
+ // remove the edge
+ graph.removeEdge(edgeToBeRemoved);
+
+ // configure the iteration
+ IterationConfiguration parameters = new IterationConfiguration();
+
+ if(isInSSSP(edgeToBeRemoved, edgesInSSSP)) {
+
+ parameters.setDirection(EdgeDirection.IN);
+ parameters.setOptDegrees(true);
+
+ // run the vertex centric iteration to propagate info
+ Graph<Long, Double, Double> result = ssspGraph.runVertexCentricIteration(new VertexDistanceUpdater(),
+ new InvalidateMessenger(edgeToBeRemoved), maxIterations, parameters);
+
+ DataSet<Vertex<Long, Double>> resultedVertices = result.getVertices();
+
+ // Emit results
+ if(fileOutput) {
+ resultedVertices.writeAsCsv(outputPath, "\n", ",");
+ } else {
+ resultedVertices.print();
+ }
+
+ env.execute("Incremental SSSP Example");
+ } else {
+ // print the vertices
+ if(fileOutput) {
+ vertices.writeAsCsv(outputPath, "\n", ",");
+ } else {
+ vertices.print();
+ }
+
+ env.execute("Incremental SSSP Example");
+ }
+ }
+
+ @Override
+ public String getDescription() {
+ return "Incremental Single Sink Shortest Paths Example";
+ }
+
+ // ******************************************************************************************************************
+ // IncrementalSSSP METHODS
+ // ******************************************************************************************************************
+
+ /**
+ * Function that verifies whether the edge to be removed is part of the SSSP or not.
+ * If it is, the src vertex will be invalidated.
+ *
+ * @param edgeToBeRemoved
+ * @param edgesInSSSP
+ * @return
+ */
+ private static boolean isInSSSP(final Edge<Long, Double> edgeToBeRemoved, DataSet<Edge<Long, Double>> edgesInSSSP) throws Exception {
+
+ return edgesInSSSP.filter(new FilterFunction<Edge<Long, Double>>() {
+ @Override
+ public boolean filter(Edge<Long, Double> edge) throws Exception {
+ return edge.equals(edgeToBeRemoved);
+ }
+ }).count() > 0;
+ }
+
+ public static final class VertexDistanceUpdater extends VertexUpdateFunction<Long, Double, Double> {
+
+ @Override
+ public void updateVertex(Vertex<Long, Double> vertex, MessageIterator<Double> inMessages) throws Exception {
+ if (inMessages.hasNext()) {
+ Long outDegree = vertex.getOutDegree() - 1;
+ // check if the vertex has another SP-Edge
+ if (outDegree > 0) {
+ // there is another shortest path from the source to this vertex
+ } else {
+ // set own value to infinity
+ setNewVertexValue(Double.MAX_VALUE);
+ }
+ }
+ }
+ }
+
+ public static final class InvalidateMessenger extends MessagingFunction<Long, Double, Double, Double> {
+
+ private Edge<Long, Double> edgeToBeRemoved;
+
+ public InvalidateMessenger(Edge<Long, Double> edgeToBeRemoved) {
+ this.edgeToBeRemoved = edgeToBeRemoved;
+ }
+
+ @Override
+ public void sendMessages(Vertex<Long, Double> vertex) throws Exception {
+
+
+ if(getSuperstepNumber() == 1) {
+ if(vertex.getId().equals(edgeToBeRemoved.getSource())) {
+ // activate the edge target
+ sendMessageTo(edgeToBeRemoved.getSource(), Double.MAX_VALUE);
+ }
+ }
+
+ if(getSuperstepNumber() > 1) {
+ // invalidate all edges
+ for(Edge<Long, Double> edge : getEdges()) {
+ sendMessageTo(edge.getSource(), Double.MAX_VALUE);
+ }
+ }
+ }
+ }
+
+ // ******************************************************************************************************************
+ // UTIL METHODS
+ // ******************************************************************************************************************
+
+ private static boolean fileOutput = false;
+
+ private static String verticesInputPath = null;
+
+ private static String edgesInputPath = null;
+
+ private static String edgesInSSSPInputPath = null;
+
+ private static String edgeToBeRemoved = null;
+
+ private static String outputPath = null;
+
+ private static int maxIterations = 5;
+
+ private static boolean parseParameters(String[] args) {
+ if (args.length > 0) {
+ if (args.length == 6) {
+ fileOutput = true;
+ verticesInputPath = args[0];
+ edgesInputPath = args[1];
+ edgesInSSSPInputPath = args[2];
+ edgeToBeRemoved = args[3];
+ outputPath = args[4];
+ maxIterations = Integer.parseInt(args[5]);
+ } else {
+ System.out.println("Executing IncrementalSSSP example with default parameters and built-in default data.");
+ System.out.println("Provide parameters to read input data from files.");
+ System.out.println("See the documentation for the correct format of input files.");
+ System.out.println("Usage: IncrementalSSSP <vertex path> <edge path> <edges in SSSP> <edge to be removed> <output path> <max iterations>");
+
+ return false;
+ }
+ }
+ return true;
+ }
+
+ private static DataSet<Vertex<Long, Double>> getVerticesDataSet(ExecutionEnvironment env) {
+ if (fileOutput) {
+ return env.readCsvFile(verticesInputPath)
+ .lineDelimiter("\n")
+ .types(Long.class, Double.class)
+ .map(new MapFunction<Tuple2<Long, Double>, Vertex<Long, Double>>() {
+
+ @Override
+ public Vertex<Long, Double> map(Tuple2<Long, Double> tuple2) throws Exception {
+ return new Vertex<Long, Double>(tuple2.f0, tuple2.f1);
+ }
+ });
+ } else {
+ System.err.println("Usage: IncrementalSSSP <vertex path> <edge path> <edges in SSSP> <edge to be removed> " +
+ "<output path> <max iterations>");
+ return IncrementalSSSPData.getDefaultVertexDataSet(env);
+ }
+ }
+
+ private static DataSet<Edge<Long, Double>> getEdgesDataSet(ExecutionEnvironment env) {
+ if (fileOutput) {
+ return env.readCsvFile(edgesInputPath)
+ .lineDelimiter("\n")
+ .types(Long.class, Long.class, Double.class)
+ .map(new MapFunction<Tuple3<Long, Long, Double>, Edge<Long, Double>>() {
+
+ @Override
+ public Edge<Long, Double> map(Tuple3<Long, Long, Double> tuple3) throws Exception {
+ return new Edge(tuple3.f0, tuple3.f1, tuple3.f2);
+ }
+ });
+ } else {
+ System.err.println("Usage: IncrementalSSSP <vertex path> <edge path> <edges in SSSP> <edge to be removed> " +
+ "<output path> <max iterations>");
+ return IncrementalSSSPData.getDefaultEdgeDataSet(env);
+ }
+ }
+
+ private static DataSet<Edge<Long, Double>> getEdgesinSSSPDataSet(ExecutionEnvironment env) {
+ if (fileOutput) {
+ return env.readCsvFile(edgesInSSSPInputPath)
+ .lineDelimiter("\n")
+ .types(Long.class, Long.class, Double.class)
+ .map(new MapFunction<Tuple3<Long, Long, Double>, Edge<Long, Double>>() {
+
+ @Override
+ public Edge<Long, Double> map(Tuple3<Long, Long, Double> tuple3) throws Exception {
+ return new Edge(tuple3.f0, tuple3.f1, tuple3.f2);
+ }
+ });
+ } else {
+ System.err.println("Usage: IncrementalSSSP <vertex path> <edge path> <edges in SSSP> <edge to be removed> " +
+ "<output path> <max iterations>");
+ return IncrementalSSSPData.getDefaultEdgesInSSSP(env);
+ }
+ }
+
+ private static Edge<Long, Double> getEdgeToBeRemoved() {
--- End diff --
how about we give this parameter as a srcId and a trgId (of the edge) instead of giving a file path for a single pair of ids?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-1523][gelly] Vertex centric iteration e...
Posted by andralungu <gi...@git.apache.org>.
Github user andralungu commented on a diff in the pull request:
https://github.com/apache/flink/pull/537#discussion_r27344502
--- Diff: flink-staging/flink-gelly/src/main/java/org/apache/flink/graph/spargel/MessagingFunction.java ---
@@ -38,11 +41,41 @@
* @param <Message> The type of the message sent between vertices along the edges.
* @param <EdgeValue> The type of the values that are associated with the edges.
*/
-public abstract class MessagingFunction<VertexKey extends Comparable<VertexKey> & Serializable,
- VertexValue extends Serializable, Message, EdgeValue extends Serializable> implements Serializable {
+public abstract class MessagingFunction<VertexKey extends Comparable<VertexKey> & Serializable,
+ VertexValue extends Serializable, Message, EdgeValue extends Serializable> implements Serializable {
private static final long serialVersionUID = 1L;
-
+
+ // --------------------------------------------------------------------------------------------
+ // Attributes that allow vertices to access their in/out degrees and the total number of vertices
+ // inside an iteration.
+ // --------------------------------------------------------------------------------------------
+
+ private long numberOfVertices;
+
+ public long getNumberOfVertices() {
+ return numberOfVertices;
+ }
+
+ public void setNumberOfVertices(long numberOfVertices) {
--- End diff --
Normally it should be private. However, I need to access that setter in Graph's createVertexCentricIteration. If you can think of an alternative, let me know.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-1523][gelly] Vertex centric iteration e...
Posted by vasia <gi...@git.apache.org>.
Github user vasia commented on a diff in the pull request:
https://github.com/apache/flink/pull/537#discussion_r28643470
--- Diff: flink-staging/flink-gelly/src/main/java/org/apache/flink/graph/example/IncrementalSSSPExample.java ---
@@ -0,0 +1,304 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.graph.example;
+
+import org.apache.flink.api.common.ProgramDescription;
+import org.apache.flink.api.common.functions.FilterFunction;
+import org.apache.flink.api.common.functions.MapFunction;
+import org.apache.flink.api.java.DataSet;
+import org.apache.flink.api.java.ExecutionEnvironment;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.api.java.tuple.Tuple3;
+import org.apache.flink.graph.Edge;
+import org.apache.flink.graph.EdgeDirection;
+import org.apache.flink.graph.Graph;
+import org.apache.flink.graph.Vertex;
+import org.apache.flink.graph.example.utils.IncrementalSSSPData;
+import org.apache.flink.graph.spargel.IterationConfiguration;
+import org.apache.flink.graph.spargel.MessageIterator;
+import org.apache.flink.graph.spargel.MessagingFunction;
+import org.apache.flink.graph.spargel.VertexUpdateFunction;
+
+/**
+ * Incremental Single Sink Shortest Paths Example.
+ *
+ * The program takes as input the resulted graph after a SSSP computation,
+ * an edge to be removed and the initial graph(i.e. before SSSP was computed).
+ *
+ * - If the removed edge does not belong to the SP-graph, no computation is necessary.
+ * The edge is simply removed from the graph.
+ * - If the removed edge is an SP-edge, then all nodes, whose shortest path contains the removed edge,
+ * potentially require re-computation.
+ * When the edge <u, v> is removed, v checks if it has another out-going SP-edge.
+ * If yes, no further computation is required.
+ * If v has no other out-going SP-edge, it invalidates its current value, by setting it to INF.
+ * Then, it informs all its SP-in-neighbors by sending them an INVALIDATE message.
+ * When a vertex u receives an INVALIDATE message from v, it checks whether it has another out-going SP-edge.
+ * If not, it invalidates its current value and propagates the INVALIDATE message.
+ * The propagation stops when a vertex with an alternative shortest path is reached
+ * or when we reach a vertex with no SP-in-neighbors.
+ *
+ * Usage <code>IncrementalSSSPExample <vertex path> <edge path> <edges in SSSP>
+ * <edge to be removed> <result path> <number of iterations></code><br>
+ * If no parameters are provided, the program is run with default data from
+ * {@link org.apache.flink.graph.example.utils.IncrementalSSSPData}
+ */
+@SuppressWarnings("serial")
+public class IncrementalSSSPExample implements ProgramDescription {
+
+ public static void main(String [] args) throws Exception {
+
+ if(!parseParameters(args)) {
+ return;
+ }
+
+ ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+
+ DataSet<Vertex<Long, Double>> vertices = getVerticesDataSet(env);
+
+ DataSet<Edge<Long, Double>> edges = getEdgesDataSet(env);
+
+ DataSet<Edge<Long, Double>> edgesInSSSP = getEdgesinSSSPDataSet(env);
+
+ Edge<Long, Double> edgeToBeRemoved = getEdgeToBeRemoved();
+
+ Graph<Long, Double, Double> graph = Graph.fromDataSet(vertices, edges, env);
+
+ // Assumption: all minimum weight paths are kept
+ Graph<Long, Double, Double> ssspGraph = Graph.fromDataSet(vertices, edgesInSSSP, env);
+
+ // remove the edge
+ graph.removeEdge(edgeToBeRemoved);
+
+ // configure the iteration
+ IterationConfiguration parameters = new IterationConfiguration();
+
+ if(isInSSSP(edgeToBeRemoved, edgesInSSSP)) {
+
+ parameters.setDirection(EdgeDirection.IN);
+ parameters.setOptDegrees(true);
+
+ // run the vertex centric iteration to propagate info
+ Graph<Long, Double, Double> result = ssspGraph.runVertexCentricIteration(new VertexDistanceUpdater(),
+ new InvalidateMessenger(edgeToBeRemoved), maxIterations, parameters);
+
+ DataSet<Vertex<Long, Double>> resultedVertices = result.getVertices();
+
+ // Emit results
+ if(fileOutput) {
+ resultedVertices.writeAsCsv(outputPath, "\n", ",");
+ } else {
+ resultedVertices.print();
+ }
+
+ env.execute("Incremental SSSP Example");
+ } else {
+ // print the vertices
+ if(fileOutput) {
+ vertices.writeAsCsv(outputPath, "\n", ",");
+ } else {
+ vertices.print();
+ }
+
+ env.execute("Incremental SSSP Example");
+ }
+ }
+
+ @Override
+ public String getDescription() {
+ return "Incremental Single Sink Shortest Paths Example";
+ }
+
+ // ******************************************************************************************************************
+ // IncrementalSSSP METHODS
+ // ******************************************************************************************************************
+
+ /**
+ * Function that verifies whether the edge to be removed is part of the SSSP or not.
+ * If it is, the src vertex will be invalidated.
+ *
+ * @param edgeToBeRemoved
+ * @param edgesInSSSP
+ * @return
+ */
+ private static boolean isInSSSP(final Edge<Long, Double> edgeToBeRemoved, DataSet<Edge<Long, Double>> edgesInSSSP) throws Exception {
+
+ return edgesInSSSP.filter(new FilterFunction<Edge<Long, Double>>() {
+ @Override
+ public boolean filter(Edge<Long, Double> edge) throws Exception {
+ return edge.equals(edgeToBeRemoved);
+ }
+ }).count() > 0;
+ }
+
+ public static final class VertexDistanceUpdater extends VertexUpdateFunction<Long, Double, Double> {
+
+ @Override
+ public void updateVertex(Vertex<Long, Double> vertex, MessageIterator<Double> inMessages) throws Exception {
+ if (inMessages.hasNext()) {
+ Long outDegree = vertex.getOutDegree() - 1;
+ // check if the vertex has another SP-Edge
+ if (outDegree > 0) {
+ // there is another shortest path from the source to this vertex
+ } else {
+ // set own value to infinity
+ setNewVertexValue(Double.MAX_VALUE);
+ }
+ }
+ }
+ }
+
+ public static final class InvalidateMessenger extends MessagingFunction<Long, Double, Double, Double> {
+
+ private Edge<Long, Double> edgeToBeRemoved;
+
+ public InvalidateMessenger(Edge<Long, Double> edgeToBeRemoved) {
+ this.edgeToBeRemoved = edgeToBeRemoved;
+ }
+
+ @Override
+ public void sendMessages(Vertex<Long, Double> vertex) throws Exception {
+
+
+ if(getSuperstepNumber() == 1) {
+ if(vertex.getId().equals(edgeToBeRemoved.getSource())) {
+ // activate the edge target
+ sendMessageTo(edgeToBeRemoved.getSource(), Double.MAX_VALUE);
+ }
+ }
+
+ if(getSuperstepNumber() > 1) {
+ // invalidate all edges
+ for(Edge<Long, Double> edge : getEdges()) {
+ sendMessageTo(edge.getSource(), Double.MAX_VALUE);
+ }
+ }
+ }
+ }
+
+ // ******************************************************************************************************************
+ // UTIL METHODS
+ // ******************************************************************************************************************
+
+ private static boolean fileOutput = false;
+
+ private static String verticesInputPath = null;
+
+ private static String edgesInputPath = null;
+
+ private static String edgesInSSSPInputPath = null;
+
+ private static String edgeToBeRemoved = null;
+
+ private static String outputPath = null;
+
+ private static int maxIterations = 5;
+
+ private static boolean parseParameters(String[] args) {
+ if (args.length > 0) {
+ if (args.length == 6) {
+ fileOutput = true;
+ verticesInputPath = args[0];
+ edgesInputPath = args[1];
+ edgesInSSSPInputPath = args[2];
+ edgeToBeRemoved = args[3];
+ outputPath = args[4];
+ maxIterations = Integer.parseInt(args[5]);
+ } else {
+ System.out.println("Executing IncrementalSSSP example with default parameters and built-in default data.");
+ System.out.println("Provide parameters to read input data from files.");
+ System.out.println("See the documentation for the correct format of input files.");
+ System.out.println("Usage: IncrementalSSSP <vertex path> <edge path> <edges in SSSP> <edge to be removed> <output path> <max iterations>");
+
+ return false;
+ }
+ }
+ return true;
+ }
+
+ private static DataSet<Vertex<Long, Double>> getVerticesDataSet(ExecutionEnvironment env) {
+ if (fileOutput) {
+ return env.readCsvFile(verticesInputPath)
+ .lineDelimiter("\n")
+ .types(Long.class, Double.class)
+ .map(new MapFunction<Tuple2<Long, Double>, Vertex<Long, Double>>() {
+
+ @Override
+ public Vertex<Long, Double> map(Tuple2<Long, Double> tuple2) throws Exception {
+ return new Vertex<Long, Double>(tuple2.f0, tuple2.f1);
+ }
+ });
+ } else {
+ System.err.println("Usage: IncrementalSSSP <vertex path> <edge path> <edges in SSSP> <edge to be removed> " +
+ "<output path> <max iterations>");
+ return IncrementalSSSPData.getDefaultVertexDataSet(env);
+ }
+ }
+
+ private static DataSet<Edge<Long, Double>> getEdgesDataSet(ExecutionEnvironment env) {
+ if (fileOutput) {
+ return env.readCsvFile(edgesInputPath)
+ .lineDelimiter("\n")
+ .types(Long.class, Long.class, Double.class)
+ .map(new MapFunction<Tuple3<Long, Long, Double>, Edge<Long, Double>>() {
+
+ @Override
+ public Edge<Long, Double> map(Tuple3<Long, Long, Double> tuple3) throws Exception {
+ return new Edge(tuple3.f0, tuple3.f1, tuple3.f2);
+ }
+ });
+ } else {
+ System.err.println("Usage: IncrementalSSSP <vertex path> <edge path> <edges in SSSP> <edge to be removed> " +
+ "<output path> <max iterations>");
+ return IncrementalSSSPData.getDefaultEdgeDataSet(env);
+ }
+ }
+
+ private static DataSet<Edge<Long, Double>> getEdgesinSSSPDataSet(ExecutionEnvironment env) {
+ if (fileOutput) {
+ return env.readCsvFile(edgesInSSSPInputPath)
+ .lineDelimiter("\n")
+ .types(Long.class, Long.class, Double.class)
+ .map(new MapFunction<Tuple3<Long, Long, Double>, Edge<Long, Double>>() {
--- End diff --
same
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-1523][gelly] Vertex centric iteration e...
Posted by vasia <gi...@git.apache.org>.
Github user vasia commented on a diff in the pull request:
https://github.com/apache/flink/pull/537#discussion_r28643437
--- Diff: flink-staging/flink-gelly/src/main/java/org/apache/flink/graph/example/IncrementalSSSPExample.java ---
@@ -0,0 +1,304 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.graph.example;
+
+import org.apache.flink.api.common.ProgramDescription;
+import org.apache.flink.api.common.functions.FilterFunction;
+import org.apache.flink.api.common.functions.MapFunction;
+import org.apache.flink.api.java.DataSet;
+import org.apache.flink.api.java.ExecutionEnvironment;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.api.java.tuple.Tuple3;
+import org.apache.flink.graph.Edge;
+import org.apache.flink.graph.EdgeDirection;
+import org.apache.flink.graph.Graph;
+import org.apache.flink.graph.Vertex;
+import org.apache.flink.graph.example.utils.IncrementalSSSPData;
+import org.apache.flink.graph.spargel.IterationConfiguration;
+import org.apache.flink.graph.spargel.MessageIterator;
+import org.apache.flink.graph.spargel.MessagingFunction;
+import org.apache.flink.graph.spargel.VertexUpdateFunction;
+
+/**
+ * Incremental Single Sink Shortest Paths Example.
--- End diff --
hmm this might be a bit misleading, as this program only incrementally updates the shortest paths when removing an edge from the initial graph.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-1523][gelly] Vertex centric iteration e...
Posted by vasia <gi...@git.apache.org>.
Github user vasia commented on the pull request:
https://github.com/apache/flink/pull/537#issuecomment-99070508
Hey @andralungu ,
I'm not sure what @StephanEwen had in mind when suggesting to subclass Vertex, but the current implementation forces the user to cast the `Vertex` to `VertexWithDegrees` in order to access the degrees, which is not a very user-friendly :S
I guess the point was to have `VertexWithDegrees` as an argument to the update and messaging functions instead? If that's the case however, it's not quite clear to me how this will play along with the configuration option setting.
Back to my previous comment on only accessing the degrees inside the update/messaging functions instead of adding public methods to the Vertex class, a simple way to do this would be the following:
Add set/get vertexDegree methods in `VertexUpdateFunction` and `MessagingFunction`. These would be the methods that the user will call to access the vertex degrees, in a way similar to `getDirection()`, `getBroadcastSet()` etc.
Inside `createResultVerticesWithDegrees`, you're already creating the "augmented" vertex dataset with the degrees and pass this as the solution set of the delta iteration. Then, in the corresponding coGroup method of the Messaging/VertexUpdate UDFs, you iterate over this dataset (state) and call the user-defined functions for each vertex. This is where you can set the degrees inside the update/messaging functions, instead of how you currently set them on the vertex object.
What do you think of this solution?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-1523][gelly] Vertex centric iteration e...
Posted by andralungu <gi...@git.apache.org>.
Github user andralungu commented on the pull request:
https://github.com/apache/flink/pull/537#issuecomment-94915061
Hi @vasia ,
I added some answers to your inline comments! I will push my latest version for this tomorrow.
Regarding the suggestion that the Vertex class might not be the best place for the getDegree methods, there is a reason why implementing this took a while ^^. I wanted to make the degrees available only in the iteration. The problem is that with the current code(this one and the one in "production"), a vertex is only accessible in the updateVertex() method. This means that there is no way to get the vertex within the iteration, you don't have a Vertex object, it's not something you send to the class. That was the quick workaround I found for this issue. If you know a better way, I am eager to hear your suggestions.
Thanks! :)
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-1523][gelly] Vertex centric iteration e...
Posted by vasia <gi...@git.apache.org>.
Github user vasia commented on a diff in the pull request:
https://github.com/apache/flink/pull/537#discussion_r28643463
--- Diff: flink-staging/flink-gelly/src/main/java/org/apache/flink/graph/example/IncrementalSSSPExample.java ---
@@ -0,0 +1,304 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.graph.example;
+
+import org.apache.flink.api.common.ProgramDescription;
+import org.apache.flink.api.common.functions.FilterFunction;
+import org.apache.flink.api.common.functions.MapFunction;
+import org.apache.flink.api.java.DataSet;
+import org.apache.flink.api.java.ExecutionEnvironment;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.api.java.tuple.Tuple3;
+import org.apache.flink.graph.Edge;
+import org.apache.flink.graph.EdgeDirection;
+import org.apache.flink.graph.Graph;
+import org.apache.flink.graph.Vertex;
+import org.apache.flink.graph.example.utils.IncrementalSSSPData;
+import org.apache.flink.graph.spargel.IterationConfiguration;
+import org.apache.flink.graph.spargel.MessageIterator;
+import org.apache.flink.graph.spargel.MessagingFunction;
+import org.apache.flink.graph.spargel.VertexUpdateFunction;
+
+/**
+ * Incremental Single Sink Shortest Paths Example.
+ *
+ * The program takes as input the resulted graph after a SSSP computation,
+ * an edge to be removed and the initial graph(i.e. before SSSP was computed).
+ *
+ * - If the removed edge does not belong to the SP-graph, no computation is necessary.
+ * The edge is simply removed from the graph.
+ * - If the removed edge is an SP-edge, then all nodes, whose shortest path contains the removed edge,
+ * potentially require re-computation.
+ * When the edge <u, v> is removed, v checks if it has another out-going SP-edge.
+ * If yes, no further computation is required.
+ * If v has no other out-going SP-edge, it invalidates its current value, by setting it to INF.
+ * Then, it informs all its SP-in-neighbors by sending them an INVALIDATE message.
+ * When a vertex u receives an INVALIDATE message from v, it checks whether it has another out-going SP-edge.
+ * If not, it invalidates its current value and propagates the INVALIDATE message.
+ * The propagation stops when a vertex with an alternative shortest path is reached
+ * or when we reach a vertex with no SP-in-neighbors.
+ *
+ * Usage <code>IncrementalSSSPExample <vertex path> <edge path> <edges in SSSP>
+ * <edge to be removed> <result path> <number of iterations></code><br>
+ * If no parameters are provided, the program is run with default data from
+ * {@link org.apache.flink.graph.example.utils.IncrementalSSSPData}
+ */
+@SuppressWarnings("serial")
+public class IncrementalSSSPExample implements ProgramDescription {
+
+ public static void main(String [] args) throws Exception {
+
+ if(!parseParameters(args)) {
+ return;
+ }
+
+ ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+
+ DataSet<Vertex<Long, Double>> vertices = getVerticesDataSet(env);
+
+ DataSet<Edge<Long, Double>> edges = getEdgesDataSet(env);
+
+ DataSet<Edge<Long, Double>> edgesInSSSP = getEdgesinSSSPDataSet(env);
+
+ Edge<Long, Double> edgeToBeRemoved = getEdgeToBeRemoved();
+
+ Graph<Long, Double, Double> graph = Graph.fromDataSet(vertices, edges, env);
+
+ // Assumption: all minimum weight paths are kept
+ Graph<Long, Double, Double> ssspGraph = Graph.fromDataSet(vertices, edgesInSSSP, env);
+
+ // remove the edge
+ graph.removeEdge(edgeToBeRemoved);
+
+ // configure the iteration
+ IterationConfiguration parameters = new IterationConfiguration();
+
+ if(isInSSSP(edgeToBeRemoved, edgesInSSSP)) {
+
+ parameters.setDirection(EdgeDirection.IN);
+ parameters.setOptDegrees(true);
+
+ // run the vertex centric iteration to propagate info
+ Graph<Long, Double, Double> result = ssspGraph.runVertexCentricIteration(new VertexDistanceUpdater(),
+ new InvalidateMessenger(edgeToBeRemoved), maxIterations, parameters);
+
+ DataSet<Vertex<Long, Double>> resultedVertices = result.getVertices();
+
+ // Emit results
+ if(fileOutput) {
+ resultedVertices.writeAsCsv(outputPath, "\n", ",");
+ } else {
+ resultedVertices.print();
+ }
+
+ env.execute("Incremental SSSP Example");
+ } else {
+ // print the vertices
+ if(fileOutput) {
+ vertices.writeAsCsv(outputPath, "\n", ",");
+ } else {
+ vertices.print();
+ }
+
+ env.execute("Incremental SSSP Example");
+ }
+ }
+
+ @Override
+ public String getDescription() {
+ return "Incremental Single Sink Shortest Paths Example";
+ }
+
+ // ******************************************************************************************************************
+ // IncrementalSSSP METHODS
+ // ******************************************************************************************************************
+
+ /**
+ * Function that verifies whether the edge to be removed is part of the SSSP or not.
+ * If it is, the src vertex will be invalidated.
+ *
+ * @param edgeToBeRemoved
+ * @param edgesInSSSP
+ * @return
+ */
+ private static boolean isInSSSP(final Edge<Long, Double> edgeToBeRemoved, DataSet<Edge<Long, Double>> edgesInSSSP) throws Exception {
+
+ return edgesInSSSP.filter(new FilterFunction<Edge<Long, Double>>() {
+ @Override
+ public boolean filter(Edge<Long, Double> edge) throws Exception {
+ return edge.equals(edgeToBeRemoved);
+ }
+ }).count() > 0;
+ }
+
+ public static final class VertexDistanceUpdater extends VertexUpdateFunction<Long, Double, Double> {
+
+ @Override
+ public void updateVertex(Vertex<Long, Double> vertex, MessageIterator<Double> inMessages) throws Exception {
+ if (inMessages.hasNext()) {
+ Long outDegree = vertex.getOutDegree() - 1;
+ // check if the vertex has another SP-Edge
+ if (outDegree > 0) {
+ // there is another shortest path from the source to this vertex
+ } else {
+ // set own value to infinity
+ setNewVertexValue(Double.MAX_VALUE);
+ }
+ }
+ }
+ }
+
+ public static final class InvalidateMessenger extends MessagingFunction<Long, Double, Double, Double> {
+
+ private Edge<Long, Double> edgeToBeRemoved;
+
+ public InvalidateMessenger(Edge<Long, Double> edgeToBeRemoved) {
+ this.edgeToBeRemoved = edgeToBeRemoved;
+ }
+
+ @Override
+ public void sendMessages(Vertex<Long, Double> vertex) throws Exception {
+
+
+ if(getSuperstepNumber() == 1) {
+ if(vertex.getId().equals(edgeToBeRemoved.getSource())) {
+ // activate the edge target
+ sendMessageTo(edgeToBeRemoved.getSource(), Double.MAX_VALUE);
+ }
+ }
+
+ if(getSuperstepNumber() > 1) {
+ // invalidate all edges
+ for(Edge<Long, Double> edge : getEdges()) {
+ sendMessageTo(edge.getSource(), Double.MAX_VALUE);
+ }
+ }
+ }
+ }
+
+ // ******************************************************************************************************************
+ // UTIL METHODS
+ // ******************************************************************************************************************
+
+ private static boolean fileOutput = false;
+
+ private static String verticesInputPath = null;
+
+ private static String edgesInputPath = null;
+
+ private static String edgesInSSSPInputPath = null;
+
+ private static String edgeToBeRemoved = null;
+
+ private static String outputPath = null;
+
+ private static int maxIterations = 5;
+
+ private static boolean parseParameters(String[] args) {
+ if (args.length > 0) {
+ if (args.length == 6) {
+ fileOutput = true;
+ verticesInputPath = args[0];
+ edgesInputPath = args[1];
+ edgesInSSSPInputPath = args[2];
+ edgeToBeRemoved = args[3];
+ outputPath = args[4];
+ maxIterations = Integer.parseInt(args[5]);
+ } else {
+ System.out.println("Executing IncrementalSSSP example with default parameters and built-in default data.");
+ System.out.println("Provide parameters to read input data from files.");
+ System.out.println("See the documentation for the correct format of input files.");
+ System.out.println("Usage: IncrementalSSSP <vertex path> <edge path> <edges in SSSP> <edge to be removed> <output path> <max iterations>");
+
+ return false;
+ }
+ }
+ return true;
+ }
+
+ private static DataSet<Vertex<Long, Double>> getVerticesDataSet(ExecutionEnvironment env) {
+ if (fileOutput) {
+ return env.readCsvFile(verticesInputPath)
+ .lineDelimiter("\n")
+ .types(Long.class, Double.class)
+ .map(new MapFunction<Tuple2<Long, Double>, Vertex<Long, Double>>() {
+
+ @Override
+ public Vertex<Long, Double> map(Tuple2<Long, Double> tuple2) throws Exception {
+ return new Vertex<Long, Double>(tuple2.f0, tuple2.f1);
+ }
+ });
+ } else {
+ System.err.println("Usage: IncrementalSSSP <vertex path> <edge path> <edges in SSSP> <edge to be removed> " +
+ "<output path> <max iterations>");
+ return IncrementalSSSPData.getDefaultVertexDataSet(env);
+ }
+ }
+
+ private static DataSet<Edge<Long, Double>> getEdgesDataSet(ExecutionEnvironment env) {
+ if (fileOutput) {
+ return env.readCsvFile(edgesInputPath)
+ .lineDelimiter("\n")
+ .types(Long.class, Long.class, Double.class)
+ .map(new MapFunction<Tuple3<Long, Long, Double>, Edge<Long, Double>>() {
+
+ @Override
+ public Edge<Long, Double> map(Tuple3<Long, Long, Double> tuple3) throws Exception {
--- End diff --
and a Tuple3ToEdgeMap :-)
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-1523][gelly] Vertex centric iteration e...
Posted by vasia <gi...@git.apache.org>.
Github user vasia commented on a diff in the pull request:
https://github.com/apache/flink/pull/537#discussion_r28643344
--- Diff: docs/gelly_guide.md ---
@@ -380,6 +380,16 @@ all aggregates globally once per superstep and makes them available in the next
* <strong>Broadcast Variables</strong>: DataSets can be added as [Broadcast Variables](programming_guide.html#broadcast-variables) to the `VertexUpdateFunction` and `MessagingFunction`, using the `addBroadcastSetForUpdateFunction()` and `addBroadcastSetForMessagingFunction()` methods, respectively.
+* <strong>Number of Vertices</strong>: Accessing the total number of vertices within the iteration. This property can be set using the `setOptNumVertices()` method.
+
+The number of vertices can then be accessed in the vertex update function and in the messaging function using the `getNumberOfVertices()` method.
+
+* <strong>Degrees</strong>: Accessing the in/out degree for a vertex within an iteration. This property can be set using the `setOptDegrees()` method.
+
+The in/out degrees can then be accessed in the vertex update function and in the messaging function, per vertex using `vertex.getInDegree()` or `vertex.getOutDegree()`.
+
+* <strong>Messaging Direction</strong>: The direction in which messages are sent. This can be either EdgeDirection.IN, EdgeDirection.OUT, EdgeDirection.ALL. The messaging direction also dictates the update direction which would be EdgeDirection.OUT, EdgeDirection.IN and EdgeDirection.ALL, respectively. This property can be set using the `setDirection()` method.
+
--- End diff --
First, I would highlight the `EdgeDirection.IN` etc. Then, maybe we can rephrase this to make a bit clearer. I would write something like the following: "By default a vertex sends messages to its out-neighbors and thus, updates its value based on received messages from its in-neighbors. This configuration option allows to change the messaging direction and set it to [...]. The messaging direction also dictates the update direction."
We could also add a couple of figures to illustrate the messaging-update directions. I can create those, so that they're in the same format as the rest of the figures in the guide.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-1523][gelly] Vertex centric iteration e...
Posted by vasia <gi...@git.apache.org>.
Github user vasia commented on a diff in the pull request:
https://github.com/apache/flink/pull/537#discussion_r28643293
--- Diff: docs/gelly_guide.md ---
@@ -380,6 +380,16 @@ all aggregates globally once per superstep and makes them available in the next
* <strong>Broadcast Variables</strong>: DataSets can be added as [Broadcast Variables](programming_guide.html#broadcast-variables) to the `VertexUpdateFunction` and `MessagingFunction`, using the `addBroadcastSetForUpdateFunction()` and `addBroadcastSetForMessagingFunction()` methods, respectively.
+* <strong>Number of Vertices</strong>: Accessing the total number of vertices within the iteration. This property can be set using the `setOptNumVertices()` method.
+
+The number of vertices can then be accessed in the vertex update function and in the messaging function using the `getNumberOfVertices()` method.
+
+* <strong>Degrees</strong>: Accessing the in/out degree for a vertex within an iteration. This property can be set using the `setOptDegrees()` method.
+
--- End diff --
same
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-1523][gelly] Vertex centric iteration e...
Posted by vasia <gi...@git.apache.org>.
Github user vasia commented on a diff in the pull request:
https://github.com/apache/flink/pull/537#discussion_r28643520
--- Diff: flink-staging/flink-gelly/src/main/java/org/apache/flink/graph/spargel/IterationConfiguration.java ---
@@ -189,4 +199,33 @@ public void addBroadcastSetForUpdateFunction(String name, DataSet<?> data) {
public List<Tuple2<String, DataSet<?>>> getMessagingBcastVars() {
return this.bcVarsMessaging;
}
+
+ // ----------------------------------------------------------------------------------
+ // The direction, degrees and the total number of vertices should be optional.
+ // The user can access them by setting the direction, degrees or the numVertices options.
+ // ----------------------------------------------------------------------------------
--- End diff --
all parameters in this class are optional :-)
Can you add javadocs to the new methods, like in the methods above?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-1523][gelly] Vertex centric iteration e...
Posted by andralungu <gi...@git.apache.org>.
Github user andralungu closed the pull request at:
https://github.com/apache/flink/pull/537
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-1523][gelly] Vertex centric iteration e...
Posted by vasia <gi...@git.apache.org>.
Github user vasia commented on the pull request:
https://github.com/apache/flink/pull/537#issuecomment-90233701
Hey @andralungu! Thanks a lot for your work on this :)
I only took a quick look at your changes for now, but it seems that all issues are fixed.
This PR introduces a lot of changes, so I would like to look into it more carefully. Also, it'd be great if one more person could review.
Next, I think we should add a few tests to make sure we don't break existing functionality and that the newly introduced options work as expected.
More specifically, I would test the following:
- if no direction is given, the iteration works as before, i.e. collecting messages from in-neighbors and sending messages to out-neighbors .
- if direction is set to IN / OUT / ALL
- if the degrees option is not set, the vertex cannot access the degree, i.e. no method is available or an error is produced
- if the degrees option is set, the degrees can be accessed in all supersteps and their values are correct
- combination of the above, e.g. set direction to ALL and degrees.
I guess adding those tests will be easier after we merge #547 and make the new options part of the configuration.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-1523][gelly] Vertex centric iteration e...
Posted by vasia <gi...@git.apache.org>.
Github user vasia commented on the pull request:
https://github.com/apache/flink/pull/537#issuecomment-101048928
Hey @andralungu ,
I built on your changes and implemented what I described above. I have pushed in [this branch](https://github.com/vasia/flink/tree/vc-extentions).
I had to manually resolve conflicts caused by #657, so I guess the easiest way to allow people to review would be for me to open a fresh PR.
Since you are the one mostly familiar with these features, could you please take a look and let me know what you think?
Thanks a lot!
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-1523][gelly] Vertex centric iteration e...
Posted by andralungu <gi...@git.apache.org>.
Github user andralungu commented on the pull request:
https://github.com/apache/flink/pull/537#issuecomment-90061271
@vasia ,
PR updated :)
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-1523][gelly] Vertex centric iteration e...
Posted by vasia <gi...@git.apache.org>.
Github user vasia commented on a diff in the pull request:
https://github.com/apache/flink/pull/537#discussion_r28643494
--- Diff: flink-staging/flink-gelly/src/main/java/org/apache/flink/graph/library/SingleSourceShortestPaths.java ---
@@ -77,7 +77,7 @@ public Double map(Vertex<K, Double> value) {
extends VertexUpdateFunction<K, Double, Double> {
--- End diff --
I really like what the examples look like now! Much better having the Vertex object instead of separated key-value ^^
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-1523][gelly] Vertex centric iteration e...
Posted by vasia <gi...@git.apache.org>.
Github user vasia commented on a diff in the pull request:
https://github.com/apache/flink/pull/537#discussion_r28643695
--- Diff: flink-staging/flink-gelly/src/main/java/org/apache/flink/graph/spargel/VertexCentricIteration.java ---
@@ -86,7 +92,9 @@
private DataSet<Vertex<VertexKey, VertexValue>> initialVertices;
private IterationConfiguration configuration;
-
+
+ private DataSet<Vertex<VertexKey, Tuple3<VertexValue, Long, Long>>> verticesWithDegrees;
--- End diff --
it seems that this isn't used anywhere?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---