[FLINK-1694] [gelly] added IterationConfiguration as a way to configu…

…re a VertexCentricIteration

docs/gelly_guide.md

@@ -347,29 +347,86 @@ Vertex-centric Iterations
 Gelly wraps Flink's [Spargel API](spargel_guide.html) to provide methods for vertex-centric iterations.
 Like in Spargel, the user only needs to implement two functions: a `VertexUpdateFunction`, which defines how a vertex will update its value
 based on the received messages and a `MessagingFunction`, which allows a vertex to send out messages for the next superstep.
-These functions are given as parameters to Gelly's `createVertexCentricIteration`, which returns a `VertexCentricIteration`. 
-The user can configure this iteration (set the name, the parallelism, aggregators, etc.) and then run the computation, using the `runVertexCentricIteration` method:
+These functions and the maximum number of iterations to run are given as parameters to Gelly's `runVertexCentricIteration`.
+This method will execute the vertex-centric iteration on the input Graph and return a new Graph, with updated vertex values:
 
 {% highlight java %}
 Graph<Long, Double, Double> graph = ...
 
-// create the vertex-centric iteration
-VertexCentricIteration<Long, Double, Double, Double> iteration = 
-			graph.createVertexCentricIteration(
+// run Single-Source-Shortest-Paths vertex-centric iteration
+Graph<Long, Double, Double> result = 
+			graph.runVertexCentricIteration(
 			new VertexDistanceUpdater(), new MinDistanceMessenger(), maxIterations);
 
+// user-defined functions
+public static final class VertexDistanceUpdater {...}
+public static final class MinDistanceMessenger {...}
+
+{% endhighlight %}
+
+### Configuring a Vertex-Centric Iteration
+A vertex-centric iteration can be configured using an `IterationConfiguration` object.
+Currently, the following parameters can be specified:
+
+* <strong>Name</strong>: The name for the vertex-centric iteration. The name is displayed in logs and messages 
+and can be specified using the `setName()` method.
+
+* <strong>Parallelism</strong>: The parallelism for the iteration. It can be set using the `setParallelism()` method.	
+
+* <strong>Solution set in unmanaged memory</strong>: Defines whether the solution set is kept in managed memory (Flink's internal way of keeping object in serialized form) or as a simple object map. By default, the solution set runs in managed memory. This property can be set using the `setSolutionSetUnmanagedMemory()` method.
+
+* <strong>Aggregators</strong>: Iteration aggregators can be registered using the `registerAggregator()` method. An iteration aggregator combines
+all aggregates globally once per superstep and makes them available in the next superstep. Registered aggregators can be accessed inside the user-defined `VertexUpdateFunction` and `MessagingFunction`.
+
+* <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.
+
+{% highlight java %}
+
+Graph<Long, Double, Double> graph = ...
+
+// configure the iteration
+IterationConfiguration parameters = new IterationConfiguration();
+
 // set the iteration name
-iteration.setName("Single Source Shortest Paths");
+parameters.setName("Gelly Iteration");
 
 // set the parallelism
-iteration.setParallelism(16);
+parameters.setParallelism(16);
+
+// register an aggregator
+parameters.registerAggregator("sumAggregator", new LongSumAggregator());
 
-// run the computation
-graph.runVertexCentricIteration(iteration);
+// run the vertex-centric iteration, also passing the configuration parameters
+Graph<Long, Double, Double> result = 
+			graph.runVertexCentricIteration(
+			new VertexUpdater(), new Messenger(), maxIterations, parameters);
 
 // user-defined functions
-public static final class VertexDistanceUpdater {...}
-public static final class MinDistanceMessenger {...}
+public static final class VertexUpdater extends VertexUpdateFunction {
+
+	LongSumAggregator aggregator = new LongSumAggregator();
+
+	public void preSuperstep() {
+
+		// retrieve the Aggregator
+		aggregator = getIterationAggregator("sumAggregator");
+	}
+
+
+	public void updateVertex(Long vertexKey, Long vertexValue, MessageIterator inMessages) {
+		
+		//do some computation
+		Long partialValue = ...
+
+		// aggregate the partial value
+		aggregator.aggregate(partialValue);
+
+		// update the vertex value
+		setNewVertexValue(...);
+	}
+}
+
+public static final class Messenger extends MessagingFunction {...}
 
 {% endhighlight %}
 

flink-staging/flink-gelly/src/main/java/org/apache/flink/graph/Graph.java

@@ -47,6 +47,7 @@
 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.graph.spargel.IterationConfiguration;
 import org.apache.flink.graph.spargel.MessagingFunction;
 import org.apache.flink.graph.spargel.VertexCentricIteration;
 import org.apache.flink.graph.spargel.VertexUpdateFunction;
@@ -1149,30 +1150,52 @@ public Graph<K, VV, EV> union(Graph<K, VV, EV> graph) {
 	}
 
 	/**
-	 * Create a Vertex-Centric iteration on the graph.
-	 * 
+	 * Runs a Vertex-Centric iteration on the graph.
+	 * No configuration options are provided.
+
 	 * @param vertexUpdateFunction the vertex update function
 	 * @param messagingFunction the messaging function
 	 * @param maximumNumberOfIterations maximum number of iterations to perform
-	 * @return
+	 * 
+	 * @return the updated Graph after the vertex-centric iteration has converged or
+	 * after maximumNumberOfIterations.
 	 */
-	public <M> VertexCentricIteration<K, VV, M, EV> createVertexCentricIteration(
+	public <M> Graph<K, VV, EV> runVertexCentricIteration(
 			VertexUpdateFunction<K, VV, M> vertexUpdateFunction,
 			MessagingFunction<K, VV, M, EV> messagingFunction,
 			int maximumNumberOfIterations) {
-		return VertexCentricIteration.withEdges(edges, vertexUpdateFunction,
-				messagingFunction, maximumNumberOfIterations);
+
+		VertexCentricIteration<K, VV, M, EV> iteration = VertexCentricIteration.withEdges(
+				edges, vertexUpdateFunction, messagingFunction, maximumNumberOfIterations);
+
+		DataSet<Vertex<K, VV>> newVertices = vertices.runOperation(iteration);
+
+		return new Graph<K, VV, EV>(newVertices, this.edges, this.context);
 	}
-    
+
 	/**
-	 * Runs a Vertex-Centric iteration on the graph.
+	 * Runs a Vertex-Centric iteration on the graph with configuration options.
+	 * 
+	 * @param vertexUpdateFunction the vertex update function
+	 * @param messagingFunction the messaging function
+	 * @param maximumNumberOfIterations maximum number of iterations to perform
+	 * @param parameters the iteration configuration parameters
 	 * 
-	 * @param iteration the Vertex-Centric iteration to run
-	 * @return
+	 * @return the updated Graph after the vertex-centric iteration has converged or
+	 * after maximumNumberOfIterations.
 	 */
 	public <M> Graph<K, VV, EV> runVertexCentricIteration(
-			VertexCentricIteration<K, VV, M, EV> iteration) {
+			VertexUpdateFunction<K, VV, M> vertexUpdateFunction,
+			MessagingFunction<K, VV, M, EV> messagingFunction,
+			int maximumNumberOfIterations, IterationConfiguration parameters) {
+
+		VertexCentricIteration<K, VV, M, EV> iteration = VertexCentricIteration.withEdges(
+				edges, vertexUpdateFunction, messagingFunction, maximumNumberOfIterations);
+
+		iteration.configure(parameters);
+
 		DataSet<Vertex<K, VV>> newVertices = vertices.runOperation(iteration);
+
 		return new Graph<K, VV, EV>(newVertices, this.edges, this.context);
 	}
 

flink-staging/flink-gelly/src/main/java/org/apache/flink/graph/library/LabelPropagation.java

@@ -22,7 +22,6 @@
 import org.apache.flink.graph.GraphAlgorithm;
 import org.apache.flink.graph.spargel.MessageIterator;
 import org.apache.flink.graph.spargel.MessagingFunction;
-import org.apache.flink.graph.spargel.VertexCentricIteration;
 import org.apache.flink.graph.spargel.VertexUpdateFunction;
 import org.apache.flink.types.NullValue;
 
@@ -56,9 +55,8 @@ public Graph<K, Long, NullValue> run(Graph<K, Long, NullValue> input) {
 
 		// iteratively adopt the most frequent label among the neighbors
 		// of each vertex
-		VertexCentricIteration<K, Long, Long, NullValue> iteration = input.createVertexCentricIteration(
-				new UpdateVertexLabel<K>(), new SendNewLabelToNeighbors<K>(), maxIterations);
-		return input.runVertexCentricIteration(iteration);
+		return input.runVertexCentricIteration(new UpdateVertexLabel<K>(), new SendNewLabelToNeighbors<K>(),
+				maxIterations);
 	}
 
 	/**

flink-staging/flink-gelly/src/main/java/org/apache/flink/graph/library/PageRank.java

@@ -25,7 +25,6 @@
 import org.apache.flink.graph.GraphAlgorithm;
 import org.apache.flink.graph.spargel.MessageIterator;
 import org.apache.flink.graph.spargel.MessagingFunction;
-import org.apache.flink.graph.spargel.VertexCentricIteration;
 import org.apache.flink.graph.spargel.VertexUpdateFunction;
 
 public class PageRank<K extends Comparable<K> & Serializable> implements
@@ -43,11 +42,8 @@ public PageRank(double beta, int maxIterations) {
 	public Graph<K, Double, Double> run(Graph<K, Double, Double> network) throws Exception {
 
 		final long numberOfVertices = network.numberOfVertices();
-
-		VertexCentricIteration<K, Double, Double, Double> iteration = network.createVertexCentricIteration(
-				new VertexRankUpdater<K>(beta, numberOfVertices), new RankMessenger<K>(numberOfVertices),
+		return network.runVertexCentricIteration(new VertexRankUpdater<K>(beta, numberOfVertices), new RankMessenger<K>(numberOfVertices),
 				maxIterations);
-		return network.runVertexCentricIteration(iteration);
 	}
 
 	/**

flink-staging/flink-gelly/src/main/java/org/apache/flink/graph/library/SimpleCommunityDetection.java

@@ -26,7 +26,6 @@
 import org.apache.flink.graph.Vertex;
 import org.apache.flink.graph.spargel.MessageIterator;
 import org.apache.flink.graph.spargel.MessagingFunction;
-import org.apache.flink.graph.spargel.VertexCentricIteration;
 import org.apache.flink.graph.spargel.VertexUpdateFunction;
 
 import java.util.Map;
@@ -65,11 +64,8 @@ public Graph<Long, Long, Double> run(Graph<Long, Long, Double> graph) {
 		Graph<Long, Tuple2<Long, Double>, Double> graphWithScoredVertices = undirectedGraph
 				.mapVertices(new AddScoreToVertexValuesMapper());
 
-		VertexCentricIteration<Long, Tuple2<Long, Double>, Tuple2<Long, Double>, Double>
-				iteration = graphWithScoredVertices.createVertexCentricIteration(new VertexLabelUpdater(delta),
-				new LabelMessenger(), maxIterations);
-
-		return graphWithScoredVertices.runVertexCentricIteration(iteration)
+		return graphWithScoredVertices.runVertexCentricIteration(new VertexLabelUpdater(delta),
+				new LabelMessenger(), maxIterations)
 				.mapVertices(new RemoveScoreFromVertexValuesMapper());
 	}
 

flink-staging/flink-gelly/src/main/java/org/apache/flink/graph/library/SingleSourceShortestPaths.java

@@ -25,7 +25,6 @@
 import org.apache.flink.graph.Vertex;
 import org.apache.flink.graph.spargel.MessageIterator;
 import org.apache.flink.graph.spargel.MessagingFunction;
-import org.apache.flink.graph.spargel.VertexCentricIteration;
 import org.apache.flink.graph.spargel.VertexUpdateFunction;
 
 import java.io.Serializable;
@@ -45,12 +44,9 @@ public SingleSourceShortestPaths(K srcVertexId, Integer maxIterations) {
 	@Override
 	public Graph<K, Double, Double> run(Graph<K, Double, Double> input) {
 
-		Graph<K, Double, Double> mappedInput = input.mapVertices(new InitVerticesMapper<K>(srcVertexId));
-
-		VertexCentricIteration<K, Double, Double, Double> iteration = mappedInput.createVertexCentricIteration(
-				new VertexDistanceUpdater<K>(), new MinDistanceMessenger<K>(), maxIterations);
-
-		return mappedInput.runVertexCentricIteration(iteration);
+		return input.mapVertices(new InitVerticesMapper<K>(srcVertexId))
+				.runVertexCentricIteration(new VertexDistanceUpdater<K>(), new MinDistanceMessenger<K>(),
+				maxIterations);
 	}
 
 	public static final class InitVerticesMapper<K extends Comparable<K> & Serializable>