http://git-wip-us.apache.org/repos/asf/flink/blob/918e5d0c/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/library/SingleSourceShortestPaths.java ---------------------------------------------------------------------- diff --git a/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/library/SingleSourceShortestPaths.java b/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/library/SingleSourceShortestPaths.java index 29183e9..4ff4e79 100644 --- a/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/library/SingleSourceShortestPaths.java +++ b/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/library/SingleSourceShortestPaths.java @@ -24,9 +24,9 @@ import org.apache.flink.graph.Edge; import org.apache.flink.graph.Graph; import org.apache.flink.graph.GraphAlgorithm; import org.apache.flink.graph.Vertex; +import org.apache.flink.graph.spargel.GatherFunction; import org.apache.flink.graph.spargel.MessageIterator; -import org.apache.flink.graph.spargel.MessagingFunction; -import org.apache.flink.graph.spargel.VertexUpdateFunction; +import org.apache.flink.graph.spargel.ScatterFunction; /** * This is an implementation of the Single-Source-Shortest Paths algorithm, using a scatter-gather iteration. @@ -52,7 +52,7 @@ public class SingleSourceShortestPaths<K> implements GraphAlgorithm<K, Double, D public DataSet<Vertex<K, Double>> run(Graph<K, Double, Double> input) { return input.mapVertices(new InitVerticesMapper<K>(srcVertexId)) - .runScatterGatherIteration(new VertexDistanceUpdater<K>(), new MinDistanceMessenger<K>(), + .runScatterGatherIteration(new MinDistanceMessenger<K>(), new VertexDistanceUpdater<K>(), maxIterations).getVertices(); } @@ -74,12 +74,30 @@ public class SingleSourceShortestPaths<K> implements GraphAlgorithm<K, Double, D } /** + * Distributes the minimum distance associated with a given vertex among all + * the target vertices summed up with the edge's value. + * + * @param <K> + */ + public static final class MinDistanceMessenger<K> extends ScatterFunction<K, Double, Double, Double> { + + @Override + public void sendMessages(Vertex<K, Double> vertex) { + if (vertex.getValue() < Double.POSITIVE_INFINITY) { + for (Edge<K, Double> edge : getEdges()) { + sendMessageTo(edge.getTarget(), vertex.getValue() + edge.getValue()); + } + } + } + } + + /** * Function that updates the value of a vertex by picking the minimum * distance from all incoming messages. * * @param <K> */ - public static final class VertexDistanceUpdater<K> extends VertexUpdateFunction<K, Double, Double> { + public static final class VertexDistanceUpdater<K> extends GatherFunction<K, Double, Double> { @Override public void updateVertex(Vertex<K, Double> vertex, @@ -98,22 +116,4 @@ public class SingleSourceShortestPaths<K> implements GraphAlgorithm<K, Double, D } } } - - /** - * Distributes the minimum distance associated with a given vertex among all - * the target vertices summed up with the edge's value. - * - * @param <K> - */ - public static final class MinDistanceMessenger<K> extends MessagingFunction<K, Double, Double, Double> { - - @Override - public void sendMessages(Vertex<K, Double> vertex) { - if (vertex.getValue() < Double.POSITIVE_INFINITY) { - for (Edge<K, Double> edge : getEdges()) { - sendMessageTo(edge.getTarget(), vertex.getValue() + edge.getValue()); - } - } - } - } }
http://git-wip-us.apache.org/repos/asf/flink/blob/918e5d0c/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/library/TriangleEnumerator.java ---------------------------------------------------------------------- diff --git a/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/library/TriangleEnumerator.java b/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/library/TriangleEnumerator.java index 8272d8f..681d060 100644 --- a/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/library/TriangleEnumerator.java +++ b/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/library/TriangleEnumerator.java @@ -18,11 +18,11 @@ package org.apache.flink.graph.library; -import org.apache.flink.api.common.functions.MapFunction; -import org.apache.flink.api.common.functions.ReduceFunction; import org.apache.flink.api.common.functions.FlatMapFunction; -import org.apache.flink.api.common.functions.JoinFunction; import org.apache.flink.api.common.functions.GroupReduceFunction; +import org.apache.flink.api.common.functions.JoinFunction; +import org.apache.flink.api.common.functions.MapFunction; +import org.apache.flink.api.common.functions.ReduceFunction; import org.apache.flink.api.common.operators.Order; import org.apache.flink.api.common.operators.base.JoinOperatorBase.JoinHint; import org.apache.flink.api.java.DataSet; http://git-wip-us.apache.org/repos/asf/flink/blob/918e5d0c/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/spargel/GatherFunction.java ---------------------------------------------------------------------- diff --git a/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/spargel/GatherFunction.java b/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/spargel/GatherFunction.java new file mode 100644 index 0000000..d56c0da --- /dev/null +++ b/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/spargel/GatherFunction.java @@ -0,0 +1,251 @@ +/* + * 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 org.apache.flink.api.common.aggregators.Aggregator; +import org.apache.flink.api.common.functions.IterationRuntimeContext; +import org.apache.flink.api.java.tuple.Tuple3; +import org.apache.flink.graph.Vertex; +import org.apache.flink.types.Value; +import org.apache.flink.util.Collector; + +import java.io.Serializable; +import java.util.Collection; + +/** + * This class must be extended by functions that compute the state of the vertex depending on the old state and the + * incoming messages. The central method is {@link #updateVertex(Vertex, MessageIterator)}, which is + * invoked once per vertex per superstep. + * + * {@code <K>} The vertex key type. + * {@code <VV>} The vertex value type. + * {@code <Message>} The message type. + */ +public abstract class GatherFunction<K, VV, 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; + + /** + * Retrieves the number of vertices in the graph. + * @return the number of vertices if the {@link org.apache.flink.graph.IterationConfiguration#setOptNumVertices(boolean)} + * option has been set; -1 otherwise. + */ + public long getNumberOfVertices() { + return numberOfVertices; + } + + void setNumberOfVertices(long numberOfVertices) { + this.numberOfVertices = numberOfVertices; + } + + //--------------------------------------------------------------------------------------------- + + private boolean optDegrees; + + boolean isOptDegrees() { + return optDegrees; + } + + void setOptDegrees(boolean optDegrees) { + this.optDegrees = optDegrees; + } + + // -------------------------------------------------------------------------------------------- + // Public API Methods + // -------------------------------------------------------------------------------------------- + + /** + * This method is invoked once per vertex per superstep. It receives the current state of the vertex, as well as + * the incoming messages. It may set a new vertex state via {@link #setNewVertexValue(Object)}. If the vertex + * state is changed, it will trigger the sending of messages via the {@link ScatterFunction}. + * + * @param vertex The vertex. + * @param inMessages The incoming messages to this vertex. + * + * @throws Exception The computation may throw exceptions, which causes the superstep to fail. + */ + public abstract void updateVertex(Vertex<K, VV> vertex, MessageIterator<Message> inMessages) throws Exception; + + /** + * This method is executed once per superstep before the gather function is invoked for each vertex. + * + * @throws Exception Exceptions in the pre-superstep phase cause the superstep to fail. + */ + public void preSuperstep() throws Exception {} + + /** + * This method is executed once per superstep after the gather function has been invoked for each vertex. + * + * @throws Exception Exceptions in the post-superstep phase cause the superstep to fail. + */ + public void postSuperstep() throws Exception {} + + /** + * Sets the new value of this vertex. Setting a new value triggers the sending of outgoing messages from this vertex. + * + * This should be called at most once per updateVertex. + * + * @param newValue The new vertex value. + */ + public void setNewVertexValue(VV newValue) { + if(setNewVertexValueCalled) { + throw new IllegalStateException("setNewVertexValue should only be called at most once per updateVertex"); + } + setNewVertexValueCalled = true; + if(isOptDegrees()) { + outValWithDegrees.f1.f0 = newValue; + outWithDegrees.collect(outValWithDegrees); + } else { + outVal.setValue(newValue); + out.collect(outVal); + } + } + + /** + * Gets the number of the superstep, starting at <tt>1</tt>. + * + * @return The number of the current superstep. + */ + public int getSuperstepNumber() { + return this.runtimeContext.getSuperstepNumber(); + } + + /** + * Gets the iteration aggregator registered under the given name. The iteration aggregator combines + * all aggregates globally once per superstep and makes them available in the next superstep. + * + * @param name The name of the aggregator. + * @return The aggregator registered under this name, or null, if no aggregator was registered. + */ + public <T extends Aggregator<?>> T getIterationAggregator(String name) { + return this.runtimeContext.<T>getIterationAggregator(name); + } + + /** + * Get the aggregated value that an aggregator computed in the previous iteration. + * + * @param name The name of the aggregator. + * @return The aggregated value of the previous iteration. + */ + public <T extends Value> T getPreviousIterationAggregate(String name) { + return this.runtimeContext.<T>getPreviousIterationAggregate(name); + } + + /** + * 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.ScatterGatherConfiguration#addBroadcastSetForGatherFunction(String, org.apache.flink.api.java.DataSet)}. + * + * @param name The name under which the broadcast set is registered. + * @return The broadcast data set. + */ + public <T> Collection<T> getBroadcastSet(String name) { + return this.runtimeContext.<T>getBroadcastVariable(name); + } + + // -------------------------------------------------------------------------------------------- + // internal methods + // -------------------------------------------------------------------------------------------- + + private IterationRuntimeContext runtimeContext; + + private Collector<Vertex<K, VV>> out; + + private Collector<Vertex<K, Tuple3<VV, Long, Long>>> outWithDegrees; + + private Vertex<K, VV> outVal; + + private Vertex<K, Tuple3<VV, Long, Long>> outValWithDegrees; + + private long inDegree = -1; + + private long outDegree = -1; + + private boolean setNewVertexValueCalled; + + void init(IterationRuntimeContext context) { + this.runtimeContext = context; + } + + void setOutput(Vertex<K, VV> outVal, Collector<Vertex<K, VV>> out) { + this.outVal = outVal; + this.out = out; + setNewVertexValueCalled = false; + } + + @SuppressWarnings({ "unchecked", "rawtypes" }) + <ValueWithDegree> void setOutputWithDegrees(Vertex<K, ValueWithDegree> outVal, + Collector out) { + this.outValWithDegrees = (Vertex<K, Tuple3<VV, Long, Long>>) outVal; + this.outWithDegrees = out; + setNewVertexValueCalled = false; + } + + /** + * Retrieves the vertex in-degree (number of in-coming edges). + * @return The in-degree of this vertex + */ + public long getInDegree() { + return inDegree; + } + + void setInDegree(long inDegree) { + this.inDegree = inDegree; + } + + /** + * Retrieve the vertex out-degree (number of out-going edges). + * @return The out-degree of this vertex + */ + public long getOutDegree() { + return outDegree; + } + + void setOutDegree(long outDegree) { + this.outDegree = outDegree; + } + + /** + * 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.ScatterGatherIteration}. + * + * This function will retrieve the vertex from the vertexState and will set its degrees, afterwards calling + * the regular updateVertex function. + * + * @param vertexState + * @param inMessages + * @throws Exception + */ + @SuppressWarnings("unchecked") + <VertexWithDegree> void updateVertexFromScatterGatherIteration(Vertex<K, VertexWithDegree> vertexState, + MessageIterator<Message> inMessages) throws Exception { + + Vertex<K, VV> vertex = new Vertex<K, VV>(vertexState.f0, + ((Tuple3<VV, Long, Long>)vertexState.getValue()).f0); + + updateVertex(vertex, inMessages); + } +} http://git-wip-us.apache.org/repos/asf/flink/blob/918e5d0c/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/spargel/MessagingFunction.java ---------------------------------------------------------------------- diff --git a/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/spargel/MessagingFunction.java b/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/spargel/MessagingFunction.java deleted file mode 100644 index e12d779..0000000 --- a/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/spargel/MessagingFunction.java +++ /dev/null @@ -1,338 +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.io.Serializable; -import java.util.Collection; -import java.util.Iterator; - -import org.apache.flink.api.common.aggregators.Aggregator; -import org.apache.flink.api.common.functions.IterationRuntimeContext; -import org.apache.flink.api.java.tuple.Tuple; -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.Vertex; -import org.apache.flink.types.Value; -import org.apache.flink.util.Collector; - -/** - * The base class for functions that produce messages between vertices as a part of a {@link ScatterGatherIteration}. - * - * @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 abstract class MessagingFunction<K, VV, Message, EV> 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; - - /** - * Retrieves the number of vertices in the graph. - * @return the number of vertices if the {@link org.apache.flink.graph.IterationConfiguration#setOptNumVertices(boolean)} - * option has been set; -1 otherwise. - */ - public long getNumberOfVertices() { - 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 scatter gather iteration. - // -------------------------------------------------------------------------------------------- - - private EdgeDirection direction; - - /** - * Retrieves the edge direction in which messages are propagated in the scatter-gather iteration. - * @return the messaging {@link EdgeDirection} - */ - public EdgeDirection getDirection() { - return direction; - } - - void setDirection(EdgeDirection direction) { - this.direction = direction; - } - - // -------------------------------------------------------------------------------------------- - // Public API Methods - // -------------------------------------------------------------------------------------------- - - /** - * This method is invoked once per superstep for each vertex that was changed in that superstep. - * It needs to produce the messages that will be received by vertices in the next superstep. - * - * @param vertex The vertex that was changed. - * - * @throws Exception The computation may throw exceptions, which causes the superstep to fail. - */ - public abstract void sendMessages(Vertex<K, VV> vertex) throws Exception; - - /** - * This method is executed once per superstep before the vertex update function is invoked for each vertex. - * - * @throws Exception Exceptions in the pre-superstep phase cause the superstep to fail. - */ - public void preSuperstep() throws Exception {} - - /** - * This method is executed once per superstep after the vertex update function has been invoked for each vertex. - * - * @throws Exception Exceptions in the post-superstep phase cause the superstep to fail. - */ - public void postSuperstep() throws Exception {} - - - /** - * Gets an {@link java.lang.Iterable} with all edges. This method is mutually exclusive with - * {@link #sendMessageToAllNeighbors(Object)} and may be called only once. - * <p> - * If the {@link EdgeDirection} is OUT (default), then this iterator contains outgoing edges. - * If the {@link EdgeDirection} is IN, then this iterator contains incoming edges. - * If the {@link EdgeDirection} is ALL, then this iterator contains both outgoing and incoming edges. - * - * @return An iterator with all edges. - */ - @SuppressWarnings("unchecked") - public Iterable<Edge<K, EV>> getEdges() { - if (edgesUsed) { - throw new IllegalStateException("Can use either 'getEdges()' or 'sendMessageToAllNeighbors()' exactly once."); - } - edgesUsed = true; - this.edgeIterator.set((Iterator<Edge<K, EV>>) edges); - return this.edgeIterator; - } - - /** - * Sends the given message to all vertices that are targets of an edge of the changed vertex. - * This method is mutually exclusive to the method {@link #getEdges()} and may be called only once. - * <p> - * If the {@link EdgeDirection} is OUT (default), the message will be sent to out-neighbors. - * If the {@link EdgeDirection} is IN, the message will be sent to in-neighbors. - * If the {@link EdgeDirection} is ALL, the message will be sent to all neighbors. - * - * @param m The message to send. - */ - public void sendMessageToAllNeighbors(Message m) { - if (edgesUsed) { - throw new IllegalStateException("Can use either 'getEdges()' or 'sendMessageToAllNeighbors()'" - + "exactly once."); - } - - edgesUsed = true; - outValue.f1 = m; - - while (edges.hasNext()) { - Tuple next = (Tuple) edges.next(); - - /* - * When EdgeDirection is OUT, the edges iterator only has the out-edges - * of the vertex, i.e. the ones where this vertex is src. - * next.getField(1) gives the neighbor of the vertex running this MessagingFunction. - */ - if (getDirection().equals(EdgeDirection.OUT)) { - outValue.f0 = next.getField(1); - } - /* - * When EdgeDirection is IN, the edges iterator only has the in-edges - * of the vertex, i.e. the ones where this vertex is trg. - * next.getField(10) gives the neighbor of the vertex running this MessagingFunction. - */ - else if (getDirection().equals(EdgeDirection.IN)) { - outValue.f0 = next.getField(0); - } - // When EdgeDirection is ALL, the edges iterator contains both in- and out- edges - if (getDirection().equals(EdgeDirection.ALL)) { - if (next.getField(0).equals(vertexId)) { - // send msg to the trg - outValue.f0 = next.getField(1); - } - else { - // send msg to the src - outValue.f0 = next.getField(0); - } - } - out.collect(outValue); - } - } - - /** - * Sends the given message to the vertex identified by the given key. If the target vertex does not exist, - * the next superstep will cause an exception due to a non-deliverable message. - * - * @param target The key (id) of the target vertex to message. - * @param m The message. - */ - public void sendMessageTo(K target, Message m) { - outValue.f0 = target; - outValue.f1 = m; - out.collect(outValue); - } - - // -------------------------------------------------------------------------------------------- - - /** - * Gets the number of the superstep, starting at <tt>1</tt>. - * - * @return The number of the current superstep. - */ - public int getSuperstepNumber() { - return this.runtimeContext.getSuperstepNumber(); - } - - /** - * Gets the iteration aggregator registered under the given name. The iteration aggregator combines - * all aggregates globally once per superstep and makes them available in the next superstep. - * - * @param name The name of the aggregator. - * @return The aggregator registered under this name, or null, if no aggregator was registered. - */ - public <T extends Aggregator<?>> T getIterationAggregator(String name) { - return this.runtimeContext.<T>getIterationAggregator(name); - } - - /** - * Get the aggregated value that an aggregator computed in the previous iteration. - * - * @param name The name of the aggregator. - * @return The aggregated value of the previous iteration. - */ - public <T extends Value> T getPreviousIterationAggregate(String name) { - return this.runtimeContext.<T>getPreviousIterationAggregate(name); - } - - /** - * 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.ScatterGatherConfiguration#addBroadcastSetForMessagingFunction(String, org.apache.flink.api.java.DataSet)}. - * - * @param name The name under which the broadcast set is registered. - * @return The broadcast data set. - */ - public <T> Collection<T> getBroadcastSet(String name) { - return this.runtimeContext.<T>getBroadcastVariable(name); - } - - // -------------------------------------------------------------------------------------------- - // internal methods and state - // -------------------------------------------------------------------------------------------- - - private Tuple2<K, Message> outValue; - - private IterationRuntimeContext runtimeContext; - - private Iterator<?> edges; - - private Collector<Tuple2<K, Message>> out; - - private K vertexId; - - private EdgesIterator<K, EV> edgeIterator; - - private boolean edgesUsed; - - private long inDegree = -1; - - private long outDegree = -1; - - void init(IterationRuntimeContext context) { - this.runtimeContext = context; - this.outValue = new Tuple2<K, Message>(); - this.edgeIterator = new EdgesIterator<K, EV>(); - } - - void set(Iterator<?> edges, Collector<Tuple2<K, Message>> out, K id) { - this.edges = edges; - this.out = out; - this.vertexId = id; - this.edgesUsed = false; - } - - private static final class EdgesIterator<K, EV> - implements Iterator<Edge<K, EV>>, Iterable<Edge<K, EV>> - { - private Iterator<Edge<K, EV>> input; - - private Edge<K, EV> edge = new Edge<K, EV>(); - - void set(Iterator<Edge<K, EV>> input) { - this.input = input; - } - - @Override - public boolean hasNext() { - return input.hasNext(); - } - - @Override - public Edge<K, EV> next() { - Edge<K, EV> next = input.next(); - edge.setSource(next.f0); - edge.setTarget(next.f1); - edge.setValue(next.f2); - return edge; - } - - @Override - public void remove() { - throw new UnsupportedOperationException(); - } - @Override - public Iterator<Edge<K, EV>> iterator() { - return this; - } - } - - /** - * Retrieves the vertex in-degree (number of in-coming edges). - * @return The in-degree of this vertex - */ - public long getInDegree() { - return inDegree; - } - - void setInDegree(long inDegree) { - this.inDegree = inDegree; - } - - /** - * Retrieve the vertex out-degree (number of out-going edges). - * @return The out-degree of this vertex - */ - public long getOutDegree() { - return outDegree; - } - - void setOutDegree(long outDegree) { - this.outDegree = outDegree; - } -} http://git-wip-us.apache.org/repos/asf/flink/blob/918e5d0c/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/spargel/ScatterFunction.java ---------------------------------------------------------------------- diff --git a/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/spargel/ScatterFunction.java b/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/spargel/ScatterFunction.java new file mode 100644 index 0000000..336e73d --- /dev/null +++ b/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/spargel/ScatterFunction.java @@ -0,0 +1,338 @@ +/* + * 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 org.apache.flink.api.common.aggregators.Aggregator; +import org.apache.flink.api.common.functions.IterationRuntimeContext; +import org.apache.flink.api.java.tuple.Tuple; +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.Vertex; +import org.apache.flink.types.Value; +import org.apache.flink.util.Collector; + +import java.io.Serializable; +import java.util.Collection; +import java.util.Iterator; + +/** + * The base class for functions that produce messages between vertices as a part of a {@link ScatterGatherIteration}. + * + * @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 abstract class ScatterFunction<K, VV, Message, EV> 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; + + /** + * Retrieves the number of vertices in the graph. + * @return the number of vertices if the {@link org.apache.flink.graph.IterationConfiguration#setOptNumVertices(boolean)} + * option has been set; -1 otherwise. + */ + public long getNumberOfVertices() { + 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 scatter gather iteration. + // -------------------------------------------------------------------------------------------- + + private EdgeDirection direction; + + /** + * Retrieves the edge direction in which messages are propagated in the scatter-gather iteration. + * @return the messaging {@link EdgeDirection} + */ + public EdgeDirection getDirection() { + return direction; + } + + void setDirection(EdgeDirection direction) { + this.direction = direction; + } + + // -------------------------------------------------------------------------------------------- + // Public API Methods + // -------------------------------------------------------------------------------------------- + + /** + * This method is invoked once per superstep for each vertex that was changed in that superstep. + * It needs to produce the messages that will be received by vertices in the next superstep. + * + * @param vertex The vertex that was changed. + * + * @throws Exception The computation may throw exceptions, which causes the superstep to fail. + */ + public abstract void sendMessages(Vertex<K, VV> vertex) throws Exception; + + /** + * This method is executed once per superstep before the scatter function is invoked for each vertex. + * + * @throws Exception Exceptions in the pre-superstep phase cause the superstep to fail. + */ + public void preSuperstep() throws Exception {} + + /** + * This method is executed once per superstep after the scatter function has been invoked for each vertex. + * + * @throws Exception Exceptions in the post-superstep phase cause the superstep to fail. + */ + public void postSuperstep() throws Exception {} + + + /** + * Gets an {@link java.lang.Iterable} with all edges. This method is mutually exclusive with + * {@link #sendMessageToAllNeighbors(Object)} and may be called only once. + * <p> + * If the {@link EdgeDirection} is OUT (default), then this iterator contains outgoing edges. + * If the {@link EdgeDirection} is IN, then this iterator contains incoming edges. + * If the {@link EdgeDirection} is ALL, then this iterator contains both outgoing and incoming edges. + * + * @return An iterator with all edges. + */ + @SuppressWarnings("unchecked") + public Iterable<Edge<K, EV>> getEdges() { + if (edgesUsed) { + throw new IllegalStateException("Can use either 'getEdges()' or 'sendMessageToAllNeighbors()' exactly once."); + } + edgesUsed = true; + this.edgeIterator.set((Iterator<Edge<K, EV>>) edges); + return this.edgeIterator; + } + + /** + * Sends the given message to all vertices that are targets of an edge of the changed vertex. + * This method is mutually exclusive to the method {@link #getEdges()} and may be called only once. + * <p> + * If the {@link EdgeDirection} is OUT (default), the message will be sent to out-neighbors. + * If the {@link EdgeDirection} is IN, the message will be sent to in-neighbors. + * If the {@link EdgeDirection} is ALL, the message will be sent to all neighbors. + * + * @param m The message to send. + */ + public void sendMessageToAllNeighbors(Message m) { + if (edgesUsed) { + throw new IllegalStateException("Can use either 'getEdges()' or 'sendMessageToAllNeighbors()'" + + "exactly once."); + } + + edgesUsed = true; + outValue.f1 = m; + + while (edges.hasNext()) { + Tuple next = (Tuple) edges.next(); + + /* + * When EdgeDirection is OUT, the edges iterator only has the out-edges + * of the vertex, i.e. the ones where this vertex is src. + * next.getField(1) gives the neighbor of the vertex running this ScatterFunction. + */ + if (getDirection().equals(EdgeDirection.OUT)) { + outValue.f0 = next.getField(1); + } + /* + * When EdgeDirection is IN, the edges iterator only has the in-edges + * of the vertex, i.e. the ones where this vertex is trg. + * next.getField(10) gives the neighbor of the vertex running this ScatterFunction. + */ + else if (getDirection().equals(EdgeDirection.IN)) { + outValue.f0 = next.getField(0); + } + // When EdgeDirection is ALL, the edges iterator contains both in- and out- edges + if (getDirection().equals(EdgeDirection.ALL)) { + if (next.getField(0).equals(vertexId)) { + // send msg to the trg + outValue.f0 = next.getField(1); + } + else { + // send msg to the src + outValue.f0 = next.getField(0); + } + } + out.collect(outValue); + } + } + + /** + * Sends the given message to the vertex identified by the given key. If the target vertex does not exist, + * the next superstep will cause an exception due to a non-deliverable message. + * + * @param target The key (id) of the target vertex to message. + * @param m The message. + */ + public void sendMessageTo(K target, Message m) { + outValue.f0 = target; + outValue.f1 = m; + out.collect(outValue); + } + + // -------------------------------------------------------------------------------------------- + + /** + * Gets the number of the superstep, starting at <tt>1</tt>. + * + * @return The number of the current superstep. + */ + public int getSuperstepNumber() { + return this.runtimeContext.getSuperstepNumber(); + } + + /** + * Gets the iteration aggregator registered under the given name. The iteration aggregator combines + * all aggregates globally once per superstep and makes them available in the next superstep. + * + * @param name The name of the aggregator. + * @return The aggregator registered under this name, or null, if no aggregator was registered. + */ + public <T extends Aggregator<?>> T getIterationAggregator(String name) { + return this.runtimeContext.<T>getIterationAggregator(name); + } + + /** + * Get the aggregated value that an aggregator computed in the previous iteration. + * + * @param name The name of the aggregator. + * @return The aggregated value of the previous iteration. + */ + public <T extends Value> T getPreviousIterationAggregate(String name) { + return this.runtimeContext.<T>getPreviousIterationAggregate(name); + } + + /** + * 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.ScatterGatherConfiguration#addBroadcastSetForScatterFunction(String, org.apache.flink.api.java.DataSet)}. + * + * @param name The name under which the broadcast set is registered. + * @return The broadcast data set. + */ + public <T> Collection<T> getBroadcastSet(String name) { + return this.runtimeContext.<T>getBroadcastVariable(name); + } + + // -------------------------------------------------------------------------------------------- + // internal methods and state + // -------------------------------------------------------------------------------------------- + + private Tuple2<K, Message> outValue; + + private IterationRuntimeContext runtimeContext; + + private Iterator<?> edges; + + private Collector<Tuple2<K, Message>> out; + + private K vertexId; + + private EdgesIterator<K, EV> edgeIterator; + + private boolean edgesUsed; + + private long inDegree = -1; + + private long outDegree = -1; + + void init(IterationRuntimeContext context) { + this.runtimeContext = context; + this.outValue = new Tuple2<K, Message>(); + this.edgeIterator = new EdgesIterator<K, EV>(); + } + + void set(Iterator<?> edges, Collector<Tuple2<K, Message>> out, K id) { + this.edges = edges; + this.out = out; + this.vertexId = id; + this.edgesUsed = false; + } + + private static final class EdgesIterator<K, EV> + implements Iterator<Edge<K, EV>>, Iterable<Edge<K, EV>> + { + private Iterator<Edge<K, EV>> input; + + private Edge<K, EV> edge = new Edge<K, EV>(); + + void set(Iterator<Edge<K, EV>> input) { + this.input = input; + } + + @Override + public boolean hasNext() { + return input.hasNext(); + } + + @Override + public Edge<K, EV> next() { + Edge<K, EV> next = input.next(); + edge.setSource(next.f0); + edge.setTarget(next.f1); + edge.setValue(next.f2); + return edge; + } + + @Override + public void remove() { + throw new UnsupportedOperationException(); + } + @Override + public Iterator<Edge<K, EV>> iterator() { + return this; + } + } + + /** + * Retrieves the vertex in-degree (number of in-coming edges). + * @return The in-degree of this vertex + */ + public long getInDegree() { + return inDegree; + } + + void setInDegree(long inDegree) { + this.inDegree = inDegree; + } + + /** + * Retrieve the vertex out-degree (number of out-going edges). + * @return The out-degree of this vertex + */ + public long getOutDegree() { + return outDegree; + } + + void setOutDegree(long outDegree) { + this.outDegree = outDegree; + } +} http://git-wip-us.apache.org/repos/asf/flink/blob/918e5d0c/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/spargel/ScatterGatherConfiguration.java ---------------------------------------------------------------------- diff --git a/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/spargel/ScatterGatherConfiguration.java b/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/spargel/ScatterGatherConfiguration.java index 3a3de64..4ac1ae1 100644 --- a/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/spargel/ScatterGatherConfiguration.java +++ b/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/spargel/ScatterGatherConfiguration.java @@ -29,20 +29,20 @@ import java.util.List; /** * A ScatterGatherConfiguration object can be used to set the iteration name and * degree of parallelism, to register aggregators and use broadcast sets in - * the {@link org.apache.flink.graph.spargel.VertexUpdateFunction} and {@link org.apache.flink.graph.spargel.MessagingFunction} + * the {@link GatherFunction} and {@link ScatterFunction} * * The VertexCentricConfiguration object is passed as an argument to * {@link org.apache.flink.graph.Graph#runScatterGatherIteration ( - * org.apache.flink.graph.spargel.VertexUpdateFunction, org.apache.flink.graph.spargel.MessagingFunction, int, + * org.apache.flink.graph.spargel.GatherFunction, org.apache.flink.graph.spargel.ScatterFunction, int, * ScatterGatherConfiguration)}. */ public class ScatterGatherConfiguration extends IterationConfiguration { - /** the broadcast variables for the update function **/ - private List<Tuple2<String, DataSet<?>>> bcVarsUpdate = new ArrayList<Tuple2<String,DataSet<?>>>(); + /** the broadcast variables for the scatter function **/ + private List<Tuple2<String, DataSet<?>>> bcVarsScatter = new ArrayList<>(); - /** the broadcast variables for the messaging function **/ - private List<Tuple2<String, DataSet<?>>> bcVarsMessaging = new ArrayList<Tuple2<String,DataSet<?>>>(); + /** the broadcast variables for the gather function **/ + private List<Tuple2<String, DataSet<?>>> bcVarsGather = new ArrayList<>(); /** flag that defines whether the degrees option is set **/ private boolean optDegrees = false; @@ -53,43 +53,43 @@ public class ScatterGatherConfiguration extends IterationConfiguration { public ScatterGatherConfiguration() {} /** - * Adds a data set as a broadcast set to the messaging function. + * Adds a data set as a broadcast set to the scatter function. * - * @param name The name under which the broadcast data is available in the messaging function. + * @param name The name under which the broadcast data is available in the scatter function. * @param data The data set to be broadcasted. */ - public void addBroadcastSetForMessagingFunction(String name, DataSet<?> data) { - this.bcVarsMessaging.add(new Tuple2<String, DataSet<?>>(name, data)); + public void addBroadcastSetForScatterFunction(String name, DataSet<?> data) { + this.bcVarsScatter.add(new Tuple2<String, DataSet<?>>(name, data)); } /** - * Adds a data set as a broadcast set to the vertex update function. + * Adds a data set as a broadcast set to the gather function. * - * @param name The name under which the broadcast data is available in the vertex update function. + * @param name The name under which the broadcast data is available in the gather function. * @param data The data set to be broadcasted. */ - public void addBroadcastSetForUpdateFunction(String name, DataSet<?> data) { - this.bcVarsUpdate.add(new Tuple2<String, DataSet<?>>(name, data)); + public void addBroadcastSetForGatherFunction(String name, DataSet<?> data) { + this.bcVarsGather.add(new Tuple2<String, DataSet<?>>(name, data)); } /** - * Get the broadcast variables of the VertexUpdateFunction. + * Get the broadcast variables of the ScatterFunction. * * @return a List of Tuple2, where the first field is the broadcast variable name * and the second field is the broadcast DataSet. */ - public List<Tuple2<String, DataSet<?>>> getUpdateBcastVars() { - return this.bcVarsUpdate; + public List<Tuple2<String, DataSet<?>>> getScatterBcastVars() { + return this.bcVarsScatter; } /** - * Get the broadcast variables of the MessagingFunction. + * Get the broadcast variables of the GatherFunction. * * @return a List of Tuple2, where the first field is the broadcast variable name * and the second field is the broadcast DataSet. */ - public List<Tuple2<String, DataSet<?>>> getMessagingBcastVars() { - return this.bcVarsMessaging; + public List<Tuple2<String, DataSet<?>>> getGatherBcastVars() { + return this.bcVarsGather; } /** @@ -113,7 +113,7 @@ public class ScatterGatherConfiguration extends IterationConfiguration { } /** - * Gets the direction in which messages are sent in the MessagingFunction. + * Gets the direction in which messages are sent in the ScatterFunction. * By default the messaging direction is OUT. * * @return an EdgeDirection, which can be either IN, OUT or ALL. @@ -123,7 +123,7 @@ public class ScatterGatherConfiguration extends IterationConfiguration { } /** - * Sets the direction in which messages are sent in the MessagingFunction. + * Sets the direction in which messages are sent in the ScatterFunction. * By default the messaging direction is OUT. * * @param direction - IN, OUT or ALL http://git-wip-us.apache.org/repos/asf/flink/blob/918e5d0c/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/spargel/ScatterGatherIteration.java ---------------------------------------------------------------------- diff --git a/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/spargel/ScatterGatherIteration.java b/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/spargel/ScatterGatherIteration.java index fc5c210..fde305f 100644 --- a/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/spargel/ScatterGatherIteration.java +++ b/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/spargel/ScatterGatherIteration.java @@ -53,21 +53,21 @@ import java.util.Map; * Scatter-Gather 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. + * The computation terminates once no vertex updates its 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 + * <li>The {@link GatherFunction} 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 + * <li>The {@link ScatterFunction} 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> * * Scatter-Gather graph iterations are are run by calling - * {@link Graph#runScatterGatherIteration(VertexUpdateFunction, MessagingFunction, int)}. + * {@link Graph#runScatterGatherIteration(ScatterFunction, GatherFunction, 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). @@ -77,47 +77,47 @@ import java.util.Map; public class ScatterGatherIteration<K, VV, Message, EV> implements CustomUnaryOperation<Vertex<K, VV>, Vertex<K, VV>> { - private final VertexUpdateFunction<K, VV, Message> updateFunction; + private final ScatterFunction<K, VV, Message, EV> scatterFunction; + + private final GatherFunction<K, VV, Message> gatherFunction; - 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 ScatterGatherConfiguration configuration; // ---------------------------------------------------------------------------------- - - private ScatterGatherIteration(VertexUpdateFunction<K, VV, Message> uf, - MessagingFunction<K, VV, Message, EV> mf, + + private ScatterGatherIteration(ScatterFunction<K, VV, Message, EV> sf, + GatherFunction<K, VV, Message> gf, DataSet<Edge<K, EV>> edgesWithValue, int maximumNumberOfIterations) { - Preconditions.checkNotNull(uf); - Preconditions.checkNotNull(mf); + Preconditions.checkNotNull(sf); + Preconditions.checkNotNull(gf); Preconditions.checkNotNull(edgesWithValue); Preconditions.checkArgument(maximumNumberOfIterations > 0, "The maximum number of iterations must be at least one."); - this.updateFunction = uf; - this.messagingFunction = mf; + this.scatterFunction = sf; + this.gatherFunction = gf; this.edgesWithValue = edgesWithValue; - this.maximumNumberOfIterations = maximumNumberOfIterations; - this.messageType = getMessageType(mf); + this.maximumNumberOfIterations = maximumNumberOfIterations; + this.messageType = getMessageType(sf); } - - private TypeInformation<Message> getMessageType(MessagingFunction<K, VV, Message, EV> mf) { - return TypeExtractor.createTypeInfo(mf, MessagingFunction.class, mf.getClass(), 2); + + private TypeInformation<Message> getMessageType(ScatterFunction<K, VV, Message, EV> mf) { + return TypeExtractor.createTypeInfo(mf, ScatterFunction.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. @@ -131,7 +131,7 @@ public class ScatterGatherIteration<K, VV, Message, EV> public void setInput(DataSet<Vertex<K, VV>> inputData) { this.initialVertices = inputData; } - + /** * Creates the operator that represents this scatter-gather graph computation. * @@ -145,14 +145,14 @@ public class ScatterGatherIteration<K, VV, Message, EV> // prepare some type information TypeInformation<K> keyType = ((TupleTypeInfo<?>) initialVertices.getType()).getTypeAt(0); - TypeInformation<Tuple2<K, Message>> messageTypeInfo = new TupleTypeInfo<Tuple2<K,Message>>(keyType, messageType); + TypeInformation<Tuple2<K, Message>> messageTypeInfo = new TupleTypeInfo<>(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 + // and set it within the scatter and gather functions DataSet<LongValue> numberOfVertices = null; if (this.configuration != null && this.configuration.isOptNumVertices()) { @@ -164,13 +164,13 @@ public class ScatterGatherIteration<K, VV, Message, EV> } if(this.configuration != null) { - messagingFunction.setDirection(this.configuration.getDirection()); + scatterFunction.setDirection(this.configuration.getDirection()); } else { - messagingFunction.setDirection(EdgeDirection.OUT); + scatterFunction.setDirection(EdgeDirection.OUT); } // retrieve the direction in which the updates are made and in which the messages are sent - EdgeDirection messagingDirection = messagingFunction.getDirection(); + EdgeDirection messagingDirection = scatterFunction.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 @@ -186,9 +186,9 @@ public class ScatterGatherIteration<K, VV, Message, EV> * 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 sf The function that turns changed vertex states into messages along the edges. + * @param gf The function that updates the state of the vertices from the incoming messages. + * * @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. @@ -196,14 +196,11 @@ public class ScatterGatherIteration<K, VV, Message, EV> * * @return An in stance of the scatter-gather graph computation operator. */ - public static final <K, VV, Message, EV> - ScatterGatherIteration<K, VV, Message, EV> withEdges( - DataSet<Edge<K, EV>> edgesWithValue, - VertexUpdateFunction<K, VV, Message> uf, - MessagingFunction<K, VV, Message, EV> mf, - int maximumNumberOfIterations) + public static final <K, VV, Message, EV> ScatterGatherIteration<K, VV, Message, EV> withEdges( + DataSet<Edge<K, EV>> edgesWithValue, ScatterFunction<K, VV, Message, EV> sf, + GatherFunction<K, VV, Message> gf, int maximumNumberOfIterations) { - return new ScatterGatherIteration<K, VV, Message, EV>(uf, mf, edgesWithValue, maximumNumberOfIterations); + return new ScatterGatherIteration<>(sf, gf, edgesWithValue, maximumNumberOfIterations); } /** @@ -226,23 +223,122 @@ public class ScatterGatherIteration<K, VV, Message, EV> // Wrapping UDFs // -------------------------------------------------------------------------------------------- - private static abstract class VertexUpdateUdf<K, VVWithDegrees, Message> extends RichCoGroupFunction< + /* + * UDF that encapsulates the message sending function for graphs where the edges have an associated value. + */ + private static abstract class ScatterUdfWithEdgeValues<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 ScatterFunction<K, VV, Message, EV> scatterFunction; + + private transient TypeInformation<Tuple2<K, Message>> resultType; + + + private ScatterUdfWithEdgeValues(ScatterFunction<K, VV, Message, EV> scatterFunction, + TypeInformation<Tuple2<K, Message>> resultType) + { + this.scatterFunction = scatterFunction; + this.resultType = resultType; + } + + @Override + public void open(Configuration parameters) throws Exception { + if (getRuntimeContext().hasBroadcastVariable("number of vertices")) { + Collection<LongValue> numberOfVertices = getRuntimeContext().getBroadcastVariable("number of vertices"); + this.scatterFunction.setNumberOfVertices(numberOfVertices.iterator().next().getValue()); + } + if (getIterationRuntimeContext().getSuperstepNumber() == 1) { + this.scatterFunction.init(getIterationRuntimeContext()); + } + this.scatterFunction.preSuperstep(); + } + + @Override + public void close() throws Exception { + this.scatterFunction.postSuperstep(); + } + + @Override + public TypeInformation<Tuple2<K, Message>> getProducedType() { + return this.resultType; + } + } + + @SuppressWarnings("serial") + private static final class ScatterUdfWithEVsSimpleVV<K, VV, Message, EV> + extends ScatterUdfWithEdgeValues<K, VV, VV, Message, EV> { + + private ScatterUdfWithEVsSimpleVV(ScatterFunction<K, VV, Message, EV> scatterFunction, + TypeInformation<Tuple2<K, Message>> resultType) { + super(scatterFunction, 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(); + scatterFunction.set(edges.iterator(), out, newVertexState.getId()); + scatterFunction.sendMessages(newVertexState); + } + } + } + + @SuppressWarnings("serial") + private static final class ScatterUdfWithEVsVVWithDegrees<K, VV, Message, EV> + extends ScatterUdfWithEdgeValues<K, Tuple3<VV, LongValue, LongValue>, VV, Message, EV> { + + private Vertex<K, VV> nextVertex = new Vertex<>(); + + private ScatterUdfWithEVsVVWithDegrees(ScatterFunction<K, VV, Message, EV> scatterFunction, + TypeInformation<Tuple2<K, Message>> resultType) { + super(scatterFunction, resultType); + } + + @Override + public void coGroup(Iterable<Edge<K, EV>> edges, Iterable<Vertex<K, Tuple3<VV, LongValue, LongValue>>> state, + Collector<Tuple2<K, Message>> out) throws Exception { + + final Iterator<Vertex<K, Tuple3<VV, LongValue, LongValue>>> stateIter = state.iterator(); + + if (stateIter.hasNext()) { + Vertex<K, Tuple3<VV, LongValue, LongValue>> vertexWithDegrees = stateIter.next(); + + nextVertex.setField(vertexWithDegrees.f0, 0); + nextVertex.setField(vertexWithDegrees.f1.f0, 1); + + scatterFunction.setInDegree(vertexWithDegrees.f1.f1.getValue()); + scatterFunction.setOutDegree(vertexWithDegrees.f1.f2.getValue()); + + scatterFunction.set(edges.iterator(), out, vertexWithDegrees.getId()); + scatterFunction.sendMessages(nextVertex); + } + } + } + + private static abstract class GatherUdf<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>(); - + final GatherFunction<K, VVWithDegrees, Message> gatherFunction; + + final MessageIterator<Message> messageIter = new MessageIterator<>(); + private transient TypeInformation<Vertex<K, VVWithDegrees>> resultType; - - - private VertexUpdateUdf(VertexUpdateFunction<K, VVWithDegrees, Message> vertexUpdateFunction, + + + private GatherUdf(GatherFunction<K, VVWithDegrees, Message> gatherFunction, TypeInformation<Vertex<K, VVWithDegrees>> resultType) { - this.vertexUpdateFunction = vertexUpdateFunction; + this.gatherFunction = gatherFunction; this.resultType = resultType; } @@ -250,17 +346,17 @@ public class ScatterGatherIteration<K, VV, Message, EV> public void open(Configuration parameters) throws Exception { if (getRuntimeContext().hasBroadcastVariable("number of vertices")) { Collection<LongValue> numberOfVertices = getRuntimeContext().getBroadcastVariable("number of vertices"); - this.vertexUpdateFunction.setNumberOfVertices(numberOfVertices.iterator().next().getValue()); + this.gatherFunction.setNumberOfVertices(numberOfVertices.iterator().next().getValue()); } if (getIterationRuntimeContext().getSuperstepNumber() == 1) { - this.vertexUpdateFunction.init(getIterationRuntimeContext()); + this.gatherFunction.init(getIterationRuntimeContext()); } - this.vertexUpdateFunction.preSuperstep(); + this.gatherFunction.preSuperstep(); } - + @Override public void close() throws Exception { - this.vertexUpdateFunction.postSuperstep(); + this.gatherFunction.postSuperstep(); } @Override @@ -270,10 +366,10 @@ public class ScatterGatherIteration<K, VV, Message, EV> } @SuppressWarnings("serial") - private static final class VertexUpdateUdfSimpleVV<K, VV, Message> extends VertexUpdateUdf<K, VV, Message> { + private static final class GatherUdfSimpleVV<K, VV, Message> extends GatherUdf<K, VV, Message> { - private VertexUpdateUdfSimpleVV(VertexUpdateFunction<K, VV, Message> vertexUpdateFunction, TypeInformation<Vertex<K, VV>> resultType) { - super(vertexUpdateFunction, resultType); + private GatherUdfSimpleVV(GatherFunction<K, VV, Message> gatherFunction, TypeInformation<Vertex<K, VV>> resultType) { + super(gatherFunction, resultType); } @Override @@ -289,8 +385,8 @@ public class ScatterGatherIteration<K, VV, Message, EV> Iterator<Tuple2<?, Message>> downcastIter = (Iterator<Tuple2<?, Message>>) (Iterator<?>) messages.iterator(); messageIter.setSource(downcastIter); - vertexUpdateFunction.setOutput(vertexState, out); - vertexUpdateFunction.updateVertex(vertexState, messageIter); + gatherFunction.setOutput(vertexState, out); + gatherFunction.updateVertex(vertexState, messageIter); } else { final Iterator<Tuple2<K, Message>> messageIter = messages.iterator(); @@ -299,7 +395,7 @@ public class ScatterGatherIteration<K, VV, Message, EV> try { Tuple2<K, Message> next = messageIter.next(); message = "Target vertex '" + next.f0 + "' does not exist!."; - } catch (Throwable t) {} + } catch (Throwable ignored) {} throw new Exception(message); } else { throw new Exception(); @@ -309,31 +405,31 @@ public class ScatterGatherIteration<K, VV, Message, EV> } @SuppressWarnings("serial") - private static final class VertexUpdateUdfVVWithDegrees<K, VV, Message> extends VertexUpdateUdf<K, Tuple3<VV, LongValue, LongValue>, Message> { + private static final class GatherUdfVVWithDegrees<K, VV, Message> extends GatherUdf<K, Tuple3<VV, LongValue, LongValue>, Message> { - private VertexUpdateUdfVVWithDegrees(VertexUpdateFunction<K, Tuple3<VV, LongValue, LongValue>, Message> vertexUpdateFunction, + private GatherUdfVVWithDegrees(GatherFunction<K, Tuple3<VV, LongValue, LongValue>, Message> gatherFunction, TypeInformation<Vertex<K, Tuple3<VV, LongValue, LongValue>>> resultType) { - super(vertexUpdateFunction, resultType); + super(gatherFunction, resultType); } - + @Override public void coGroup(Iterable<Tuple2<K, Message>> messages, Iterable<Vertex<K, Tuple3<VV, LongValue, LongValue>>> vertex, Collector<Vertex<K, Tuple3<VV, LongValue, LongValue>>> out) throws Exception { final Iterator<Vertex<K, Tuple3<VV, LongValue, LongValue>>> vertexIter = vertex.iterator(); - + if (vertexIter.hasNext()) { Vertex<K, Tuple3<VV, LongValue, LongValue>> vertexWithDegrees = vertexIter.next(); - + @SuppressWarnings("unchecked") Iterator<Tuple2<?, Message>> downcastIter = (Iterator<Tuple2<?, Message>>) (Iterator<?>) messages.iterator(); messageIter.setSource(downcastIter); - vertexUpdateFunction.setInDegree(vertexWithDegrees.f1.f1.getValue()); - vertexUpdateFunction.setOutDegree(vertexWithDegrees.f1.f2.getValue()); + gatherFunction.setInDegree(vertexWithDegrees.f1.f1.getValue()); + gatherFunction.setOutDegree(vertexWithDegrees.f1.f2.getValue()); - vertexUpdateFunction.setOutputWithDegrees(vertexWithDegrees, out); - vertexUpdateFunction.updateVertexFromScatterGatherIteration(vertexWithDegrees, messageIter); + gatherFunction.setOutputWithDegrees(vertexWithDegrees, out); + gatherFunction.updateVertexFromScatterGatherIteration(vertexWithDegrees, messageIter); } else { final Iterator<Tuple2<K, Message>> messageIter = messages.iterator(); @@ -342,7 +438,7 @@ public class ScatterGatherIteration<K, VV, Message, EV> try { Tuple2<K, Message> next = messageIter.next(); message = "Target vertex '" + next.f0 + "' does not exist!."; - } catch (Throwable t) {} + } catch (Throwable ignored) {} throw new Exception(message); } else { throw new Exception(); @@ -351,112 +447,12 @@ public class ScatterGatherIteration<K, VV, Message, EV> } } - /* - * 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 (getRuntimeContext().hasBroadcastVariable("number of vertices")) { - Collection<LongValue> numberOfVertices = getRuntimeContext().getBroadcastVariable("number of vertices"); - this.messagingFunction.setNumberOfVertices(numberOfVertices.iterator().next().getValue()); - } - 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, LongValue, LongValue>, 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, LongValue, LongValue>>> state, - Collector<Tuple2<K, Message>> out) throws Exception { - - final Iterator<Vertex<K, Tuple3<VV, LongValue, LongValue>>> stateIter = state.iterator(); - - if (stateIter.hasNext()) { - Vertex<K, Tuple3<VV, LongValue, LongValue>> vertexWithDegrees = stateIter.next(); - - nextVertex.setField(vertexWithDegrees.f0, 0); - nextVertex.setField(vertexWithDegrees.f1.f0, 1); - - messagingFunction.setInDegree(vertexWithDegrees.f1.f1.getValue()); - messagingFunction.setOutDegree(vertexWithDegrees.f1.f2.getValue()); - - 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 + * Method that builds the scatter function using a coGroup operator for a simple vertex (without * degrees). * It afterwards configures the function with a custom name and broadcast variables. * @@ -464,17 +460,17 @@ public class ScatterGatherIteration<K, VV, Message, EV> * @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 + * @return the scatter function */ - private CoGroupOperator<?, ?, Tuple2<K, Message>> buildMessagingFunction( + private CoGroupOperator<?, ?, Tuple2<K, Message>> buildScatterFunction( DeltaIteration<Vertex<K, VV>, Vertex<K, VV>> iteration, TypeInformation<Tuple2<K, Message>> messageTypeInfo, int whereArg, int equalToArg, DataSet<LongValue> numberOfVertices) { - // build the messaging function (co group) + // build the scatter function (co group) CoGroupOperator<?, ?, Tuple2<K, Message>> messages; - MessagingUdfWithEdgeValues<K, VV, VV, Message, EV> messenger = - new MessagingUdfWithEVsSimpleVV<K, VV, Message, EV>(messagingFunction, messageTypeInfo); + ScatterUdfWithEdgeValues<K, VV, VV, Message, EV> messenger = + new ScatterUdfWithEVsSimpleVV<>(scatterFunction, messageTypeInfo); messages = this.edgesWithValue.coGroup(iteration.getWorkset()).where(whereArg) .equalTo(equalToArg).with(messenger); @@ -482,7 +478,7 @@ public class ScatterGatherIteration<K, VV, Message, EV> // 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()) { + for (Tuple2<String, DataSet<?>> e : this.configuration.getScatterBcastVars()) { messages = messages.withBroadcastSet(e.f1, e.f0); } if (this.configuration.isOptNumVertices()) { @@ -494,7 +490,7 @@ public class ScatterGatherIteration<K, VV, Message, EV> } /** - * Method that builds the messaging function using a coGroup operator for a vertex + * Method that builds the scatter function using a coGroup operator for a vertex * containing degree information. * It afterwards configures the function with a custom name and broadcast variables. * @@ -502,17 +498,17 @@ public class ScatterGatherIteration<K, VV, Message, EV> * @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 + * @return the scatter function */ - private CoGroupOperator<?, ?, Tuple2<K, Message>> buildMessagingFunctionVerticesWithDegrees( + private CoGroupOperator<?, ?, Tuple2<K, Message>> buildScatterFunctionVerticesWithDegrees( DeltaIteration<Vertex<K, Tuple3<VV, LongValue, LongValue>>, Vertex<K, Tuple3<VV, LongValue, LongValue>>> iteration, TypeInformation<Tuple2<K, Message>> messageTypeInfo, int whereArg, int equalToArg, DataSet<LongValue> numberOfVertices) { - // build the messaging function (co group) + // build the scatter function (co group) CoGroupOperator<?, ?, Tuple2<K, Message>> messages; - MessagingUdfWithEdgeValues<K, Tuple3<VV, LongValue, LongValue>, VV, Message, EV> messenger = - new MessagingUdfWithEVsVVWithDegrees<K, VV, Message, EV>(messagingFunction, messageTypeInfo); + ScatterUdfWithEdgeValues<K, Tuple3<VV, LongValue, LongValue>, VV, Message, EV> messenger = + new ScatterUdfWithEVsVVWithDegrees<>(scatterFunction, messageTypeInfo); messages = this.edgesWithValue.coGroup(iteration.getWorkset()).where(whereArg) .equalTo(equalToArg).with(messenger); @@ -521,7 +517,7 @@ public class ScatterGatherIteration<K, VV, Message, EV> messages = messages.name("Messaging"); if (this.configuration != null) { - for (Tuple2<String, DataSet<?>> e : this.configuration.getMessagingBcastVars()) { + for (Tuple2<String, DataSet<?>> e : this.configuration.getScatterBcastVars()) { messages = messages.withBroadcastSet(e.f1, e.f0); } if (this.configuration.isOptNumVertices()) { @@ -543,7 +539,7 @@ public class ScatterGatherIteration<K, VV, Message, EV> // set up the iteration operator if (this.configuration != null) { - iteration.name(this.configuration.getName("Scatter-gather iteration (" + updateFunction + " | " + messagingFunction + ")")); + iteration.name(this.configuration.getName("Scatter-gather iteration (" + gatherFunction + " | " + scatterFunction + ")")); iteration.parallelism(this.configuration.getParallelism()); iteration.setSolutionSetUnManaged(this.configuration.isSolutionSetUnmanagedMemory()); @@ -554,7 +550,7 @@ public class ScatterGatherIteration<K, VV, Message, EV> } else { // no configuration provided; set default name - iteration.name("Scatter-gather iteration (" + updateFunction + " | " + messagingFunction + ")"); + iteration.name("Scatter-gather iteration (" + gatherFunction + " | " + scatterFunction + ")"); } } @@ -579,21 +575,21 @@ public class ScatterGatherIteration<K, VV, Message, EV> switch (messagingDirection) { case IN: - messages = buildMessagingFunction(iteration, messageTypeInfo, 1, 0, numberOfVertices); + messages = buildScatterFunction(iteration, messageTypeInfo, 1, 0, numberOfVertices); break; case OUT: - messages = buildMessagingFunction(iteration, messageTypeInfo, 0, 0, numberOfVertices); + messages = buildScatterFunction(iteration, messageTypeInfo, 0, 0, numberOfVertices); break; case ALL: - messages = buildMessagingFunction(iteration, messageTypeInfo, 1, 0, numberOfVertices) - .union(buildMessagingFunction(iteration, messageTypeInfo, 0, 0, numberOfVertices)) ; + messages = buildScatterFunction(iteration, messageTypeInfo, 1, 0, numberOfVertices) + .union(buildScatterFunction(iteration, messageTypeInfo, 0, 0, numberOfVertices)) ; break; default: throw new IllegalArgumentException("Illegal edge direction"); } - VertexUpdateUdf<K, VV, Message> updateUdf = - new VertexUpdateUdfSimpleVV<K, VV, Message>(updateFunction, vertexTypes); + GatherUdf<K, VV, Message> updateUdf = + new GatherUdfSimpleVV<K, VV, Message>(gatherFunction, vertexTypes); // build the update function (co group) CoGroupOperator<?, ?, Vertex<K, VV>> updates = @@ -624,7 +620,7 @@ public class ScatterGatherIteration<K, VV, Message, EV> DataSet<Tuple2<K, Message>> messages; - this.updateFunction.setOptDegrees(this.configuration.isOptDegrees()); + this.gatherFunction.setOptDegrees(this.configuration.isOptDegrees()); DataSet<Tuple2<K, LongValue>> inDegrees = graph.inDegrees(); DataSet<Tuple2<K, LongValue>> outDegrees = graph.outDegrees(); @@ -634,7 +630,7 @@ public class ScatterGatherIteration<K, VV, Message, EV> @Override public void join(Tuple2<K, LongValue> first, Tuple2<K, LongValue> second, Collector<Tuple3<K, LongValue, LongValue>> out) { - out.collect(new Tuple3<K, LongValue, LongValue>(first.f0, first.f1, second.f1)); + out.collect(new Tuple3<>(first.f0, first.f1, second.f1)); } }).withForwardedFieldsFirst("f0;f1").withForwardedFieldsSecond("f1"); @@ -644,9 +640,8 @@ public class ScatterGatherIteration<K, VV, Message, EV> @Override public void join(Vertex<K, VV> vertex, Tuple3<K, LongValue, LongValue> degrees, Collector<Vertex<K, Tuple3<VV, LongValue, LongValue>>> out) throws Exception { - - out.collect(new Vertex<K, Tuple3<VV, LongValue, LongValue>>(vertex.getId(), - new Tuple3<VV, LongValue, LongValue>(vertex.getValue(), degrees.f1, degrees.f2))); + out.collect(new Vertex<>(vertex.getId(), + new Tuple3<>(vertex.getValue(), degrees.f1, degrees.f2))); } }).withForwardedFieldsFirst("f0"); @@ -659,22 +654,22 @@ public class ScatterGatherIteration<K, VV, Message, EV> switch (messagingDirection) { case IN: - messages = buildMessagingFunctionVerticesWithDegrees(iteration, messageTypeInfo, 1, 0, numberOfVertices); + messages = buildScatterFunctionVerticesWithDegrees(iteration, messageTypeInfo, 1, 0, numberOfVertices); break; case OUT: - messages = buildMessagingFunctionVerticesWithDegrees(iteration, messageTypeInfo, 0, 0, numberOfVertices); + messages = buildScatterFunctionVerticesWithDegrees(iteration, messageTypeInfo, 0, 0, numberOfVertices); break; case ALL: - messages = buildMessagingFunctionVerticesWithDegrees(iteration, messageTypeInfo, 1, 0, numberOfVertices) - .union(buildMessagingFunctionVerticesWithDegrees(iteration, messageTypeInfo, 0, 0, numberOfVertices)) ; + messages = buildScatterFunctionVerticesWithDegrees(iteration, messageTypeInfo, 1, 0, numberOfVertices) + .union(buildScatterFunctionVerticesWithDegrees(iteration, messageTypeInfo, 0, 0, numberOfVertices)) ; break; default: throw new IllegalArgumentException("Illegal edge direction"); } @SuppressWarnings({ "unchecked", "rawtypes" }) - VertexUpdateUdf<K, Tuple3<VV, LongValue, LongValue>, Message> updateUdf = - new VertexUpdateUdfVVWithDegrees(updateFunction, vertexTypes); + GatherUdf<K, Tuple3<VV, LongValue, LongValue>, Message> updateUdf = + new GatherUdfVVWithDegrees(gatherFunction, vertexTypes); // build the update function (co group) CoGroupOperator<?, ?, Vertex<K, Tuple3<VV, LongValue, LongValue>>> updates = @@ -690,7 +685,7 @@ public class ScatterGatherIteration<K, VV, Message, EV> new MapFunction<Vertex<K, Tuple3<VV, LongValue, LongValue>>, Vertex<K, VV>>() { public Vertex<K, VV> map(Vertex<K, Tuple3<VV, LongValue, LongValue>> vertex) { - return new Vertex<K, VV>(vertex.getId(), vertex.getValue().f0); + return new Vertex<>(vertex.getId(), vertex.getValue().f0); } }); } @@ -700,7 +695,7 @@ public class ScatterGatherIteration<K, VV, Message, EV> // 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()) { + for (Tuple2<String, DataSet<?>> e : this.configuration.getGatherBcastVars()) { updates = updates.withBroadcastSet(e.f1, e.f0); } } http://git-wip-us.apache.org/repos/asf/flink/blob/918e5d0c/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 deleted file mode 100644 index 9085432..0000000 --- a/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/spargel/VertexUpdateFunction.java +++ /dev/null @@ -1,251 +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.io.Serializable; -import java.util.Collection; - -import org.apache.flink.api.common.aggregators.Aggregator; -import org.apache.flink.api.common.functions.IterationRuntimeContext; -import org.apache.flink.api.java.tuple.Tuple3; -import org.apache.flink.graph.Vertex; -import org.apache.flink.types.Value; -import org.apache.flink.util.Collector; - -/** - * This class must be extended by functions that compute the state of the vertex depending on the old state and the - * incoming messages. The central method is {@link #updateVertex(Vertex, MessageIterator)}, which is - * invoked once per vertex per superstep. - * - * {@code <K>} The vertex key type. - * {@code <VV>} The vertex value type. - * {@code <Message>} The message type. - */ -public abstract class VertexUpdateFunction<K, VV, 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; - - /** - * Retrieves the number of vertices in the graph. - * @return the number of vertices if the {@link org.apache.flink.graph.IterationConfiguration#setOptNumVertices(boolean)} - * option has been set; -1 otherwise. - */ - public long getNumberOfVertices() { - return numberOfVertices; - } - - void setNumberOfVertices(long numberOfVertices) { - this.numberOfVertices = numberOfVertices; - } - - //--------------------------------------------------------------------------------------------- - - private boolean optDegrees; - - boolean isOptDegrees() { - return optDegrees; - } - - void setOptDegrees(boolean optDegrees) { - this.optDegrees = optDegrees; - } - - // -------------------------------------------------------------------------------------------- - // Public API Methods - // -------------------------------------------------------------------------------------------- - - /** - * This method is invoked once per vertex per superstep. It receives the current state of the vertex, as well as - * the incoming messages. It may set a new vertex state via {@link #setNewVertexValue(Object)}. If the vertex - * state is changed, it will trigger the sending of messages via the {@link MessagingFunction}. - * - * @param vertex The vertex. - * @param inMessages The incoming messages to this vertex. - * - * @throws Exception The computation may throw exceptions, which causes the superstep to fail. - */ - public abstract void updateVertex(Vertex<K, VV> vertex, MessageIterator<Message> inMessages) throws Exception; - - /** - * This method is executed one per superstep before the vertex update function is invoked for each vertex. - * - * @throws Exception Exceptions in the pre-superstep phase cause the superstep to fail. - */ - public void preSuperstep() throws Exception {} - - /** - * This method is executed one per superstep after the vertex update function has been invoked for each vertex. - * - * @throws Exception Exceptions in the post-superstep phase cause the superstep to fail. - */ - public void postSuperstep() throws Exception {} - - /** - * Sets the new value of this vertex. Setting a new value triggers the sending of outgoing messages from this vertex. - * - * This should be called at most once per updateVertex. - * - * @param newValue The new vertex value. - */ - public void setNewVertexValue(VV newValue) { - if(setNewVertexValueCalled) { - throw new IllegalStateException("setNewVertexValue should only be called at most once per updateVertex"); - } - setNewVertexValueCalled = true; - if(isOptDegrees()) { - outValWithDegrees.f1.f0 = newValue; - outWithDegrees.collect(outValWithDegrees); - } else { - outVal.setValue(newValue); - out.collect(outVal); - } - } - - /** - * Gets the number of the superstep, starting at <tt>1</tt>. - * - * @return The number of the current superstep. - */ - public int getSuperstepNumber() { - return this.runtimeContext.getSuperstepNumber(); - } - - /** - * Gets the iteration aggregator registered under the given name. The iteration aggregator combines - * all aggregates globally once per superstep and makes them available in the next superstep. - * - * @param name The name of the aggregator. - * @return The aggregator registered under this name, or null, if no aggregator was registered. - */ - public <T extends Aggregator<?>> T getIterationAggregator(String name) { - return this.runtimeContext.<T>getIterationAggregator(name); - } - - /** - * Get the aggregated value that an aggregator computed in the previous iteration. - * - * @param name The name of the aggregator. - * @return The aggregated value of the previous iteration. - */ - public <T extends Value> T getPreviousIterationAggregate(String name) { - return this.runtimeContext.<T>getPreviousIterationAggregate(name); - } - - /** - * 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.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. - */ - public <T> Collection<T> getBroadcastSet(String name) { - return this.runtimeContext.<T>getBroadcastVariable(name); - } - - // -------------------------------------------------------------------------------------------- - // internal methods - // -------------------------------------------------------------------------------------------- - - private IterationRuntimeContext runtimeContext; - - private Collector<Vertex<K, VV>> out; - - private Collector<Vertex<K, Tuple3<VV, Long, Long>>> outWithDegrees; - - private Vertex<K, VV> outVal; - - private Vertex<K, Tuple3<VV, Long, Long>> outValWithDegrees; - - private long inDegree = -1; - - private long outDegree = -1; - - private boolean setNewVertexValueCalled; - - void init(IterationRuntimeContext context) { - this.runtimeContext = context; - } - - void setOutput(Vertex<K, VV> outVal, Collector<Vertex<K, VV>> out) { - this.outVal = outVal; - this.out = out; - setNewVertexValueCalled = false; - } - - @SuppressWarnings({ "unchecked", "rawtypes" }) - <ValueWithDegree> void setOutputWithDegrees(Vertex<K, ValueWithDegree> outVal, - Collector out) { - this.outValWithDegrees = (Vertex<K, Tuple3<VV, Long, Long>>) outVal; - this.outWithDegrees = out; - setNewVertexValueCalled = false; - } - - /** - * Retrieves the vertex in-degree (number of in-coming edges). - * @return The in-degree of this vertex - */ - public long getInDegree() { - return inDegree; - } - - void setInDegree(long inDegree) { - this.inDegree = inDegree; - } - - /** - * Retrieve the vertex out-degree (number of out-going edges). - * @return The out-degree of this vertex - */ - public long getOutDegree() { - return outDegree; - } - - void setOutDegree(long outDegree) { - this.outDegree = outDegree; - } - - /** - * 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.ScatterGatherIteration}. - * - * This function will retrieve the vertex from the vertexState and will set its degrees, afterwards calling - * the regular updateVertex function. - * - * @param vertexState - * @param inMessages - * @throws Exception - */ - @SuppressWarnings("unchecked") - <VertexWithDegree> void updateVertexFromScatterGatherIteration(Vertex<K, VertexWithDegree> vertexState, - MessageIterator<Message> inMessages) throws Exception { - - Vertex<K, VV> vertex = new Vertex<K, VV>(vertexState.f0, - ((Tuple3<VV, Long, Long>)vertexState.getValue()).f0); - - updateVertex(vertex, inMessages); - } -} http://git-wip-us.apache.org/repos/asf/flink/blob/918e5d0c/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 3a750af..14c2fb4 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 @@ -75,8 +75,8 @@ public class SpargelCompilerTest extends CompilerTestBase { Graph<Long, Long, NullValue> graph = Graph.fromDataSet(initialVertices, edges, env); DataSet<Vertex<Long, Long>> result = graph.runScatterGatherIteration( - new ConnectedComponents.CCUpdater<Long, Long>(), - new ConnectedComponents.CCMessenger<Long, Long>(BasicTypeInfo.LONG_TYPE_INFO), 100) + new ConnectedComponents.CCMessenger<Long, Long>(BasicTypeInfo.LONG_TYPE_INFO), + new ConnectedComponents.CCUpdater<Long, Long>(), 100) .getVertices(); result.output(new DiscardingOutputFormat<Vertex<Long, Long>>()); @@ -157,12 +157,12 @@ public class SpargelCompilerTest extends CompilerTestBase { Graph<Long, Long, NullValue> graph = Graph.fromDataSet(initialVertices, edges, env); ScatterGatherConfiguration parameters = new ScatterGatherConfiguration(); - parameters.addBroadcastSetForMessagingFunction(BC_VAR_NAME, bcVar); - parameters.addBroadcastSetForUpdateFunction(BC_VAR_NAME, bcVar); + parameters.addBroadcastSetForScatterFunction(BC_VAR_NAME, bcVar); + parameters.addBroadcastSetForGatherFunction(BC_VAR_NAME, bcVar); DataSet<Vertex<Long, Long>> result = graph.runScatterGatherIteration( - new ConnectedComponents.CCUpdater<Long, Long>(), - new ConnectedComponents.CCMessenger<Long, Long>(BasicTypeInfo.LONG_TYPE_INFO), 100) + new ConnectedComponents.CCMessenger<Long, Long>(BasicTypeInfo.LONG_TYPE_INFO), + new ConnectedComponents.CCUpdater<Long, Long>(), 100) .getVertices(); result.output(new DiscardingOutputFormat<Vertex<Long, Long>>());
