[FLINK-2451] [gelly] removed redundant examples; added comments descr…

…ibing which gelly method each example illustrates.

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

@@ -31,12 +31,11 @@
 import org.apache.flink.types.NullValue;
 
 /**
- * This example shows how to use the {@link org.apache.flink.graph.library.ConnectedComponentsAlgorithm}
- * library method:
- * <ul>
- * 	<li> with the edge data set given as a parameter
- * 	<li> with default data
- * </ul>
+ * This example shows how to use Gelly's library methods.
+ * You can find all available library methods in {@link org.apache.flink.graph.library}. 
+ * 
+ * In particular, this example uses the {@link org.apache.flink.graph.library.ConnectedComponentsAlgorithm}
+ * library method to compute the connected components of the input graph.
  *
  * The input file is a plain text file and must be formatted as follows:
  * Edges are represented by tuples of srcVertexId, trgVertexId which are

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

@@ -33,6 +33,9 @@
 import java.io.Serializable;
 
 /**
+ * This example shows how to use Gelly's {@link Graph#getTriplets()} and
+ * {@link Graph#joinWithEdges(DataSet, MapFunction)} methods.
+ * 
  * Given a directed, unweighted graph, with vertex values representing points in a plan,
  * return a weighted graph where the edge weights are equal to the Euclidean distance between the
  * src and the trg vertex values.

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

@@ -33,7 +33,19 @@
 import org.apache.flink.graph.utils.Tuple3ToEdgeMap;
 
 /**
- * This is an implementation of the Single Source Shortest Paths algorithm, using a gather-sum-apply iteration
+ * This example shows how to use Gelly's Gather-Sum-Apply iterations.
+ * 
+ * It is an implementation of the Single-Source-Shortest-Paths algorithm.
+ * For a vertex-centric implementation of the same algorithm, please refer to {@link SingleSourceShortestPaths}. 
+ *
+ * The input file is a plain text file and must be formatted as follows:
+ * Edges are represented by tuples of srcVertexId, trgVertexId, distance which are
+ * separated by tabs. Edges themselves are separated by newlines.
+ * For example: <code>1\t2\t0.1\n1\t3\t1.4\n</code> defines two edges,
+ * edge 1-2 with distance 0.1, and edge 1-3 with distance 1.4.
+ *
+ * If no parameters are provided, the program is run with default data from
+ * {@link org.apache.flink.graph.example.utils.SingleSourceShortestPathsData}
  */
 public class GSASingleSourceShortestPaths implements ProgramDescription {
 
@@ -54,9 +66,8 @@ public static void main(String[] args) throws Exception {
 		Graph<Long, Double, Double> graph = Graph.fromDataSet(edges, new InitVertices(srcVertexId), env);
 
 		// Execute the GSA iteration
-		Graph<Long, Double, Double> result = graph
-				.runGatherSumApplyIteration(new CalculateDistances(), new ChooseMinDistance(),
-						new UpdateDistance(), maxIterations);
+		Graph<Long, Double, Double> result = graph.runGatherSumApplyIteration(
+				new CalculateDistances(), new ChooseMinDistance(), new UpdateDistance(), maxIterations);
 
 		// Extract the vertices as the result
 		DataSet<Vertex<Long, Double>> singleSourceShortestPaths = result.getVertices();
@@ -73,6 +84,10 @@ public static void main(String[] args) throws Exception {
 
 	}
 
+	// --------------------------------------------------------------------------------------------
+	//  Single Source Shortest Path UDFs
+	// --------------------------------------------------------------------------------------------
+
 	@SuppressWarnings("serial")
 	private static final class InitVertices implements MapFunction<Long, Double>{
 
@@ -92,10 +107,6 @@ public Double map(Long id) {
 		}
 	}
 
-	// --------------------------------------------------------------------------------------------
-	//  Single Source Shortest Path UDFs
-	// --------------------------------------------------------------------------------------------
-
 	@SuppressWarnings("serial")
 	private static final class CalculateDistances extends GatherFunction<Double, Double, Double> {
 

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

@@ -30,8 +30,9 @@
 import org.apache.flink.types.NullValue;
 
 /**
+ * This example illustrate how to use Gelly metrics methods and get simple statistics
+ * from the input graph.  
  * 
- * A simple example to illustrate the basic functionality of the graph-api.
  * The program creates a random graph and computes and prints
  * the following metrics:
  * - number of vertices

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

@@ -35,11 +35,11 @@
 import org.apache.flink.graph.utils.Tuple3ToEdgeMap;
 
 /**
- * Incremental Single Sink Shortest Paths Example. Shortest Paths are incrementally updated
- * upon edge removal.
- *
  * This example illustrates the usage of vertex-centric iteration's
  * messaging direction configuration option.
+ * 
+ * Incremental Single Sink Shortest Paths Example. Shortest Paths are incrementally updated
+ * upon edge removal.
  *
  * The program takes as input the resulted graph after a SSSP computation,
  * an edge to be removed and the initial graph(i.e. before SSSP was computed).

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

@@ -34,6 +34,13 @@
 import java.util.HashSet;
 
 /**
+ * This example shows how to use
+ * <ul>
+ *  <li> neighborhood methods
+ *  <li> join with vertices
+ *  <li> triplets
+ * </ul>
+ * 
  * Given a directed, unweighted graph, return a weighted graph where the edge values are equal
  * to the Jaccard similarity coefficient - the number of common neighbors divided by the the size
  * of the union of neighbor sets - for the src and target vertices.
@@ -117,8 +124,7 @@ public String getDescription() {
 	private static final class GatherNeighbors implements ReduceNeighborsFunction<HashSet<Long>> {
 
 		@Override
-		public HashSet<Long> reduceNeighbors(HashSet<Long> first,
-											HashSet<Long> second) {
+		public HashSet<Long> reduceNeighbors(HashSet<Long> first, HashSet<Long> second) {
 			first.addAll(second);
 			return new HashSet<Long>(first);
 		}