[FLINK-1522][gelly] Fixed faulty split into create/runVertexCentricIt…

…eration This closes #429
apache · Mar 4, 2015 · 8961bd1 · 8961bd1
1 parent f329e4e
commit 8961bd1
Show file tree

Hide file tree

Showing 2 changed files with 36 additions and 4 deletions.
diff --git a/flink-staging/flink-gelly/pom.xml b/flink-staging/flink-gelly/pom.xml
@@ -52,4 +52,35 @@ under the License.
 			<scope>test</scope>
 		</dependency>
 	</dependencies>
+
+	<!-- See main pom.xml for explanation of profiles -->
+	<profiles>
+		<profile>
+			<id>hadoop-1</id>
+			<activation>
+				<property>
+					<!-- Please do not remove the 'hadoop1' comment. See ./tools/generate_specific_pom.sh -->
+					<!--hadoop1--><name>hadoop.profile</name><value>1</value>
+				</property>
+			</activation>
+			<dependencies>
+				<!-- Add this here, for hadoop-2 we don't need it since we get guava transitively -->
+				<dependency>
+					<groupId>com.google.guava</groupId>
+					<artifactId>guava</artifactId>
+					<version>${guava.version}</version>
+					<scope>provided</scope>
+				</dependency>
+			</dependencies>
+		</profile>
+		<profile>
+			<id>hadoop-2</id>
+			<activation>
+				<property>
+					<!-- Please do not remove the 'hadoop2' comment. See ./tools/generate_specific_pom.sh -->
+					<!--hadoop2--><name>!hadoop.profile</name>
+				</property>
+			</activation>
+		</profile>
+	</profiles>
 </project>
diff --git a/...g/flink-gelly/src/main/java/org/apache/flink/graph/library/SingleSourceShortestPaths.java b/...g/flink-gelly/src/main/java/org/apache/flink/graph/library/SingleSourceShortestPaths.java
@@ -45,11 +45,12 @@ public SingleSourceShortestPaths(K srcVertexId, Integer maxIterations) {
 	@Override
 	public Graph<K, Double, Double> run(Graph<K, Double, Double> input) {
 
-		VertexCentricIteration<K, Double, Double, Double> iteration = input.mapVertices(
-				new InitVerticesMapper<K>(srcVertexId)).createVertexCentricIteration(new VertexDistanceUpdater<K>(),
-				new MinDistanceMessenger<K>(), maxIterations);
+		Graph<K, Double, Double> mappedInput = input.mapVertices(new InitVerticesMapper<K>(srcVertexId));
 
-		return input.runVertexCentricIteration(iteration);
+		VertexCentricIteration<K, Double, Double, Double> iteration = mappedInput.createVertexCentricIteration(
+				new VertexDistanceUpdater<K>(), new MinDistanceMessenger<K>(), maxIterations);
+
+		return mappedInput.runVertexCentricIteration(iteration);
 	}
 
 	public static final class InitVerticesMapper<K extends Comparable<K> & Serializable>