You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by va...@apache.org on 2016/02/02 11:08:28 UTC
[2/3] flink git commit: [FLINK-3208] [gelly] rename vertex-centric
iteration model to scatter-gather
http://git-wip-us.apache.org/repos/asf/flink/blob/f8677464/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/spargel/VertexCentricIteration.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/spargel/VertexCentricIteration.java b/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/spargel/VertexCentricIteration.java
deleted file mode 100644
index 5c60833..0000000
--- a/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/spargel/VertexCentricIteration.java
+++ /dev/null
@@ -1,686 +0,0 @@
-/*
- * 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.spargel;
-
-import java.util.Iterator;
-import java.util.Map;
-
-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.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;
-
-import com.google.common.base.Preconditions;
-
-/**
- * This class represents iterative graph computations, programmed in a vertex-centric perspective.
- * It is a special case of <i>Bulk Synchronous Parallel</i> computation. The paradigm has also been
- * implemented by Google's <i>Pregel</i> system and by <i>Apache Giraph</i>.
- * <p>
- * Vertex centric algorithms operate on graphs, which are defined through vertices and edges. The
- * algorithms send messages along the edges and update the state of vertices based on
- * the old state and the incoming messages. All vertices have an initial state.
- * The computation terminates once no vertex updates it state any more.
- * Additionally, a maximum number of iterations (supersteps) may be specified.
- * <p>
- * The computation is here represented by two functions:
- * <ul>
- * <li>The {@link VertexUpdateFunction} receives incoming messages and may updates the state for
- * the vertex. If a state is updated, messages are sent from this vertex. Initially, all vertices are
- * considered updated.</li>
- * <li>The {@link MessagingFunction} takes the new vertex state and sends messages along the outgoing
- * edges of the vertex. The outgoing edges may optionally have an associated value, such as a weight.</li>
- * </ul>
- * <p>
- *
- * Vertex-centric graph iterations are are run by calling
- * {@link Graph#runVertexCentricIteration(VertexUpdateFunction, MessagingFunction, int)}.
- *
- * @param <K> The type of the vertex key (the vertex identifier).
- * @param <VV> The type of the vertex value (the state of the vertex).
- * @param <Message> The type of the message sent between vertices along the edges.
- * @param <EV> The type of the values that are associated with the edges.
- */
-public class VertexCentricIteration<K, VV, Message, EV>
- implements CustomUnaryOperation<Vertex<K, VV>, Vertex<K, VV>>
-{
- private final VertexUpdateFunction<K, VV, Message> updateFunction;
-
- private final MessagingFunction<K, VV, Message, EV> messagingFunction;
-
- private final DataSet<Edge<K, EV>> edgesWithValue;
-
- private final int maximumNumberOfIterations;
-
- private final TypeInformation<Message> messageType;
-
- private DataSet<Vertex<K, VV>> initialVertices;
-
- private VertexCentricConfiguration configuration;
-
- // ----------------------------------------------------------------------------------
-
- private VertexCentricIteration(VertexUpdateFunction<K, VV, Message> uf,
- MessagingFunction<K, VV, Message, EV> mf,
- DataSet<Edge<K, EV>> edgesWithValue,
- int maximumNumberOfIterations)
- {
- Preconditions.checkNotNull(uf);
- Preconditions.checkNotNull(mf);
- Preconditions.checkNotNull(edgesWithValue);
- Preconditions.checkArgument(maximumNumberOfIterations > 0, "The maximum number of iterations must be at least one.");
-
- this.updateFunction = uf;
- this.messagingFunction = mf;
- this.edgesWithValue = edgesWithValue;
- this.maximumNumberOfIterations = maximumNumberOfIterations;
- this.messageType = getMessageType(mf);
- }
-
- private TypeInformation<Message> getMessageType(MessagingFunction<K, VV, Message, EV> mf) {
- return TypeExtractor.createTypeInfo(mf, MessagingFunction.class, mf.getClass(), 2);
- }
-
- // --------------------------------------------------------------------------------------------
- // Custom Operator behavior
- // --------------------------------------------------------------------------------------------
-
- /**
- * Sets the input data set for this operator. In the case of this operator this input data set represents
- * the set of vertices with their initial state.
- *
- * @param inputData The input data set, which in the case of this operator represents the set of
- * vertices with their initial state.
- *
- * @see org.apache.flink.api.java.operators.CustomUnaryOperation#setInput(org.apache.flink.api.java.DataSet)
- */
- @Override
- public void setInput(DataSet<Vertex<K, VV>> inputData) {
- this.initialVertices = inputData;
- }
-
- /**
- * Creates the operator that represents this vertex-centric graph computation.
- *
- * @return The operator that represents this vertex-centric graph computation.
- */
- @Override
- public DataSet<Vertex<K, VV>> createResult() {
- if (this.initialVertices == null) {
- throw new IllegalStateException("The input data set has not been set.");
- }
-
- // prepare some type information
- TypeInformation<K> keyType = ((TupleTypeInfo<?>) initialVertices.getType()).getTypeAt(0);
- TypeInformation<Tuple2<K, Message>> messageTypeInfo = new TupleTypeInfo<Tuple2<K,Message>>(keyType, messageType);
-
- // create a graph
- Graph<K, VV, EV> graph =
- Graph.fromDataSet(initialVertices, edgesWithValue, initialVertices.getExecutionEnvironment());
-
- // 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
-
- if (this.configuration != null && this.configuration.isOptNumVertices()) {
- try {
- long numberOfVertices = graph.numberOfVertices();
- messagingFunction.setNumberOfVertices(numberOfVertices);
- updateFunction.setNumberOfVertices(numberOfVertices);
- } catch (Exception e) {
- e.printStackTrace();
- }
- }
-
- if(this.configuration != null) {
- messagingFunction.setDirection(this.configuration.getDirection());
- } else {
- messagingFunction.setDirection(EdgeDirection.OUT);
- }
-
- // retrieve the direction in which the updates are made and in which the messages are sent
- EdgeDirection messagingDirection = messagingFunction.getDirection();
-
- // check whether the degrees option is set and, if so, compute the in and the out degrees and
- // add them to the vertex value
- if(this.configuration != null && this.configuration.isOptDegrees()) {
- return createResultVerticesWithDegrees(graph, messagingDirection, messageTypeInfo);
- } else {
- return createResultSimpleVertex(messagingDirection, messageTypeInfo);
- }
- }
-
- /**
- * Creates a new vertex-centric iteration operator for graphs where the edges are associated with a value (such as
- * a weight or distance).
- *
- * @param edgesWithValue The data set containing edges.
- * @param uf The function that updates the state of the vertices from the incoming messages.
- * @param mf The function that turns changed vertex states into messages along the edges.
- *
- * @param <K> The type of the vertex key (the vertex identifier).
- * @param <VV> The type of the vertex value (the state of the vertex).
- * @param <Message> The type of the message sent between vertices along the edges.
- * @param <EV> The type of the values that are associated with the edges.
- *
- * @return An in stance of the vertex-centric graph computation operator.
- */
- public static final <K, VV, Message, EV>
- VertexCentricIteration<K, VV, Message, EV> withEdges(
- DataSet<Edge<K, EV>> edgesWithValue,
- VertexUpdateFunction<K, VV, Message> uf,
- MessagingFunction<K, VV, Message, EV> mf,
- int maximumNumberOfIterations)
- {
- return new VertexCentricIteration<K, VV, Message, EV>(uf, mf, edgesWithValue, maximumNumberOfIterations);
- }
-
- /**
- * Configures this vertex-centric iteration with the provided parameters.
- *
- * @param parameters the configuration parameters
- */
- public void configure(VertexCentricConfiguration parameters) {
- this.configuration = parameters;
- }
-
- /**
- * @return the configuration parameters of this vertex-centric iteration
- */
- public VertexCentricConfiguration getIterationConfiguration() {
- return this.configuration;
- }
-
- // --------------------------------------------------------------------------------------------
- // Wrapping UDFs
- // --------------------------------------------------------------------------------------------
-
- private static abstract class VertexUpdateUdf<K, VVWithDegrees, Message> extends RichCoGroupFunction<
- Tuple2<K, Message>, Vertex<K, VVWithDegrees>, Vertex<K, VVWithDegrees>>
- implements ResultTypeQueryable<Vertex<K, VVWithDegrees>>
- {
- private static final long serialVersionUID = 1L;
-
- final VertexUpdateFunction<K, VVWithDegrees, Message> vertexUpdateFunction;
-
- final MessageIterator<Message> messageIter = new MessageIterator<Message>();
-
- private transient TypeInformation<Vertex<K, VVWithDegrees>> resultType;
-
-
- private VertexUpdateUdf(VertexUpdateFunction<K, VVWithDegrees, Message> vertexUpdateFunction,
- TypeInformation<Vertex<K, VVWithDegrees>> resultType)
- {
- this.vertexUpdateFunction = vertexUpdateFunction;
- this.resultType = resultType;
- }
-
- @Override
- public void open(Configuration parameters) throws Exception {
- if (getIterationRuntimeContext().getSuperstepNumber() == 1) {
- this.vertexUpdateFunction.init(getIterationRuntimeContext());
- }
- this.vertexUpdateFunction.preSuperstep();
- }
-
- @Override
- public void close() throws Exception {
- this.vertexUpdateFunction.postSuperstep();
- }
-
- @Override
- public TypeInformation<Vertex<K, VVWithDegrees>> getProducedType() {
- return this.resultType;
- }
- }
-
- @SuppressWarnings("serial")
- private static final class VertexUpdateUdfSimpleVV<K, VV, Message> extends VertexUpdateUdf<K, VV, Message> {
-
- private VertexUpdateUdfSimpleVV(VertexUpdateFunction<K, VV, Message> vertexUpdateFunction, TypeInformation<Vertex<K, VV>> resultType) {
- super(vertexUpdateFunction, resultType);
- }
-
- @Override
- public void coGroup(Iterable<Tuple2<K, Message>> messages,
- Iterable<Vertex<K, VV>> vertex,
- Collector<Vertex<K, VV>> out) throws Exception {
- final Iterator<Vertex<K, VV>> vertexIter = vertex.iterator();
-
- if (vertexIter.hasNext()) {
- Vertex<K, VV> vertexState = vertexIter.next();
-
- @SuppressWarnings("unchecked")
- Iterator<Tuple2<?, Message>> downcastIter = (Iterator<Tuple2<?, Message>>) (Iterator<?>) messages.iterator();
- messageIter.setSource(downcastIter);
-
- vertexUpdateFunction.setOutput(vertexState, out);
- vertexUpdateFunction.updateVertex(vertexState, messageIter);
- }
- else {
- final Iterator<Tuple2<K, Message>> messageIter = messages.iterator();
- if (messageIter.hasNext()) {
- String message = "Target vertex does not exist!.";
- try {
- Tuple2<K, Message> next = messageIter.next();
- message = "Target vertex '" + next.f0 + "' does not exist!.";
- } catch (Throwable t) {}
- throw new Exception(message);
- } else {
- throw new Exception();
- }
- }
- }
- }
-
- @SuppressWarnings("serial")
- private static final class VertexUpdateUdfVVWithDegrees<K, VV, Message> extends VertexUpdateUdf<K, Tuple3<VV, Long, Long>, Message> {
-
- private VertexUpdateUdfVVWithDegrees(VertexUpdateFunction<K, Tuple3<VV, Long, Long>, Message> vertexUpdateFunction,
- TypeInformation<Vertex<K, Tuple3<VV, Long, Long>>> resultType) {
- super(vertexUpdateFunction, resultType);
- }
-
- @Override
- public void coGroup(Iterable<Tuple2<K, Message>> messages, Iterable<Vertex<K, Tuple3<VV, Long, Long>>> vertex,
- Collector<Vertex<K, Tuple3<VV, Long, Long>>> out) throws Exception {
-
- final Iterator<Vertex<K, Tuple3<VV, Long, Long>>> vertexIter = vertex.iterator();
-
- if (vertexIter.hasNext()) {
- Vertex<K, Tuple3<VV, Long, Long>> vertexWithDegrees = vertexIter.next();
-
- @SuppressWarnings("unchecked")
- Iterator<Tuple2<?, Message>> downcastIter = (Iterator<Tuple2<?, Message>>) (Iterator<?>) messages.iterator();
- messageIter.setSource(downcastIter);
-
- vertexUpdateFunction.setInDegree(vertexWithDegrees.f1.f1);
- vertexUpdateFunction.setOutDegree(vertexWithDegrees.f1.f2);
-
- vertexUpdateFunction.setOutputWithDegrees(vertexWithDegrees, out);
- vertexUpdateFunction.updateVertexFromVertexCentricIteration(vertexWithDegrees, messageIter);
- }
- else {
- final Iterator<Tuple2<K, Message>> messageIter = messages.iterator();
- if (messageIter.hasNext()) {
- String message = "Target vertex does not exist!.";
- try {
- Tuple2<K, Message> next = messageIter.next();
- message = "Target vertex '" + next.f0 + "' does not exist!.";
- } catch (Throwable t) {}
- throw new Exception(message);
- } else {
- throw new Exception();
- }
- }
- }
- }
-
- /*
- * UDF that encapsulates the message sending function for graphs where the edges have an associated value.
- */
- private static abstract class MessagingUdfWithEdgeValues<K, VVWithDegrees, VV, Message, EV>
- extends RichCoGroupFunction<Edge<K, EV>, Vertex<K, VVWithDegrees>, Tuple2<K, Message>>
- implements ResultTypeQueryable<Tuple2<K, Message>>
- {
- private static final long serialVersionUID = 1L;
-
- final MessagingFunction<K, VV, Message, EV> messagingFunction;
-
- private transient TypeInformation<Tuple2<K, Message>> resultType;
-
-
- private MessagingUdfWithEdgeValues(MessagingFunction<K, VV, Message, EV> messagingFunction,
- TypeInformation<Tuple2<K, Message>> resultType)
- {
- this.messagingFunction = messagingFunction;
- this.resultType = resultType;
- }
-
- @Override
- public void open(Configuration parameters) throws Exception {
- if (getIterationRuntimeContext().getSuperstepNumber() == 1) {
- this.messagingFunction.init(getIterationRuntimeContext());
- }
-
- this.messagingFunction.preSuperstep();
- }
-
- @Override
- public void close() throws Exception {
- this.messagingFunction.postSuperstep();
- }
-
- @Override
- public TypeInformation<Tuple2<K, Message>> getProducedType() {
- return this.resultType;
- }
- }
-
- @SuppressWarnings("serial")
- private static final class MessagingUdfWithEVsSimpleVV<K, VV, Message, EV>
- extends MessagingUdfWithEdgeValues<K, VV, VV, Message, EV> {
-
- private MessagingUdfWithEVsSimpleVV(MessagingFunction<K, VV, Message, EV> messagingFunction,
- TypeInformation<Tuple2<K, Message>> resultType) {
- super(messagingFunction, resultType);
- }
-
- @Override
- public void coGroup(Iterable<Edge<K, EV>> edges,
- Iterable<Vertex<K, VV>> state,
- Collector<Tuple2<K, Message>> out) throws Exception {
- final Iterator<Vertex<K, VV>> stateIter = state.iterator();
-
- if (stateIter.hasNext()) {
- Vertex<K, VV> newVertexState = stateIter.next();
- messagingFunction.set((Iterator<?>) edges.iterator(), out, newVertexState.getId());
- messagingFunction.sendMessages(newVertexState);
- }
- }
- }
-
- @SuppressWarnings("serial")
- private static final class MessagingUdfWithEVsVVWithDegrees<K, VV, Message, EV>
- extends MessagingUdfWithEdgeValues<K, Tuple3<VV, Long, Long>, VV, Message, EV> {
-
- private Vertex<K, VV> nextVertex = new Vertex<K, VV>();
-
- private MessagingUdfWithEVsVVWithDegrees(MessagingFunction<K, VV, Message, EV> messagingFunction,
- TypeInformation<Tuple2<K, Message>> resultType) {
- super(messagingFunction, resultType);
- }
-
- @Override
- public void coGroup(Iterable<Edge<K, EV>> edges, Iterable<Vertex<K, Tuple3<VV, Long, Long>>> state,
- Collector<Tuple2<K, Message>> out) throws Exception {
-
- final Iterator<Vertex<K, Tuple3<VV, Long, Long>>> stateIter = state.iterator();
-
- if (stateIter.hasNext()) {
- Vertex<K, Tuple3<VV, Long, Long>> vertexWithDegrees = stateIter.next();
-
- nextVertex.setField(vertexWithDegrees.f0, 0);
- nextVertex.setField(vertexWithDegrees.f1.f0, 1);
-
- messagingFunction.setInDegree(vertexWithDegrees.f1.f1);
- messagingFunction.setOutDegree(vertexWithDegrees.f1.f2);
-
- messagingFunction.set((Iterator<?>) edges.iterator(), out, vertexWithDegrees.getId());
- messagingFunction.sendMessages(nextVertex);
- }
- }
- }
-
-
- // --------------------------------------------------------------------------------------------
- // UTIL methods
- // --------------------------------------------------------------------------------------------
-
- /**
- * Method that builds the messaging function using a coGroup operator for a simple vertex(without
- * degrees).
- * It afterwards configures the function with a custom name and broadcast variables.
- *
- * @param iteration
- * @param messageTypeInfo
- * @param whereArg the argument for the where within the coGroup
- * @param equalToArg the argument for the equalTo within the coGroup
- * @return the messaging function
- */
- private CoGroupOperator<?, ?, Tuple2<K, Message>> buildMessagingFunction(
- DeltaIteration<Vertex<K, VV>, Vertex<K, VV>> iteration,
- TypeInformation<Tuple2<K, Message>> messageTypeInfo, int whereArg, int equalToArg) {
-
- // build the messaging function (co group)
- CoGroupOperator<?, ?, Tuple2<K, Message>> messages;
- MessagingUdfWithEdgeValues<K, VV, VV, Message, EV> messenger =
- new MessagingUdfWithEVsSimpleVV<K, VV, Message, EV>(messagingFunction, messageTypeInfo);
-
- messages = this.edgesWithValue.coGroup(iteration.getWorkset()).where(whereArg)
- .equalTo(equalToArg).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);
- }
- }
-
- return messages;
- }
-
- /**
- * Method that builds the messaging function using a coGroup operator for a vertex
- * containing degree information.
- * It afterwards configures the function with a custom name and broadcast variables.
- *
- * @param iteration
- * @param messageTypeInfo
- * @param whereArg the argument for the where within the coGroup
- * @param equalToArg the argument for the equalTo within the coGroup
- * @return the messaging function
- */
- private CoGroupOperator<?, ?, Tuple2<K, Message>> buildMessagingFunctionVerticesWithDegrees(
- DeltaIteration<Vertex<K, Tuple3<VV, Long, Long>>, Vertex<K, Tuple3<VV, Long, Long>>> iteration,
- TypeInformation<Tuple2<K, Message>> messageTypeInfo, int whereArg, int equalToArg) {
-
- // build the messaging function (co group)
- CoGroupOperator<?, ?, Tuple2<K, Message>> messages;
- MessagingUdfWithEdgeValues<K, Tuple3<VV, Long, Long>, VV, Message, EV> messenger =
- new MessagingUdfWithEVsVVWithDegrees<K, VV, Message, EV>(messagingFunction, messageTypeInfo);
-
- messages = this.edgesWithValue.coGroup(iteration.getWorkset()).where(whereArg)
- .equalTo(equalToArg).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);
- }
- }
-
- return messages;
- }
-
- /**
- * Helper method which sets up an iteration with the given vertex value(either simple or with degrees)
- *
- * @param iteration
- */
-
- private void setUpIteration(DeltaIteration<?, ?> iteration) {
-
- // set up the iteration operator
- if (this.configuration != null) {
-
- 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());
- }
- }
- else {
- // no configuration provided; set default name
- iteration.name("Vertex-centric iteration (" + updateFunction + " | " + messagingFunction + ")");
- }
- }
-
- /**
- * Creates the operator that represents this vertex centric graph computation for a simple vertex.
- *
- * @param messagingDirection
- * @param messageTypeInfo
- * @return the operator
- */
- private DataSet<Vertex<K, VV>> createResultSimpleVertex(EdgeDirection messagingDirection,
- TypeInformation<Tuple2<K, Message>> messageTypeInfo) {
-
- DataSet<Tuple2<K, Message>> messages;
-
- TypeInformation<Vertex<K, VV>> vertexTypes = initialVertices.getType();
-
- final DeltaIteration<Vertex<K, VV>, Vertex<K, VV>> iteration =
- initialVertices.iterateDelta(initialVertices, this.maximumNumberOfIterations, 0);
- setUpIteration(iteration);
-
- switch (messagingDirection) {
- case IN:
- messages = buildMessagingFunction(iteration, messageTypeInfo, 1, 0);
- break;
- case OUT:
- messages = buildMessagingFunction(iteration, messageTypeInfo, 0, 0);
- break;
- case ALL:
- messages = buildMessagingFunction(iteration, messageTypeInfo, 1, 0)
- .union(buildMessagingFunction(iteration, messageTypeInfo, 0, 0)) ;
- break;
- default:
- throw new IllegalArgumentException("Illegal edge direction");
- }
-
- VertexUpdateUdf<K, VV, Message> updateUdf =
- new VertexUpdateUdfSimpleVV<K, VV, Message>(updateFunction, vertexTypes);
-
- // build the update function (co group)
- CoGroupOperator<?, ?, Vertex<K, VV>> updates =
- messages.coGroup(iteration.getSolutionSet()).where(0).equalTo(0).with(updateUdf);
-
- configureUpdateFunction(updates);
-
- return iteration.closeWith(updates, updates);
- }
-
- /**
- * Creates the operator that represents this vertex centric graph computation for a vertex with in
- * and out degrees added to the vertex value.
- *
- * @param graph
- * @param messagingDirection
- * @param messageTypeInfo
- * @return the operator
- */
- @SuppressWarnings("serial")
- private DataSet<Vertex<K, VV>> createResultVerticesWithDegrees(Graph<K, VV, EV> graph, EdgeDirection messagingDirection,
- TypeInformation<Tuple2<K, Message>> messageTypeInfo) {
-
- DataSet<Tuple2<K, Message>> messages;
-
- this.updateFunction.setOptDegrees(this.configuration.isOptDegrees());
-
- DataSet<Tuple2<K, Long>> inDegrees = graph.inDegrees();
- DataSet<Tuple2<K, Long>> outDegrees = graph.outDegrees();
-
- DataSet<Tuple3<K, Long, Long>> degrees = 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) {
- out.collect(new Tuple3<K, Long, Long>(first.f0, first.f1, second.f1));
- }
- }).withForwardedFieldsFirst("f0;f1").withForwardedFieldsSecond("f1");
-
- DataSet<Vertex<K, Tuple3<VV, Long, Long>>> verticesWithDegrees = initialVertices
- .join(degrees).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> vertex, Tuple3<K, Long, Long> degrees,
- Collector<Vertex<K, Tuple3<VV, Long, Long>>> out) throws Exception {
-
- out.collect(new Vertex<K, Tuple3<VV, Long, Long>>(vertex.getId(),
- new Tuple3<VV, Long, Long>(vertex.getValue(), degrees.f1, degrees.f2)));
- }
- }).withForwardedFieldsFirst("f0");
-
- // add type info
- TypeInformation<Vertex<K, Tuple3<VV, Long, Long>>> vertexTypes = verticesWithDegrees.getType();
-
- final DeltaIteration<Vertex<K, Tuple3<VV, Long, Long>>, Vertex<K, Tuple3<VV, Long, Long>>> iteration =
- verticesWithDegrees.iterateDelta(verticesWithDegrees, this.maximumNumberOfIterations, 0);
- setUpIteration(iteration);
-
- switch (messagingDirection) {
- case IN:
- messages = buildMessagingFunctionVerticesWithDegrees(iteration, messageTypeInfo, 1, 0);
- break;
- case OUT:
- messages = buildMessagingFunctionVerticesWithDegrees(iteration, messageTypeInfo, 0, 0);
- break;
- case ALL:
- messages = buildMessagingFunctionVerticesWithDegrees(iteration, messageTypeInfo, 1, 0)
- .union(buildMessagingFunctionVerticesWithDegrees(iteration, messageTypeInfo, 0, 0)) ;
- break;
- default:
- throw new IllegalArgumentException("Illegal edge direction");
- }
-
- @SuppressWarnings({ "unchecked", "rawtypes" })
- VertexUpdateUdf<K, Tuple3<VV, Long, Long>, Message> updateUdf =
- new VertexUpdateUdfVVWithDegrees(updateFunction, vertexTypes);
-
- // build the update function (co group)
- CoGroupOperator<?, ?, Vertex<K, Tuple3<VV, Long, Long>>> updates =
- messages.coGroup(iteration.getSolutionSet()).where(0).equalTo(0).with(updateUdf);
-
- configureUpdateFunction(updates);
-
- return iteration.closeWith(updates, updates).map(
- new MapFunction<Vertex<K, Tuple3<VV, Long, Long>>, Vertex<K, VV>>() {
-
- public Vertex<K, VV> map(Vertex<K, Tuple3<VV, Long, Long>> vertex) {
- return new Vertex<K, VV>(vertex.getId(), vertex.getValue().f0);
- }
- });
- }
-
- private <VVWithDegree> void configureUpdateFunction(CoGroupOperator<?, ?, Vertex<K, VVWithDegree>> updates) {
-
- // 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);
- }
- }
-
- // let the operator know that we preserve the key field
- updates.withForwardedFieldsFirst("0").withForwardedFieldsSecond("0");
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/f8677464/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/spargel/VertexUpdateFunction.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/spargel/VertexUpdateFunction.java b/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/spargel/VertexUpdateFunction.java
index 857cef5..9085432 100644
--- a/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/spargel/VertexUpdateFunction.java
+++ b/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/spargel/VertexUpdateFunction.java
@@ -157,7 +157,7 @@ public abstract class VertexUpdateFunction<K, VV, Message> implements Serializab
/**
* Gets the broadcast data set registered under the given name. Broadcast data sets
* are available on all parallel instances of a function. They can be registered via
- * {@link org.apache.flink.graph.spargel.VertexCentricConfiguration#addBroadcastSetForUpdateFunction(String, org.apache.flink.api.java.DataSet)}.
+ * {@link org.apache.flink.graph.spargel.ScatterGatherConfiguration#addBroadcastSetForUpdateFunction(String, org.apache.flink.api.java.DataSet)}.
*
* @param name The name under which the broadcast set is registered.
* @return The broadcast data set.
@@ -230,7 +230,7 @@ public abstract class VertexUpdateFunction<K, VV, Message> implements Serializab
/**
* In order to hide the Tuple3(actualValue, inDegree, OutDegree) vertex value from the user,
- * another function will be called from {@link org.apache.flink.graph.spargel.VertexCentricIteration}.
+ * another function will be called from {@link org.apache.flink.graph.spargel.ScatterGatherIteration}.
*
* This function will retrieve the vertex from the vertexState and will set its degrees, afterwards calling
* the regular updateVertex function.
@@ -240,7 +240,7 @@ public abstract class VertexUpdateFunction<K, VV, Message> implements Serializab
* @throws Exception
*/
@SuppressWarnings("unchecked")
- <VertexWithDegree> void updateVertexFromVertexCentricIteration(Vertex<K, VertexWithDegree> vertexState,
+ <VertexWithDegree> void updateVertexFromScatterGatherIteration(Vertex<K, VertexWithDegree> vertexState,
MessageIterator<Message> inMessages) throws Exception {
Vertex<K, VV> vertex = new Vertex<K, VV>(vertexState.f0,
http://git-wip-us.apache.org/repos/asf/flink/blob/f8677464/flink-libraries/flink-gelly/src/test/java/org/apache/flink/graph/spargel/SpargelCompilerTest.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-gelly/src/test/java/org/apache/flink/graph/spargel/SpargelCompilerTest.java b/flink-libraries/flink-gelly/src/test/java/org/apache/flink/graph/spargel/SpargelCompilerTest.java
index 335481f..f4f9859 100644
--- a/flink-libraries/flink-gelly/src/test/java/org/apache/flink/graph/spargel/SpargelCompilerTest.java
+++ b/flink-libraries/flink-gelly/src/test/java/org/apache/flink/graph/spargel/SpargelCompilerTest.java
@@ -16,7 +16,6 @@
* limitations under the License.
*/
-
package org.apache.flink.graph.spargel;
import static org.junit.Assert.assertEquals;
@@ -74,7 +73,7 @@ public class SpargelCompilerTest extends CompilerTestBase {
Graph<Long, Long, NullValue> graph = Graph.fromDataSet(initialVertices, edges, env);
- DataSet<Vertex<Long, Long>> result = graph.runVertexCentricIteration(
+ DataSet<Vertex<Long, Long>> result = graph.runScatterGatherIteration(
new ConnectedComponents.CCUpdater<Long>(),
new ConnectedComponents.CCMessenger<Long>(), 100)
.getVertices();
@@ -156,11 +155,11 @@ public class SpargelCompilerTest extends CompilerTestBase {
Graph<Long, Long, NullValue> graph = Graph.fromDataSet(initialVertices, edges, env);
- VertexCentricConfiguration parameters = new VertexCentricConfiguration();
+ ScatterGatherConfiguration parameters = new ScatterGatherConfiguration();
parameters.addBroadcastSetForMessagingFunction(BC_VAR_NAME, bcVar);
parameters.addBroadcastSetForUpdateFunction(BC_VAR_NAME, bcVar);
- DataSet<Vertex<Long, Long>> result = graph.runVertexCentricIteration(
+ DataSet<Vertex<Long, Long>> result = graph.runScatterGatherIteration(
new ConnectedComponents.CCUpdater<Long>(),
new ConnectedComponents.CCMessenger<Long>(), 100)
.getVertices();
http://git-wip-us.apache.org/repos/asf/flink/blob/f8677464/flink-libraries/flink-gelly/src/test/java/org/apache/flink/graph/spargel/SpargelTranslationTest.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-gelly/src/test/java/org/apache/flink/graph/spargel/SpargelTranslationTest.java b/flink-libraries/flink-gelly/src/test/java/org/apache/flink/graph/spargel/SpargelTranslationTest.java
index 69aa99c..7755708 100644
--- a/flink-libraries/flink-gelly/src/test/java/org/apache/flink/graph/spargel/SpargelTranslationTest.java
+++ b/flink-libraries/flink-gelly/src/test/java/org/apache/flink/graph/spargel/SpargelTranslationTest.java
@@ -81,7 +81,7 @@ public class SpargelTranslationTest {
}
}), env);
- VertexCentricConfiguration parameters = new VertexCentricConfiguration();
+ ScatterGatherConfiguration parameters = new ScatterGatherConfiguration();
parameters.addBroadcastSetForMessagingFunction(BC_SET_MESSAGES_NAME, bcMessaging);
parameters.addBroadcastSetForUpdateFunction(BC_SET_UPDATES_NAME, bcUpdate);
@@ -89,7 +89,7 @@ public class SpargelTranslationTest {
parameters.setParallelism(ITERATION_parallelism);
parameters.registerAggregator(AGGREGATOR_NAME, new LongSumAggregator());
- result = graph.runVertexCentricIteration(new UpdateFunction(), new MessageFunctionNoEdgeValue(),
+ result = graph.runScatterGatherIteration(new UpdateFunction(), new MessageFunctionNoEdgeValue(),
NUM_ITERATIONS, parameters).getVertices();
result.output(new DiscardingOutputFormat<Vertex<String, Double>>());
@@ -167,7 +167,7 @@ public class SpargelTranslationTest {
}
}), env);
- VertexCentricConfiguration parameters = new VertexCentricConfiguration();
+ ScatterGatherConfiguration parameters = new ScatterGatherConfiguration();
parameters.addBroadcastSetForMessagingFunction(BC_SET_MESSAGES_NAME, bcVar);
parameters.addBroadcastSetForUpdateFunction(BC_SET_UPDATES_NAME, bcVar);
@@ -175,7 +175,7 @@ public class SpargelTranslationTest {
parameters.setParallelism(ITERATION_parallelism);
parameters.registerAggregator(AGGREGATOR_NAME, new LongSumAggregator());
- result = graph.runVertexCentricIteration(new UpdateFunction(), new MessageFunctionNoEdgeValue(),
+ result = graph.runScatterGatherIteration(new UpdateFunction(), new MessageFunctionNoEdgeValue(),
NUM_ITERATIONS, parameters).getVertices();
result.output(new DiscardingOutputFormat<Vertex<String, Double>>());
http://git-wip-us.apache.org/repos/asf/flink/blob/f8677464/flink-libraries/flink-gelly/src/test/java/org/apache/flink/graph/test/CollectionModeSuperstepITCase.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-gelly/src/test/java/org/apache/flink/graph/test/CollectionModeSuperstepITCase.java b/flink-libraries/flink-gelly/src/test/java/org/apache/flink/graph/test/CollectionModeSuperstepITCase.java
index 61fe0c2..b1c2a2c 100644
--- a/flink-libraries/flink-gelly/src/test/java/org/apache/flink/graph/test/CollectionModeSuperstepITCase.java
+++ b/flink-libraries/flink-gelly/src/test/java/org/apache/flink/graph/test/CollectionModeSuperstepITCase.java
@@ -47,7 +47,7 @@ public class CollectionModeSuperstepITCase {
Graph<Long, Long, Long> graph = Graph.fromCollection(TestGraphUtils.getLongLongVertices(),
TestGraphUtils.getLongLongEdges(), env).mapVertices(new AssignOneMapper());
- Graph<Long, Long, Long> result = graph.runVertexCentricIteration(
+ Graph<Long, Long, Long> result = graph.runScatterGatherIteration(
new UpdateFunction(), new MessageFunction(), 10);
result.getVertices().map(
http://git-wip-us.apache.org/repos/asf/flink/blob/f8677464/flink-libraries/flink-gelly/src/test/java/org/apache/flink/graph/test/ScatterGatherConfigurationITCase.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-gelly/src/test/java/org/apache/flink/graph/test/ScatterGatherConfigurationITCase.java b/flink-libraries/flink-gelly/src/test/java/org/apache/flink/graph/test/ScatterGatherConfigurationITCase.java
new file mode 100644
index 0000000..f14e002
--- /dev/null
+++ b/flink-libraries/flink-gelly/src/test/java/org/apache/flink/graph/test/ScatterGatherConfigurationITCase.java
@@ -0,0 +1,796 @@
+/*
+ * 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;
+
+import java.util.HashSet;
+import java.util.List;
+
+import org.apache.flink.api.common.aggregators.LongSumAggregator;
+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.Tuple1;
+import org.apache.flink.api.java.tuple.Tuple2;
+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.spargel.MessageIterator;
+import org.apache.flink.graph.spargel.MessagingFunction;
+import org.apache.flink.graph.spargel.ScatterGatherConfiguration;
+import org.apache.flink.graph.spargel.ScatterGatherIteration;
+import org.apache.flink.graph.spargel.VertexUpdateFunction;
+import org.apache.flink.test.util.MultipleProgramsTestBase;
+import org.apache.flink.types.LongValue;
+import org.junit.Assert;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+import org.apache.flink.graph.utils.VertexToTuple2Map;
+
+@RunWith(Parameterized.class)
+public class ScatterGatherConfigurationITCase extends MultipleProgramsTestBase {
+
+ public ScatterGatherConfigurationITCase(TestExecutionMode mode){
+ super(mode);
+ }
+
+ private String expectedResult;
+
+ @Test
+ public void testRunWithConfiguration() throws Exception {
+ /*
+ * Test Graph's runScatterGatherIteration when configuration parameters are provided
+ */
+ final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+
+ Graph<Long, Long, Long> graph = Graph.fromCollection(TestGraphUtils.getLongLongVertices(),
+ TestGraphUtils.getLongLongEdges(), env).mapVertices(new AssignOneMapper());
+
+ // create the configuration object
+ ScatterGatherConfiguration parameters = new ScatterGatherConfiguration();
+
+ parameters.addBroadcastSetForUpdateFunction("updateBcastSet", env.fromElements(1, 2, 3));
+ parameters.addBroadcastSetForMessagingFunction("messagingBcastSet", env.fromElements(4, 5, 6));
+ parameters.registerAggregator("superstepAggregator", new LongSumAggregator());
+ parameters.setOptNumVertices(true);
+
+ Graph<Long, Long, Long> res = graph.runScatterGatherIteration(
+ new UpdateFunction(), new MessageFunction(), 10, parameters);
+
+ DataSet<Vertex<Long,Long>> data = res.getVertices();
+ List<Vertex<Long,Long>> result= data.collect();
+
+ expectedResult = "1,11\n" +
+ "2,11\n" +
+ "3,11\n" +
+ "4,11\n" +
+ "5,11";
+
+ compareResultAsTuples(result, expectedResult);
+ }
+
+ @Test
+ public void testIterationConfiguration() throws Exception {
+
+ /*
+ * Test name, parallelism and solutionSetUnmanaged parameters
+ */
+ final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+
+ ScatterGatherIteration<Long, Long, Long, Long> iteration = ScatterGatherIteration
+ .withEdges(TestGraphUtils.getLongLongEdgeData(env), new DummyUpdateFunction(),
+ new DummyMessageFunction(), 10);
+
+ ScatterGatherConfiguration parameters = new ScatterGatherConfiguration();
+ parameters.setName("gelly iteration");
+ parameters.setParallelism(2);
+ parameters.setSolutionSetUnmanagedMemory(true);
+
+ iteration.configure(parameters);
+
+ Assert.assertEquals("gelly iteration", iteration.getIterationConfiguration().getName(""));
+ Assert.assertEquals(2, iteration.getIterationConfiguration().getParallelism());
+ Assert.assertEquals(true, iteration.getIterationConfiguration().isSolutionSetUnmanagedMemory());
+
+ DataSet<Vertex<Long, Long>> data = TestGraphUtils.getLongLongVertexData(env).runOperation(iteration);
+ List<Vertex<Long,Long>> result= data.collect();
+
+ expectedResult = "1,11\n" +
+ "2,12\n" +
+ "3,13\n" +
+ "4,14\n" +
+ "5,15";
+
+ compareResultAsTuples(result, expectedResult);
+ }
+
+ @Test
+ public void testDefaultConfiguration() throws Exception {
+ /*
+ * Test Graph's runScatterGatherIteration when configuration parameters are not provided
+ * i.e. degrees and numVertices will be -1, EdgeDirection will be OUT.
+ */
+ final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+
+ Graph<Long, Long, Long> graph = Graph.fromCollection(TestGraphUtils.getLongLongVertices(),
+ TestGraphUtils.getLongLongEdges(), env).mapVertices(new AssignOneMapper());
+
+ Graph<Long, Long, Long> res = graph.runScatterGatherIteration(
+ new UpdateFunctionDefault(), new MessageFunctionDefault(), 5);
+
+
+ DataSet<Tuple2<Long, Long>> data = res.getVertices().map(new VertexToTuple2Map<Long, Long>());
+ List<Tuple2<Long, Long>> result= data.collect();
+
+ expectedResult = "1,6\n" +
+ "2,6\n" +
+ "3,6\n" +
+ "4,6\n" +
+ "5,6";
+
+ compareResultAsTuples(result, expectedResult);
+ }
+
+ @Test
+ public void testIterationDefaultDirection() throws Exception {
+
+ /*
+ * Test that if no direction parameter is given, the iteration works as before
+ * (i.e. it collects messages from the in-neighbors and sends them to the out-neighbors)
+ */
+ final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+
+ Graph<Long, HashSet<Long>, Long> graph = Graph
+ .fromCollection(TestGraphUtils.getLongLongVertices(), TestGraphUtils.getLongLongEdges(), env)
+ .mapVertices(new InitialiseHashSetMapper());
+
+ DataSet<Vertex<Long, HashSet<Long>>> resultedVertices = graph
+ .runScatterGatherIteration(new VertexUpdateDirection(), new IdMessengerTrg(), 5)
+ .getVertices();
+
+ List<Vertex<Long, HashSet<Long>>> result= resultedVertices.collect();
+
+ expectedResult = "1,[5]\n" +
+ "2,[1]\n" +
+ "3,[1, 2]\n" +
+ "4,[3]\n" +
+ "5,[3, 4]";
+
+ compareResultAsTuples(result, expectedResult);
+ }
+
+ @Test
+ public void testIterationINDirection() throws Exception {
+
+ /*
+ * Test that if the direction parameter is set to IN,
+ * messages are collected from the out-neighbors and sent to the in-neighbors.
+ */
+ final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+
+ Graph<Long, HashSet<Long>, Long> graph = Graph
+ .fromCollection(TestGraphUtils.getLongLongVertices(), TestGraphUtils.getLongLongEdges(), env)
+ .mapVertices(new InitialiseHashSetMapper());
+
+ // configure the iteration
+ ScatterGatherConfiguration parameters = new ScatterGatherConfiguration();
+
+ parameters.setDirection(EdgeDirection.IN);
+
+ DataSet<Vertex<Long, HashSet<Long>>> resultedVertices = graph
+ .runScatterGatherIteration(new VertexUpdateDirection(), new IdMessengerSrc(), 5, parameters)
+ .getVertices();
+
+ List<Vertex<Long, HashSet<Long>>> result= resultedVertices.collect();
+
+ expectedResult = "1,[2, 3]\n" +
+ "2,[3]\n" +
+ "3,[4, 5]\n" +
+ "4,[5]\n" +
+ "5,[1]";
+
+ compareResultAsTuples(result, expectedResult);
+ }
+
+ @Test
+ public void testIterationALLDirection() throws Exception {
+
+ /*
+ * Test that if the direction parameter is set to ALL,
+ * messages are collected from all the neighbors and sent to all the neighbors.
+ */
+ final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+
+ Graph<Long, HashSet<Long>, Long> graph = Graph
+ .fromCollection(TestGraphUtils.getLongLongVertices(), TestGraphUtils.getLongLongEdges(), env)
+ .mapVertices(new InitialiseHashSetMapper());
+
+ // configure the iteration
+ ScatterGatherConfiguration parameters = new ScatterGatherConfiguration();
+
+ parameters.setDirection(EdgeDirection.ALL);
+
+ DataSet<Vertex<Long, HashSet<Long>>> resultedVertices = graph
+ .runScatterGatherIteration(new VertexUpdateDirection(), new IdMessengerAll(), 5, parameters)
+ .getVertices();
+
+ List<Vertex<Long, HashSet<Long>>> result= resultedVertices.collect();
+
+ expectedResult = "1,[2, 3, 5]\n" +
+ "2,[1, 3]\n" +
+ "3,[1, 2, 4, 5]\n" +
+ "4,[3, 5]\n" +
+ "5,[1, 3, 4]";
+
+ compareResultAsTuples(result, expectedResult);
+ }
+
+ @Test
+ public void testSendToAllDirectionIN() throws Exception {
+
+ /*
+ * Test that sendMessageToAllNeighbors() works correctly
+ * when the direction is set to IN
+ */
+ final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+
+ Graph<Long, HashSet<Long>, Long> graph = Graph
+ .fromCollection(TestGraphUtils.getLongLongVertices(), TestGraphUtils.getLongLongEdges(), env)
+ .mapVertices(new InitialiseHashSetMapper());
+
+ // configure the iteration
+ ScatterGatherConfiguration parameters = new ScatterGatherConfiguration();
+
+ parameters.setDirection(EdgeDirection.IN);
+
+ DataSet<Vertex<Long, HashSet<Long>>> resultedVertices = graph
+ .runScatterGatherIteration(new VertexUpdateDirection(), new SendMsgToAll(), 5, parameters)
+ .getVertices();
+
+ List<Vertex<Long, HashSet<Long>>> result = resultedVertices.collect();
+
+ expectedResult = "1,[2, 3]\n" +
+ "2,[3]\n" +
+ "3,[4, 5]\n" +
+ "4,[5]\n" +
+ "5,[1]";
+
+ compareResultAsTuples(result, expectedResult);
+ }
+
+ @Test
+ public void testSendToAllDirectionOUT() throws Exception {
+
+ /*
+ * Test that sendMessageToAllNeighbors() works correctly
+ * when the direction is set to OUT
+ */
+ final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+
+ Graph<Long, HashSet<Long>, Long> graph = Graph
+ .fromCollection(TestGraphUtils.getLongLongVertices(), TestGraphUtils.getLongLongEdges(), env)
+ .mapVertices(new InitialiseHashSetMapper());
+
+ // configure the iteration
+ ScatterGatherConfiguration parameters = new ScatterGatherConfiguration();
+
+ parameters.setDirection(EdgeDirection.OUT);
+
+ DataSet<Vertex<Long, HashSet<Long>>> resultedVertices = graph
+ .runScatterGatherIteration(new VertexUpdateDirection(), new SendMsgToAll(), 5, parameters)
+ .getVertices();
+
+ List<Vertex<Long, HashSet<Long>>> result = resultedVertices.collect();
+
+ expectedResult = "1,[5]\n" +
+ "2,[1]\n" +
+ "3,[1, 2]\n" +
+ "4,[3]\n" +
+ "5,[3, 4]";
+
+ compareResultAsTuples(result, expectedResult);
+ }
+
+ @Test
+ public void testSendToAllDirectionALL() throws Exception {
+
+ /*
+ * Test that sendMessageToAllNeighbors() works correctly
+ * when the direction is set to ALL
+ */
+ final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+
+ Graph<Long, HashSet<Long>, Long> graph = Graph
+ .fromCollection(TestGraphUtils.getLongLongVertices(), TestGraphUtils.getLongLongEdges(), env)
+ .mapVertices(new InitialiseHashSetMapper());
+
+ // configure the iteration
+ ScatterGatherConfiguration parameters = new ScatterGatherConfiguration();
+
+ parameters.setDirection(EdgeDirection.ALL);
+
+ DataSet<Vertex<Long, HashSet<Long>>> resultedVertices = graph
+ .runScatterGatherIteration(new VertexUpdateDirection(), new SendMsgToAll(), 5, parameters)
+ .getVertices();
+
+ List<Vertex<Long, HashSet<Long>>> result = resultedVertices.collect();
+
+ expectedResult = "1,[2, 3, 5]\n" +
+ "2,[1, 3]\n" +
+ "3,[1, 2, 4, 5]\n" +
+ "4,[3, 5]\n" +
+ "5,[1, 3, 4]";
+
+ compareResultAsTuples(result, expectedResult);
+ }
+
+
+ @Test
+ public void testNumVerticesNotSet() throws Exception {
+
+ /*
+ * Test that if the number of vertices option is not set, -1 is returned as value.
+ */
+ final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+
+ Graph<Long, Long, Long> graph = Graph.fromCollection(TestGraphUtils.getLongLongVertices(),
+ TestGraphUtils.getLongLongEdges(), env);
+
+ DataSet<Vertex<Long, Long>> verticesWithNumVertices = graph.runScatterGatherIteration(new UpdateFunctionNumVertices(),
+ new DummyMessageFunction(), 2).getVertices();
+
+ List<Vertex<Long, Long>> result= verticesWithNumVertices.collect();
+
+ expectedResult = "1,-1\n" +
+ "2,-1\n" +
+ "3,-1\n" +
+ "4,-1\n" +
+ "5,-1";
+
+ compareResultAsTuples(result, expectedResult);
+ }
+
+ @Test
+ public void testInDegreesSet() throws Exception {
+
+ /*
+ * Test that if the degrees are set, they can be accessed in every superstep
+ * inside the update function and the value
+ * is correctly computed for degrees in the messaging function.
+ */
+ final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+
+ Graph<Long, Long, Long> graph = Graph.fromCollection(TestGraphUtils.getLongLongVertices(),
+ TestGraphUtils.getLongLongEdges(), env);
+
+ // configure the iteration
+ ScatterGatherConfiguration parameters = new ScatterGatherConfiguration();
+
+ parameters.setOptDegrees(true);
+
+ DataSet<Vertex<Long, Long>> verticesWithDegrees = graph.runScatterGatherIteration(
+ new UpdateFunctionInDegrees(), new DegreesMessageFunction(), 5, parameters).getVertices();
+
+ List<Vertex<Long, Long>> result= verticesWithDegrees.collect();
+
+ expectedResult = "1,1\n" +
+ "2,1\n" +
+ "3,2\n" +
+ "4,1\n" +
+ "5,2";
+
+ compareResultAsTuples(result, expectedResult);
+ }
+
+ @Test
+ public void testInDegreesNotSet() throws Exception {
+
+ /*
+ * Test that if the degrees option is not set, then -1 is returned as a value for in-degree.
+ */
+ final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+
+ Graph<Long, Long, Long> graph = Graph.fromCollection(TestGraphUtils.getLongLongVertices(),
+ TestGraphUtils.getLongLongEdges(), env);
+
+ DataSet<Vertex<Long, Long>> verticesWithDegrees = graph.runScatterGatherIteration(
+ new UpdateFunctionInDegrees(), new DummyMessageFunction(), 2).getVertices();
+
+ List<Vertex<Long, Long>> result= verticesWithDegrees.collect();
+
+ expectedResult = "1,-1\n" +
+ "2,-1\n" +
+ "3,-1\n" +
+ "4,-1\n" +
+ "5,-1";
+
+ compareResultAsTuples(result, expectedResult);
+ }
+
+ @Test
+ public void testOutDegreesSet() throws Exception {
+
+ /*
+ * Test that if the degrees are set, they can be accessed in every superstep
+ * inside the update function and the value
+ * is correctly computed for degrees in the messaging function.
+ */
+ final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+
+ Graph<Long, Long, Long> graph = Graph.fromCollection(TestGraphUtils.getLongLongVertices(),
+ TestGraphUtils.getLongLongEdges(), env);
+
+ // configure the iteration
+ ScatterGatherConfiguration parameters = new ScatterGatherConfiguration();
+
+ parameters.setOptDegrees(true);
+
+ DataSet<Vertex<Long, Long>> verticesWithDegrees = graph.runScatterGatherIteration(
+ new UpdateFunctionOutDegrees(), new DegreesMessageFunction(), 5, parameters).getVertices();
+
+ List<Vertex<Long, Long>> result= verticesWithDegrees.collect();
+
+ expectedResult = "1,2\n" +
+ "2,1\n" +
+ "3,2\n" +
+ "4,1\n" +
+ "5,1";
+
+ compareResultAsTuples(result, expectedResult);
+ }
+
+ @Test
+ public void testOutDegreesNotSet() throws Exception {
+
+ /*
+ * Test that if the degrees option is not set, then -1 is returned as a value for out-degree.
+ */
+ final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+
+ Graph<Long, Long, Long> graph = Graph.fromCollection(TestGraphUtils.getLongLongVertices(),
+ TestGraphUtils.getLongLongEdges(), env);
+
+ DataSet<Vertex<Long, Long>> verticesWithDegrees = graph.runScatterGatherIteration(
+ new UpdateFunctionInDegrees(), new DummyMessageFunction(), 2).getVertices();
+
+ List<Vertex<Long, Long>> result= verticesWithDegrees.collect();
+
+ expectedResult = "1,-1\n" +
+ "2,-1\n" +
+ "3,-1\n" +
+ "4,-1\n" +
+ "5,-1";
+
+ compareResultAsTuples(result, expectedResult);
+ }
+
+ @Test
+ public void testDirectionALLAndDegrees() throws Exception {
+
+ /*
+ * Compute the number of neighbors in a vertex - centric manner, and verify that it is equal to
+ * the sum: inDegree + outDegree.
+ */
+ final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+
+ Graph<Long, Boolean, Long> graph = Graph.fromCollection(TestGraphUtils.getLongBooleanVertices(),
+ TestGraphUtils.getLongLongEdges(), env);
+
+ // configure the iteration
+ ScatterGatherConfiguration parameters = new ScatterGatherConfiguration();
+
+ parameters.setOptDegrees(true);
+ parameters.setDirection(EdgeDirection.ALL);
+
+ DataSet<Vertex<Long, Boolean>> verticesWithNumNeighbors = graph.runScatterGatherIteration(
+ new VertexUpdateNumNeighbors(), new IdMessenger(), 1, parameters).getVertices();
+
+ List<Vertex<Long, Boolean>> result= verticesWithNumNeighbors.collect();
+
+ expectedResult = "1,true\n" +
+ "2,true\n" +
+ "3,true\n" +
+ "4,true\n" +
+ "5,true";
+
+ compareResultAsTuples(result, expectedResult);
+ }
+
+ @SuppressWarnings("serial")
+ public static final class UpdateFunction extends VertexUpdateFunction<Long, Long, Long> {
+
+ LongSumAggregator aggregator = new LongSumAggregator();
+
+ @Override
+ public void preSuperstep() {
+
+ // test bcast variable
+ @SuppressWarnings("unchecked")
+ List<Tuple1<Integer>> bcastSet = (List<Tuple1<Integer>>)(List<?>)getBroadcastSet("updateBcastSet");
+ Assert.assertEquals(1, bcastSet.get(0));
+ Assert.assertEquals(2, bcastSet.get(1));
+ Assert.assertEquals(3, bcastSet.get(2));
+
+ // test aggregator
+ aggregator = getIterationAggregator("superstepAggregator");
+
+ // test number of vertices
+ Assert.assertEquals(5, getNumberOfVertices());
+
+ }
+
+ @Override
+ public void updateVertex(Vertex<Long, Long> vertex, MessageIterator<Long> inMessages) {
+ long superstep = getSuperstepNumber();
+ aggregator.aggregate(superstep);
+
+ setNewVertexValue(vertex.getValue() + 1);
+ }
+ }
+
+ @SuppressWarnings("serial")
+ public static final class UpdateFunctionDefault extends VertexUpdateFunction<Long, Long, Long> {
+
+ @Override
+ public void updateVertex(Vertex<Long, Long> vertex, MessageIterator<Long> inMessages) {
+
+ // test number of vertices
+ Assert.assertEquals(-1, getNumberOfVertices());
+
+ // test degrees
+ Assert.assertEquals(-1, getInDegree());
+ Assert.assertEquals(-1, getOutDegree());
+
+ setNewVertexValue(vertex.getValue() + 1);
+ }
+ }
+
+ @SuppressWarnings("serial")
+ public static final class MessageFunction extends MessagingFunction<Long, Long, Long, Long> {
+
+ @Override
+ public void preSuperstep() {
+
+ // test bcast variable
+ @SuppressWarnings("unchecked")
+ List<Tuple1<Integer>> bcastSet = (List<Tuple1<Integer>>)(List<?>)getBroadcastSet("messagingBcastSet");
+ Assert.assertEquals(4, bcastSet.get(0));
+ Assert.assertEquals(5, bcastSet.get(1));
+ Assert.assertEquals(6, bcastSet.get(2));
+
+ // test number of vertices
+ Assert.assertEquals(5, getNumberOfVertices());
+
+ // test aggregator
+ if (getSuperstepNumber() == 2) {
+ long aggrValue = ((LongValue)getPreviousIterationAggregate("superstepAggregator")).getValue();
+ Assert.assertEquals(5, aggrValue);
+ }
+ }
+
+ @Override
+ public void sendMessages(Vertex<Long, Long> vertex) {
+ //send message to keep vertices active
+ sendMessageToAllNeighbors(vertex.getValue());
+ }
+ }
+
+ @SuppressWarnings("serial")
+ public static final class MessageFunctionDefault extends MessagingFunction<Long, Long, Long, Long> {
+
+ @Override
+ public void sendMessages(Vertex<Long, Long> vertex) {
+ // test number of vertices
+ Assert.assertEquals(-1, getNumberOfVertices());
+
+ // test degrees
+ Assert.assertEquals(-1, getInDegree());
+ Assert.assertEquals(-1, getOutDegree());
+ //send message to keep vertices active
+ sendMessageToAllNeighbors(vertex.getValue());
+ }
+ }
+
+ @SuppressWarnings("serial")
+ public static final class UpdateFunctionNumVertices extends VertexUpdateFunction<Long, Long, Long> {
+
+ @Override
+ public void updateVertex(Vertex<Long, Long> vertex, MessageIterator<Long> inMessages) {
+ setNewVertexValue(getNumberOfVertices());
+ }
+ }
+
+ @SuppressWarnings("serial")
+ public static final class DummyUpdateFunction extends VertexUpdateFunction<Long, Long, Long> {
+
+ @Override
+ public void updateVertex(Vertex<Long, Long> vertex, MessageIterator<Long> inMessages) {
+ setNewVertexValue(vertex.getValue() + 1);
+ }
+ }
+
+ @SuppressWarnings("serial")
+ public static final class DummyMessageFunction extends MessagingFunction<Long, Long, Long, Long> {
+
+ @Override
+ public void sendMessages(Vertex<Long, Long> vertex) {
+ //send message to keep vertices active
+ sendMessageToAllNeighbors(vertex.getValue());
+ }
+ }
+
+ @SuppressWarnings("serial")
+ public static final class DegreesMessageFunction extends MessagingFunction<Long, Long, Long, Long> {
+
+ @Override
+ public void sendMessages(Vertex<Long, Long> vertex) {
+ if (vertex.getId() == 1) {
+ Assert.assertEquals(2, getOutDegree());
+ Assert.assertEquals(1, getInDegree());
+ }
+ else if(vertex.getId() == 3) {
+ Assert.assertEquals(2, getOutDegree());
+ Assert.assertEquals(2, getInDegree());
+ }
+ //send message to keep vertices active
+ sendMessageToAllNeighbors(vertex.getValue());
+ }
+ }
+
+ @SuppressWarnings("serial")
+ public static final class VertexUpdateDirection extends VertexUpdateFunction<Long, HashSet<Long>, Long> {
+
+ @Override
+ public void updateVertex(Vertex<Long, HashSet<Long>> vertex, MessageIterator<Long> messages) throws Exception {
+ vertex.getValue().clear();
+
+ for(long msg : messages) {
+ vertex.getValue().add(msg);
+ }
+
+ setNewVertexValue(vertex.getValue());
+ }
+ }
+
+ @SuppressWarnings("serial")
+ public static final class UpdateFunctionInDegrees extends VertexUpdateFunction<Long, Long, Long> {
+
+ @Override
+ public void updateVertex(Vertex<Long, Long> vertex, MessageIterator<Long> inMessages) {
+ long inDegree = getInDegree();
+ setNewVertexValue(inDegree);
+ }
+ }
+
+ @SuppressWarnings("serial")
+ public static final class UpdateFunctionOutDegrees extends VertexUpdateFunction<Long, Long, Long> {
+
+ @Override
+ public void updateVertex(Vertex<Long, Long> vertex, MessageIterator<Long> inMessages) {
+ long outDegree = getOutDegree();
+ setNewVertexValue(outDegree);
+ }
+ }
+
+ @SuppressWarnings("serial")
+ public static final class VertexUpdateNumNeighbors extends VertexUpdateFunction<Long, Boolean,
+ Long> {
+
+ @Override
+ public void updateVertex(Vertex<Long, Boolean> vertex, MessageIterator<Long> messages) throws Exception {
+
+ long count = 0;
+
+ for(@SuppressWarnings("unused") long msg : messages) {
+ count++;
+ }
+ setNewVertexValue(count == (getInDegree() + getOutDegree()));
+ }
+ }
+
+ @SuppressWarnings("serial")
+ public static final class UpdateFunctionDegrees extends VertexUpdateFunction<Long, Long, Long> {
+
+ @Override
+ public void updateVertex(Vertex<Long, Long> vertex, MessageIterator<Long> inMessages) {
+ long inDegree = getInDegree();
+ long outDegree = getOutDegree();
+ setNewVertexValue(inDegree + outDegree);
+ }
+ }
+
+ @SuppressWarnings("serial")
+ public static final class IdMessengerSrc extends MessagingFunction<Long, HashSet<Long>, Long, Long> {
+
+ @Override
+ public void sendMessages(Vertex<Long, HashSet<Long>> vertex) throws Exception {
+ for (Edge<Long, Long> edge : getEdges()) {
+ sendMessageTo(edge.getSource(), vertex.getId());
+ }
+ }
+ }
+
+ @SuppressWarnings("serial")
+ public static final class IdMessengerAll extends MessagingFunction<Long, HashSet<Long>, Long, Long> {
+
+ @Override
+ public void sendMessages(Vertex<Long, HashSet<Long>> vertex) throws Exception {
+ for (Edge<Long, Long> edge : getEdges()) {
+ if(!edge.getSource().equals(vertex.getId())) {
+ sendMessageTo(edge.getSource(), vertex.getId());
+ } else {
+ sendMessageTo(edge.getTarget(), vertex.getId());
+ }
+ }
+ }
+ }
+
+ @SuppressWarnings("serial")
+ public static final class SendMsgToAll extends MessagingFunction<Long, HashSet<Long>, Long, Long> {
+
+ @Override
+ public void sendMessages(Vertex<Long, HashSet<Long>> vertex) throws Exception {
+ sendMessageToAllNeighbors(vertex.getId());
+ }
+ }
+
+ @SuppressWarnings("serial")
+ public static final class IdMessenger extends MessagingFunction<Long, Boolean, Long, Long> {
+
+ @Override
+ public void sendMessages(Vertex<Long, Boolean> vertex) throws Exception {
+ for (Edge<Long, Long> edge : getEdges()) {
+ if(!edge.getSource().equals(vertex.getId())) {
+ sendMessageTo(edge.getSource(), vertex.getId());
+ } else {
+ sendMessageTo(edge.getTarget(), vertex.getId());
+ }
+ }
+ }
+ }
+
+ @SuppressWarnings("serial")
+ public static final class IdMessengerTrg extends MessagingFunction<Long, HashSet<Long>, Long, Long> {
+
+ @Override
+ public void sendMessages(Vertex<Long, HashSet<Long>> vertex) throws Exception {
+ for (Edge<Long, Long> edge : getEdges()) {
+ sendMessageTo(edge.getTarget(), vertex.getId());
+ }
+ }
+ }
+
+ @SuppressWarnings("serial")
+ public static final class AssignOneMapper implements MapFunction<Vertex<Long, Long>, Long> {
+
+ public Long map(Vertex<Long, Long> value) {
+ return 1L;
+ }
+ }
+
+ @SuppressWarnings("serial")
+ public static final class InitialiseHashSetMapper implements MapFunction<Vertex<Long, Long>, HashSet<Long>> {
+
+ @Override
+ public HashSet<Long> map(Vertex<Long, Long> value) throws Exception {
+ return new HashSet<>();
+ }
+ }
+}