Github user vasia commented on a diff in the pull request:
https://github.com/apache/flink/pull/537#discussion_r28643926
--- Diff:
flink-staging/flink-gelly/src/main/java/org/apache/flink/graph/spargel/VertexCentricIteration.java
---
@@ -138,69 +146,46 @@ public void setInput(DataSet<Vertex<VertexKey,
VertexValue>> inputData) {
if (this.initialVertices == null) {
throw new IllegalStateException("The input data set has
not been set.");
}
-
+
// prepare some type information
- TypeInformation<Vertex<VertexKey, VertexValue>> vertexTypes =
initialVertices.getType();
TypeInformation<VertexKey> keyType = ((TupleTypeInfo<?>)
initialVertices.getType()).getTypeAt(0);
TypeInformation<Tuple2<VertexKey, Message>> messageTypeInfo =
new TupleTypeInfo<Tuple2<VertexKey,Message>>(keyType, messageType);
- final int[] zeroKeyPos = new int[] {0};
-
- final DeltaIteration<Vertex<VertexKey, VertexValue>,
Vertex<VertexKey, VertexValue>> iteration =
- this.initialVertices.iterateDelta(this.initialVertices,
this.maximumNumberOfIterations, zeroKeyPos);
+ // create a graph
+ Graph<VertexKey, VertexValue, EdgeValue> graph =
+ Graph.fromDataSet(initialVertices,
edgesWithValue, ExecutionEnvironment.getExecutionEnvironment());
- // set up the iteration operator
- if (this.configuration != null) {
+ // check whether the numVertices option is set and, if so,
compute the total number of vertices
+ // and set it within the messaging and update functions
- iteration.name(this.configuration.getName(
- "Vertex-centric iteration (" +
updateFunction + " | " + messagingFunction + ")"));
-
iteration.parallelism(this.configuration.getParallelism());
-
iteration.setSolutionSetUnManaged(this.configuration.isSolutionSetUnmanagedMemory());
-
- // register all aggregators
- for (Map.Entry<String, Aggregator<?>> entry :
this.configuration.getAggregators().entrySet()) {
- iteration.registerAggregator(entry.getKey(),
entry.getValue());
+ if (this.configuration != null &&
this.configuration.isOptNumVertices()) {
+ try {
+ long numberOfVertices =
graph.numberOfVertices();
+
messagingFunction.setNumberOfVertices(numberOfVertices);
+
updateFunction.setNumberOfVertices(numberOfVertices);
+ } catch (Exception e) {
+ e.printStackTrace();
}
}
- else {
- // no configuration provided; set default name
- iteration.name("Vertex-centric iteration (" +
updateFunction + " | " + messagingFunction + ")");
- }
-
- // build the messaging function (co group)
- CoGroupOperator<?, ?, Tuple2<VertexKey, Message>> messages;
- MessagingUdfWithEdgeValues<VertexKey, VertexValue, Message,
EdgeValue> messenger = new MessagingUdfWithEdgeValues<VertexKey, VertexValue,
Message, EdgeValue>(messagingFunction, messageTypeInfo);
- messages =
this.edgesWithValue.coGroup(iteration.getWorkset()).where(0).equalTo(0).with(messenger);
-
- // configure coGroup message function with name and broadcast
variables
- messages = messages.name("Messaging");
- if (this.configuration != null) {
- for (Tuple2<String, DataSet<?>> e :
this.configuration.getMessagingBcastVars()) {
- messages = messages.withBroadcastSet(e.f1,
e.f0);
- }
+ if(this.configuration != null) {
+
messagingFunction.setDirection(this.configuration.getDirection());
+ } else {
+ messagingFunction.setDirection(EdgeDirection.OUT);
}
-
- VertexUpdateUdf<VertexKey, VertexValue, Message> updateUdf =
new VertexUpdateUdf<VertexKey, VertexValue, Message>(updateFunction,
vertexTypes);
-
- // build the update function (co group)
- CoGroupOperator<?, ?, Vertex<VertexKey, VertexValue>> updates =
-
messages.coGroup(iteration.getSolutionSet()).where(0).equalTo(0).with(updateUdf);
-
- // configure coGroup update function with name and broadcast
variables
- updates = updates.name("Vertex State Updates");
- if (this.configuration != null) {
- for (Tuple2<String, DataSet<?>> e :
this.configuration.getUpdateBcastVars()) {
- updates = updates.withBroadcastSet(e.f1, e.f0);
- }
- }
+ // retrieve the direction in which the updates are made and in
which the messages are sent
+ EdgeDirection messagingDirection =
messagingFunction.getDirection();
- // let the operator know that we preserve the key field
-
updates.withForwardedFieldsFirst("0").withForwardedFieldsSecond("0");
-
- return iteration.closeWith(updates, updates);
-
+ DataSet<Tuple2<VertexKey, Message>> messages = null;
--- End diff --
why create a null dataset and pass it as a parameter to the methods below?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at [email protected] or file a JIRA ticket
with INFRA.
---
