From ccccf731940a586fad70c74a3797eb7ca51674fc Mon Sep 17 00:00:00 2001 From: Greg Hogan Date: Wed, 24 Aug 2016 11:32:43 -0400 Subject: [PATCH 1/4] [FLINK-4204] [gelly] Clean up gelly-examples Moves drivers into separate package. Adds default main class to print usage listing included classes. Includes documentation for running Gelly examples. --- docs/dev/libs/gelly/index.md | 61 ++++++++++++++++++- flink-libraries/flink-gelly-examples/pom.xml | 13 ++++ .../java/org/apache/flink/graph/Usage.java | 61 +++++++++++++++++++ .../ClusteringCoefficient.java | 40 ++++++++---- .../graph/{examples => drivers}/Graph500.java | 10 +-- .../{driver => drivers}/GraphMetrics.java | 8 +-- .../graph/{examples => drivers}/HITS.java | 8 +-- .../{examples => drivers}/JaccardIndex.java | 28 ++++++--- .../TriangleListing.java | 30 +++++++-- 9 files changed, 221 insertions(+), 38 deletions(-) create mode 100644 flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/Usage.java rename flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/{examples => drivers}/ClusteringCoefficient.java (91%) rename flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/{examples => drivers}/Graph500.java (94%) rename flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/{driver => drivers}/GraphMetrics.java (97%) rename flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/{examples => drivers}/HITS.java (97%) rename flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/{examples => drivers}/JaccardIndex.java (89%) rename flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/{examples => drivers}/TriangleListing.java (87%) diff --git a/docs/dev/libs/gelly/index.md b/docs/dev/libs/gelly/index.md index 2eeec2c2704a8..0d9bdb592e7a8 100644 --- a/docs/dev/libs/gelly/index.md +++ b/docs/dev/libs/gelly/index.md @@ -64,6 +64,65 @@ Add the following dependency to your `pom.xml` to use Gelly. Note that Gelly is currently not part of the binary distribution. See linking with it for cluster execution [here]({{ site.baseurl }}/dev/cluster_execution.html#linking-with-modules-not-contained-in-the-binary-distribution). -The remaining sections provide a description of available methods and present several examples of how to use Gelly and how to mix it with the Flink DataSet API. After reading this guide, you might also want to check the {% gh_link /flink-libraries/flink-gelly-examples/ "Gelly examples" %}. +The remaining sections provide a description of available methods and present several examples of how to use Gelly and how to mix it with the Flink DataSet API. + +Running Gelly Examples +---------------------- + +The Gelly library and examples jars are [provided](https://flink.apache.org/downloads.html "Apache Flink: Downloads") +in the Flink distribution at **opt/lib/gelly** (for versions older than Flink 1.2 these can be manually downloaded from +[Maven Central](http://search.maven.org/#search|ga|1|flink%20gelly). + +To run the Gelly examples the **flink-gelly** (for Java) or **flink-gelly-scala** (for Scala) jar must be copied to +Flink's **lib** directory. + +~~~bash +cp opt/lib/gelly/flink-gelly_*.jar lib/ +cp opt/lib/gelly/flink-gelly-scala_*.jar lib/ +~~~ + +Gelly's example jar includes both drivers for the library methods as well as additional example algorithms. After +configuring and starting the cluster, list the available algorithm classes: + +~~~bash +./bin/start-cluster.sh +./bin/flink run opt/lib/gelly/flink-gelly-examples_*.jar +~~~ + +The Gelly drivers can generate [RMat](http://www.cs.cmu.edu/~christos/PUBLICATIONS/siam04.pdf) graph data or read the +edge list from a CSV file. Each node in a cluster must have access to the input file. Calculate graph metrics on a +directed generated graph: + +~~~bash +./bin/flink run -c org.apache.flink.graph.drivers.GraphMetrics opt/lib/gelly/flink-gelly-examples_*.jar \ + --directed true --input rmat +~~~ + +The size of the graph is adjusted by the *\-\-scale* and *\-\-edge_factor* parameters. The +[library generator](./graph_generators.html#rmat-graph) provides access to additional configuration to adjust the +power-law skew and random noise. + +Sample social network data is provided by the [Stanford Network Analysis Project](http://snap.stanford.edu/data/index.html). +The [com-lj](http://snap.stanford.edu/data/bigdata/communities/com-lj.ungraph.txt.gz) data set is a good starter size. +Run a few algorithms and monitor the job progress in Flink's Web UI: + +~~~bash +wget -O - http://snap.stanford.edu/data/bigdata/communities/com-lj.ungraph.txt.gz | gunzip -c > com-lj.ungraph.txt + +./bin/flink run -q -c org.apache.flink.graph.drivers.GraphMetrics opt/lib/gelly/flink-gelly-examples_*.jar \ + --directed true --input csv --type integer --input_filename com-lj.ungraph.txt --input_field_delimiter '\t' + +./bin/flink run -q -c org.apache.flink.graph.drivers.ClusteringCoefficient opt/lib/gelly/flink-gelly-examples_*.jar \ + --directed true --input csv --type integer --input_filename com-lj.ungraph.txt --input_field_delimiter '\t' \ + --output hash + +./bin/flink run -q -c org.apache.flink.graph.drivers.JaccardIndex opt/lib/gelly/flink-gelly-examples_*.jar \ + --input csv --type integer --simplify true --input_filename com-lj.ungraph.txt --input_field_delimiter '\t' \ + --output hash +~~~ + +Please submit feature requests or issue reports through the user [mailing list](https://flink.apache.org/community.html#mailing-lists) +or [Flink Jira](https://issues.apache.org/jira/browse/FLINK). We welcome suggestions for new algorithms and features as +well as [code contributions](https://flink.apache.org/contribute-code.html). {% top %} diff --git a/flink-libraries/flink-gelly-examples/pom.xml b/flink-libraries/flink-gelly-examples/pom.xml index 80da0ffdb0582..9b90b04306b2d 100644 --- a/flink-libraries/flink-gelly-examples/pom.xml +++ b/flink-libraries/flink-gelly-examples/pom.xml @@ -143,6 +143,19 @@ + + org.apache.maven.plugins + maven-jar-plugin + 2.5 + + + + org.apache.flink.graph.Usage + + + + + org.codehaus.mojo diff --git a/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/Usage.java b/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/Usage.java new file mode 100644 index 0000000000000..f10377f0f9ae9 --- /dev/null +++ b/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/Usage.java @@ -0,0 +1,61 @@ +/* + * 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; + +/** + * This default main class prints usage listing available classes. + */ +public class Usage { + + private static final Class[] DRIVERS = new Class[]{ + org.apache.flink.graph.drivers.ClusteringCoefficient.class, + org.apache.flink.graph.drivers.Graph500.class, + org.apache.flink.graph.drivers.GraphMetrics.class, + org.apache.flink.graph.drivers.HITS.class, + org.apache.flink.graph.drivers.JaccardIndex.class, + org.apache.flink.graph.drivers.TriangleListing.class, + }; + + private static final Class[] EXAMPLES = new Class[]{ + org.apache.flink.graph.examples.ConnectedComponents.class, + org.apache.flink.graph.examples.EuclideanGraphWeighing.class, + org.apache.flink.graph.examples.GSASingleSourceShortestPaths.class, + org.apache.flink.graph.examples.IncrementalSSSP.class, + org.apache.flink.graph.examples.MusicProfiles.class, + org.apache.flink.graph.examples.PregelSSSP.class, + org.apache.flink.graph.examples.SingleSourceShortestPaths.class, + org.apache.flink.graph.scala.examples.ConnectedComponents.class, + org.apache.flink.graph.scala.examples.GraphMetrics.class, + org.apache.flink.graph.scala.examples.GSASingleSourceShortestPaths.class, + org.apache.flink.graph.scala.examples.SingleSourceShortestPaths.class, + }; + + public static void main(String[] args) throws Exception { + System.out.println("Driver classes call algorithms from the Gelly library:"); + for (Class cls : DRIVERS) { + System.out.println(" " + cls.getName()); + } + + System.out.println(""); + System.out.println("Example classes illustrate Gelly APIs or alternative algorithms:"); + for (Class cls : EXAMPLES) { + System.out.println(" " + cls.getName()); + } + } +} diff --git a/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/examples/ClusteringCoefficient.java b/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/ClusteringCoefficient.java similarity index 91% rename from flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/examples/ClusteringCoefficient.java rename to flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/ClusteringCoefficient.java index 615d765f7ab01..2d4ba7d55ad58 100644 --- a/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/examples/ClusteringCoefficient.java +++ b/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/ClusteringCoefficient.java @@ -16,7 +16,7 @@ * limitations under the License. */ -package org.apache.flink.graph.examples; +package org.apache.flink.graph.drivers; import org.apache.commons.lang3.StringEscapeUtils; import org.apache.commons.lang3.text.WordUtils; @@ -59,11 +59,11 @@ */ public class ClusteringCoefficient { - public static final int DEFAULT_SCALE = 10; + private static final int DEFAULT_SCALE = 10; - public static final int DEFAULT_EDGE_FACTOR = 16; + private static final int DEFAULT_EDGE_FACTOR = 16; - public static final boolean DEFAULT_CLIP_AND_FLIP = true; + private static final boolean DEFAULT_CLIP_AND_FLIP = true; private static void printUsage() { System.out.println(WordUtils.wrap("The local clustering coefficient measures the connectedness of each" + @@ -77,7 +77,7 @@ private static void printUsage() { System.out.println("usage: ClusteringCoefficient --directed --input --output "); System.out.println(); System.out.println("options:"); - System.out.println(" --input csv --type --input_filename FILENAME [--input_line_delimiter LINE_DELIMITER] [--input_field_delimiter FIELD_DELIMITER]"); + System.out.println(" --input csv --type [--simplify ] --input_filename FILENAME [--input_line_delimiter LINE_DELIMITER] [--input_field_delimiter FIELD_DELIMITER]"); System.out.println(" --input rmat [--scale SCALE] [--edge_factor EDGE_FACTOR]"); System.out.println(); System.out.println(" --output print"); @@ -125,6 +125,11 @@ public static void main(String[] args) throws Exception { .keyType(LongValue.class); if (directedAlgorithm) { + if (parameters.getBoolean("simplify", false)) { + graph = graph + .run(new org.apache.flink.graph.asm.simple.directed.Simplify()); + } + gcc = graph .run(new org.apache.flink.graph.library.clustering.directed.GlobalClusteringCoefficient() .setLittleParallelism(little_parallelism)); @@ -135,6 +140,11 @@ public static void main(String[] args) throws Exception { .run(new org.apache.flink.graph.library.clustering.directed.LocalClusteringCoefficient() .setLittleParallelism(little_parallelism)); } else { + if (parameters.getBoolean("simplify", false)) { + graph = graph + .run(new org.apache.flink.graph.asm.simple.undirected.Simplify(false)); + } + gcc = graph .run(new org.apache.flink.graph.library.clustering.undirected.GlobalClusteringCoefficient() .setLittleParallelism(little_parallelism)); @@ -152,6 +162,11 @@ public static void main(String[] args) throws Exception { .keyType(StringValue.class); if (directedAlgorithm) { + if (parameters.getBoolean("simplify", false)) { + graph = graph + .run(new org.apache.flink.graph.asm.simple.directed.Simplify()); + } + gcc = graph .run(new org.apache.flink.graph.library.clustering.directed.GlobalClusteringCoefficient() .setLittleParallelism(little_parallelism)); @@ -162,6 +177,11 @@ public static void main(String[] args) throws Exception { .run(new org.apache.flink.graph.library.clustering.directed.LocalClusteringCoefficient() .setLittleParallelism(little_parallelism)); } else { + if (parameters.getBoolean("simplify", false)) { + graph = graph + .run(new org.apache.flink.graph.asm.simple.undirected.Simplify(false)); + } + gcc = graph .run(new org.apache.flink.graph.library.clustering.undirected.GlobalClusteringCoefficient() .setLittleParallelism(little_parallelism)); @@ -286,14 +306,10 @@ public static void main(String[] args) throws Exception { System.out.println(result.toVerboseString()); } } - System.out.println(gcc.getResult()); - System.out.println(acc.getResult()); break; case "hash": System.out.println(DataSetUtils.checksumHashCode(lcc)); - System.out.println(gcc.getResult()); - System.out.println(acc.getResult()); break; case "csv": @@ -308,9 +324,6 @@ public static void main(String[] args) throws Exception { lcc.writeAsCsv(filename, lineDelimiter, fieldDelimiter); env.execute("Clustering Coefficient"); - - System.out.println(gcc.getResult()); - System.out.println(acc.getResult()); break; default: @@ -318,6 +331,9 @@ public static void main(String[] args) throws Exception { return; } + System.out.println(gcc.getResult()); + System.out.println(acc.getResult()); + JobExecutionResult result = env.getLastJobExecutionResult(); NumberFormat nf = NumberFormat.getInstance(); diff --git a/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/examples/Graph500.java b/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/Graph500.java similarity index 94% rename from flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/examples/Graph500.java rename to flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/Graph500.java index 73bba2cabdce2..d0c8717a7ea02 100644 --- a/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/examples/Graph500.java +++ b/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/Graph500.java @@ -16,7 +16,7 @@ * limitations under the License. */ -package org.apache.flink.graph.examples; +package org.apache.flink.graph.drivers; import org.apache.commons.math3.random.JDKRandomGenerator; import org.apache.flink.api.common.JobExecutionResult; @@ -45,13 +45,13 @@ */ public class Graph500 { - public static final int DEFAULT_SCALE = 10; + private static final int DEFAULT_SCALE = 10; - public static final int DEFAULT_EDGE_FACTOR = 16; + private static final int DEFAULT_EDGE_FACTOR = 16; - public static final boolean DEFAULT_SIMPLIFY = false; + private static final boolean DEFAULT_SIMPLIFY = false; - public static final boolean DEFAULT_CLIP_AND_FLIP = true; + private static final boolean DEFAULT_CLIP_AND_FLIP = true; public static void main(String[] args) throws Exception { // Set up the execution environment diff --git a/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/driver/GraphMetrics.java b/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/GraphMetrics.java similarity index 97% rename from flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/driver/GraphMetrics.java rename to flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/GraphMetrics.java index 79c5f805c067c..e1e06aefdf3a1 100644 --- a/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/driver/GraphMetrics.java +++ b/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/GraphMetrics.java @@ -16,7 +16,7 @@ * limitations under the License. */ -package org.apache.flink.graph.driver; +package org.apache.flink.graph.drivers; import org.apache.commons.lang3.StringEscapeUtils; import org.apache.commons.lang3.text.WordUtils; @@ -50,11 +50,11 @@ */ public class GraphMetrics { - public static final int DEFAULT_SCALE = 10; + private static final int DEFAULT_SCALE = 10; - public static final int DEFAULT_EDGE_FACTOR = 16; + private static final int DEFAULT_EDGE_FACTOR = 16; - public static final boolean DEFAULT_CLIP_AND_FLIP = true; + private static final boolean DEFAULT_CLIP_AND_FLIP = true; private static void printUsage() { System.out.println(WordUtils.wrap("Computes vertex and edge metrics on a directed or undirected graph.", 80)); diff --git a/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/examples/HITS.java b/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/HITS.java similarity index 97% rename from flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/examples/HITS.java rename to flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/HITS.java index f70d5dcf1d9e6..fe792437450c4 100644 --- a/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/examples/HITS.java +++ b/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/HITS.java @@ -16,7 +16,7 @@ * limitations under the License. */ -package org.apache.flink.graph.examples; +package org.apache.flink.graph.drivers; import org.apache.commons.lang3.StringEscapeUtils; import org.apache.commons.lang3.text.WordUtils; @@ -54,11 +54,11 @@ */ public class HITS { - public static final int DEFAULT_ITERATIONS = 10; + private static final int DEFAULT_ITERATIONS = 10; - public static final int DEFAULT_SCALE = 10; + private static final int DEFAULT_SCALE = 10; - public static final int DEFAULT_EDGE_FACTOR = 16; + private static final int DEFAULT_EDGE_FACTOR = 16; private static void printUsage() { System.out.println(WordUtils.wrap("Hyperlink-Induced Topic Search computes two interdependent" + diff --git a/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/examples/JaccardIndex.java b/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/JaccardIndex.java similarity index 89% rename from flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/examples/JaccardIndex.java rename to flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/JaccardIndex.java index 2845e2ded4769..b9f0b417a3583 100644 --- a/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/examples/JaccardIndex.java +++ b/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/JaccardIndex.java @@ -16,7 +16,7 @@ * limitations under the License. */ -package org.apache.flink.graph.examples; +package org.apache.flink.graph.drivers; import org.apache.commons.lang3.StringEscapeUtils; import org.apache.commons.lang3.text.StrBuilder; @@ -78,7 +78,7 @@ private static String getUsage(String message) { .appendln("usage: JaccardIndex --input --output ") .appendNewLine() .appendln("options:") - .appendln(" --input csv --type --input_filename FILENAME [--input_line_delimiter LINE_DELIMITER] [--input_field_delimiter FIELD_DELIMITER]") + .appendln(" --input csv --type [--simplify ] --input_filename FILENAME [--input_line_delimiter LINE_DELIMITER] [--input_field_delimiter FIELD_DELIMITER]") .appendln(" --input rmat [--scale SCALE] [--edge_factor EDGE_FACTOR]") .appendNewLine() .appendln(" --output print") @@ -116,15 +116,29 @@ public static void main(String[] args) throws Exception { switch (parameters.get("type", "")) { case "integer": { - ji = reader - .keyType(LongValue.class) + Graph graph = reader + .keyType(LongValue.class); + + if (parameters.getBoolean("simplify", false)) { + graph = graph + .run(new org.apache.flink.graph.asm.simple.undirected.Simplify(false)); + } + + ji = graph .run(new org.apache.flink.graph.library.similarity.JaccardIndex() .setLittleParallelism(little_parallelism)); } break; case "string": { - ji = reader - .keyType(StringValue.class) + Graph graph = reader + .keyType(StringValue.class); + + if (parameters.getBoolean("simplify", false)) { + graph = graph + .run(new org.apache.flink.graph.asm.simple.directed.Simplify()); + } + + ji = graph .run(new org.apache.flink.graph.library.similarity.JaccardIndex() .setLittleParallelism(little_parallelism)); } break; @@ -193,7 +207,7 @@ public static void main(String[] args) throws Exception { ji.writeAsCsv(filename, lineDelimiter, fieldDelimiter); - env.execute(); + env.execute("Jaccard Index"); break; default: diff --git a/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/examples/TriangleListing.java b/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/TriangleListing.java similarity index 87% rename from flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/examples/TriangleListing.java rename to flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/TriangleListing.java index 43c5ebaaeae23..0b1e48f16c740 100644 --- a/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/examples/TriangleListing.java +++ b/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/TriangleListing.java @@ -16,7 +16,7 @@ * limitations under the License. */ -package org.apache.flink.graph.examples; +package org.apache.flink.graph.drivers; import org.apache.commons.lang3.StringEscapeUtils; import org.apache.commons.lang3.text.WordUtils; @@ -54,11 +54,11 @@ */ public class TriangleListing { - public static final int DEFAULT_SCALE = 10; + private static final int DEFAULT_SCALE = 10; - public static final int DEFAULT_EDGE_FACTOR = 16; + private static final int DEFAULT_EDGE_FACTOR = 16; - public static final boolean DEFAULT_CLIP_AND_FLIP = true; + private static final boolean DEFAULT_CLIP_AND_FLIP = true; private static void printUsage() { System.out.println(WordUtils.wrap("Lists all triangles in a graph.", 80)); @@ -69,7 +69,7 @@ private static void printUsage() { System.out.println("usage: TriangleListing --directed --input --output "); System.out.println(); System.out.println("options:"); - System.out.println(" --input csv --type --input_filename FILENAME [--input_line_delimiter LINE_DELIMITER] [--input_field_delimiter FIELD_DELIMITER]"); + System.out.println(" --input csv --type [--simplify ] --input_filename FILENAME [--input_line_delimiter LINE_DELIMITER] [--input_field_delimiter FIELD_DELIMITER]"); System.out.println(" --input rmat [--scale SCALE] [--edge_factor EDGE_FACTOR]"); System.out.println(); System.out.println(" --output print"); @@ -111,9 +111,19 @@ public static void main(String[] args) throws Exception { .keyType(LongValue.class); if (directedAlgorithm) { + if (parameters.getBoolean("simplify", false)) { + graph = graph + .run(new org.apache.flink.graph.asm.simple.directed.Simplify()); + } + tl = graph .run(new org.apache.flink.graph.library.clustering.directed.TriangleListing()); } else { + if (parameters.getBoolean("simplify", false)) { + graph = graph + .run(new org.apache.flink.graph.asm.simple.undirected.Simplify(false)); + } + tl = graph .run(new org.apache.flink.graph.library.clustering.undirected.TriangleListing()); } @@ -124,9 +134,19 @@ public static void main(String[] args) throws Exception { .keyType(StringValue.class); if (directedAlgorithm) { + if (parameters.getBoolean("simplify", false)) { + graph = graph + .run(new org.apache.flink.graph.asm.simple.directed.Simplify()); + } + tl = graph .run(new org.apache.flink.graph.library.clustering.directed.TriangleListing()); } else { + if (parameters.getBoolean("simplify", false)) { + graph = graph + .run(new org.apache.flink.graph.asm.simple.undirected.Simplify(false)); + } + tl = graph .run(new org.apache.flink.graph.library.clustering.undirected.TriangleListing()); } From 08e26ab43337f570e08de66b6022cd2a3970bd41 Mon Sep 17 00:00:00 2001 From: Greg Hogan Date: Fri, 21 Oct 2016 13:10:02 -0400 Subject: [PATCH 2/4] Remove GraphMetrics Java and Scala examples. The GraphMetrics driver computes additional metrics and in a single pass. --- .../java/org/apache/flink/graph/Usage.java | 1 - .../flink/graph/examples/GraphMetrics.java | 171 ------------------ .../graph/examples/utils/ExampleUtils.java | 162 ----------------- .../graph/scala/examples/GraphMetrics.scala | 129 ------------- .../library/metric/directed/EdgeMetrics.java | 10 + .../metric/directed/VertexMetrics.java | 10 + .../metric/undirected/EdgeMetrics.java | 10 + .../metric/undirected/VertexMetrics.java | 10 + 8 files changed, 40 insertions(+), 463 deletions(-) delete mode 100644 flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/examples/GraphMetrics.java delete mode 100644 flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/examples/utils/ExampleUtils.java delete mode 100644 flink-libraries/flink-gelly-examples/src/main/scala/org/apache/flink/graph/scala/examples/GraphMetrics.scala diff --git a/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/Usage.java b/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/Usage.java index f10377f0f9ae9..9d8f11685fc68 100644 --- a/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/Usage.java +++ b/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/Usage.java @@ -41,7 +41,6 @@ public class Usage { org.apache.flink.graph.examples.PregelSSSP.class, org.apache.flink.graph.examples.SingleSourceShortestPaths.class, org.apache.flink.graph.scala.examples.ConnectedComponents.class, - org.apache.flink.graph.scala.examples.GraphMetrics.class, org.apache.flink.graph.scala.examples.GSASingleSourceShortestPaths.class, org.apache.flink.graph.scala.examples.SingleSourceShortestPaths.class, }; diff --git a/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/examples/GraphMetrics.java b/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/examples/GraphMetrics.java deleted file mode 100644 index e7b47bf06cb9a..0000000000000 --- a/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/examples/GraphMetrics.java +++ /dev/null @@ -1,171 +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.examples; - -import org.apache.flink.api.common.ProgramDescription; -import org.apache.flink.api.common.functions.MapFunction; -import org.apache.flink.api.java.DataSet; -import org.apache.flink.api.java.ExecutionEnvironment; -import org.apache.flink.api.java.aggregation.Aggregations; -import org.apache.flink.api.java.tuple.Tuple2; -import org.apache.flink.graph.Edge; -import org.apache.flink.graph.Graph; -import org.apache.flink.graph.examples.utils.ExampleUtils; -import org.apache.flink.types.LongValue; -import org.apache.flink.types.NullValue; - -/** - * This example illustrates how to use Gelly metrics methods and get simple statistics - * from the input graph. - * - * The program creates a random graph and computes and prints - * the following metrics: - * - number of vertices - * - number of edges - * - average node degree - * - the vertex ids with the max/min in- and out-degrees - * - * The input file is expected to contain one edge per line, - * with long IDs and no values, in the following format: - * "<sourceVertexID>\t<targetVertexID>". - * If no arguments are provided, the example runs with a random graph of 100 vertices. - * - */ -public class GraphMetrics implements ProgramDescription { - - public static void main(String[] args) throws Exception { - - if (!parseParameters(args)) { - return; - } - - ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment(); - - /** create the graph **/ - Graph graph = Graph.fromDataSet(getEdgesDataSet(env), env); - - /** get the number of vertices **/ - long numVertices = graph.numberOfVertices(); - - /** get the number of edges **/ - long numEdges = graph.numberOfEdges(); - - /** compute the average node degree **/ - DataSet> verticesWithDegrees = graph.getDegrees(); - - DataSet avgNodeDegree = verticesWithDegrees - .aggregate(Aggregations.SUM, 1).map(new AvgNodeDegreeMapper(numVertices)); - - /** find the vertex with the maximum in-degree **/ - DataSet maxInDegreeVertex = graph.inDegrees().maxBy(1).map(new ProjectVertexId()); - - /** find the vertex with the minimum in-degree **/ - DataSet minInDegreeVertex = graph.inDegrees().minBy(1).map(new ProjectVertexId()); - - /** find the vertex with the maximum out-degree **/ - DataSet maxOutDegreeVertex = graph.outDegrees().maxBy(1).map(new ProjectVertexId()); - - /** find the vertex with the minimum out-degree **/ - DataSet minOutDegreeVertex = graph.outDegrees().minBy(1).map(new ProjectVertexId()); - - /** print the results **/ - ExampleUtils.printResult(env.fromElements(numVertices), "Total number of vertices"); - ExampleUtils.printResult(env.fromElements(numEdges), "Total number of edges"); - ExampleUtils.printResult(avgNodeDegree, "Average node degree"); - ExampleUtils.printResult(maxInDegreeVertex, "Vertex with Max in-degree"); - ExampleUtils.printResult(minInDegreeVertex, "Vertex with Min in-degree"); - ExampleUtils.printResult(maxOutDegreeVertex, "Vertex with Max out-degree"); - ExampleUtils.printResult(minOutDegreeVertex, "Vertex with Min out-degree"); - - env.execute(); - } - - @SuppressWarnings("serial") - private static final class AvgNodeDegreeMapper implements MapFunction, Double> { - - private long numberOfVertices; - - public AvgNodeDegreeMapper(long numberOfVertices) { - this.numberOfVertices = numberOfVertices; - } - - public Double map(Tuple2 sumTuple) { - return (double) (sumTuple.f1.getValue() / numberOfVertices) ; - } - } - - @SuppressWarnings("serial") - private static final class ProjectVertexId implements MapFunction, Long> { - public Long map(Tuple2 value) { return value.f0; } - } - - @Override - public String getDescription() { - return "Graph Metrics Example"; - } - - // ****************************************************************************************************************** - // UTIL METHODS - // ****************************************************************************************************************** - - private static boolean fileOutput = false; - - private static String edgesInputPath = null; - - static final int NUM_VERTICES = 100; - - static final long SEED = 9876; - - private static boolean parseParameters(String[] args) { - - if(args.length > 0) { - if(args.length != 1) { - System.err.println("Usage: GraphMetrics "); - return false; - } - - fileOutput = true; - edgesInputPath = args[0]; - } else { - System.out.println("Executing Graph Metrics example with default parameters and built-in default data."); - System.out.println(" Provide parameters to read input data from files."); - System.out.println(" See the documentation for the correct format of input files."); - System.out.println("Usage: GraphMetrics "); - } - return true; - } - - @SuppressWarnings("serial") - private static DataSet> getEdgesDataSet(ExecutionEnvironment env) { - if (fileOutput) { - return env.readCsvFile(edgesInputPath) - .lineDelimiter("\n").fieldDelimiter("\t") - .types(Long.class, Long.class).map( - new MapFunction, Edge>() { - - public Edge map(Tuple2 value) { - return new Edge(value.f0, value.f1, - NullValue.getInstance()); - } - }); - } else { - return ExampleUtils.getRandomEdges(env, NUM_VERTICES); - } - } -} diff --git a/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/examples/utils/ExampleUtils.java b/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/examples/utils/ExampleUtils.java deleted file mode 100644 index b1bc8312e9ca1..0000000000000 --- a/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/examples/utils/ExampleUtils.java +++ /dev/null @@ -1,162 +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.examples.utils; - -import java.io.PrintStream; -import java.util.ArrayList; -import java.util.List; - -import org.apache.flink.api.common.functions.FlatMapFunction; -import org.apache.flink.api.common.functions.MapFunction; -import org.apache.flink.api.common.io.OutputFormat; -import org.apache.flink.api.java.DataSet; -import org.apache.flink.api.java.ExecutionEnvironment; -import org.apache.flink.configuration.Configuration; -import org.apache.flink.graph.Edge; -import org.apache.flink.graph.Vertex; -import org.apache.flink.types.NullValue; -import org.apache.flink.util.Collector; - -public class ExampleUtils { - - @SuppressWarnings({ "serial", "unchecked", "rawtypes" }) - public static void printResult(DataSet set, String msg) { - set.output(new PrintingOutputFormatWithMessage(msg) { - }); - } - - public static class PrintingOutputFormatWithMessage implements - OutputFormat { - - private static final long serialVersionUID = 1L; - - private transient PrintStream stream; - - private transient String prefix; - - private String message; - - // -------------------------------------------------------------------------------------------- - - /** - * Instantiates a printing output format that prints to standard out. - */ - public PrintingOutputFormatWithMessage() { - } - - public PrintingOutputFormatWithMessage(String msg) { - this.message = msg; - } - - @Override - public void open(int taskNumber, int numTasks) { - // get the target stream - this.stream = System.out; - - // set the prefix to message - this.prefix = message + ": "; - } - - @Override - public void writeRecord(T record) { - if (this.prefix != null) { - this.stream.println(this.prefix + record.toString()); - } else { - this.stream.println(record.toString()); - } - } - - @Override - public void close() { - this.stream = null; - this.prefix = null; - } - - @Override - public String toString() { - return "Print to System.out"; - } - - @Override - public void configure(Configuration parameters) { - } - } - - @SuppressWarnings("serial") - public static DataSet> getVertexIds( - ExecutionEnvironment env, final long numVertices) { - return env.generateSequence(1, numVertices).map( - new MapFunction>() { - public Vertex map(Long l) { - return new Vertex(l, NullValue - .getInstance()); - } - }); - } - - @SuppressWarnings("serial") - public static DataSet> getRandomEdges( - ExecutionEnvironment env, final long numVertices) { - return env.generateSequence(1, numVertices).flatMap( - new FlatMapFunction>() { - @Override - public void flatMap(Long key, Collector> out) throws Exception { - int numOutEdges = (int) (Math.random() * (numVertices / 2)); - for (int i = 0; i < numOutEdges; i++) { - long target = (long) (Math.random() * numVertices) + 1; - out.collect(new Edge(key, target, - NullValue.getInstance())); - } - } - }); - } - - public static DataSet> getLongDoubleVertexData( - ExecutionEnvironment env) { - List> vertices = new ArrayList>(); - vertices.add(new Vertex(1L, 1.0)); - vertices.add(new Vertex(2L, 2.0)); - vertices.add(new Vertex(3L, 3.0)); - vertices.add(new Vertex(4L, 4.0)); - vertices.add(new Vertex(5L, 5.0)); - - return env.fromCollection(vertices); - } - - public static DataSet> getLongDoubleEdgeData( - ExecutionEnvironment env) { - List> edges = new ArrayList>(); - edges.add(new Edge(1L, 2L, 12.0)); - edges.add(new Edge(1L, 3L, 13.0)); - edges.add(new Edge(2L, 3L, 23.0)); - edges.add(new Edge(3L, 4L, 34.0)); - edges.add(new Edge(3L, 5L, 35.0)); - edges.add(new Edge(4L, 5L, 45.0)); - edges.add(new Edge(5L, 1L, 51.0)); - - return env.fromCollection(edges); - } - - /** - * Private constructor to prevent instantiation. - */ - private ExampleUtils() { - throw new RuntimeException(); - } -} diff --git a/flink-libraries/flink-gelly-examples/src/main/scala/org/apache/flink/graph/scala/examples/GraphMetrics.scala b/flink-libraries/flink-gelly-examples/src/main/scala/org/apache/flink/graph/scala/examples/GraphMetrics.scala deleted file mode 100644 index ebf43d4174321..0000000000000 --- a/flink-libraries/flink-gelly-examples/src/main/scala/org/apache/flink/graph/scala/examples/GraphMetrics.scala +++ /dev/null @@ -1,129 +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.scala.examples - -import org.apache.flink.api.scala._ -import org.apache.flink.graph.scala._ -import org.apache.flink.types.NullValue -import org.apache.flink.graph.Edge -import org.apache.flink.util.Collector - -/** - * This example illustrates how to use Gelly metrics methods and get simple statistics - * from the input graph. - * - * The program creates a random graph and computes and prints - * the following metrics: - * - number of vertices - * - number of edges - * - average node degree - * - the vertex ids with the max/min in- and out-degrees - * - * The input file is expected to contain one edge per line, - * with long IDs and no values, in the following format: - * {{{ - * \t - * }}} - * If no arguments are provided, the example runs with a random graph of 100 vertices. - * - */ -object GraphMetrics { - def main(args: Array[String]) { - if (!parseParameters(args)) { - return - } - - val env = ExecutionEnvironment.getExecutionEnvironment - /** create the graph **/ - val graph: Graph[Long, NullValue, NullValue] = Graph.fromDataSet(getEdgeDataSet(env), env) - - /** get the number of vertices **/ - val numVertices = graph.numberOfVertices - - /** get the number of edges **/ - val numEdges = graph.numberOfEdges - - /** compute the average node degree **/ - val verticesWithDegrees = graph.getDegrees - val avgDegree = verticesWithDegrees.sum(1).map(in => (in._2.getValue / numVertices).toDouble) - - /** find the vertex with the maximum in-degree **/ - val maxInDegreeVertex = graph.inDegrees.max(1).map(in => in._1) - - /** find the vertex with the minimum in-degree **/ - val minInDegreeVertex = graph.inDegrees.min(1).map(in => in._1) - - /** find the vertex with the maximum out-degree **/ - val maxOutDegreeVertex = graph.outDegrees.max(1).map(in => in._1) - - /** find the vertex with the minimum out-degree **/ - val minOutDegreeVertex = graph.outDegrees.min(1).map(in => in._1) - - /** print the results **/ - env.fromElements(numVertices).printOnTaskManager("Total number of vertices") - env.fromElements(numEdges).printOnTaskManager("Total number of edges") - avgDegree.printOnTaskManager("Average node degree") - maxInDegreeVertex.printOnTaskManager("Vertex with Max in-degree") - minInDegreeVertex.printOnTaskManager("Vertex with Max in-degree") - maxOutDegreeVertex.printOnTaskManager("Vertex with Max out-degree") - minOutDegreeVertex.printOnTaskManager("Vertex with Max out-degree") - - } - - private def parseParameters(args: Array[String]): Boolean = { - if (args.length > 0) { - fileOutput = true - if (args.length == 1) { - edgesPath = args(0) - true - } else { - System.err.println("Usage: GraphMetrics ") - false - } - } else { - System.out.println("Executing GraphMetrics example with built-in default data.") - System.out.println(" Provide parameters to read input data from a file.") - System.out.println(" Usage: GraphMetrics ") - true - } - } - - private def getEdgeDataSet(env: ExecutionEnvironment): DataSet[Edge[Long, NullValue]] = { - if (fileOutput) { - env.readCsvFile[(Long, Long)]( - edgesPath, - fieldDelimiter = "\t").map( - in => new Edge[Long, NullValue](in._1, in._2, NullValue.getInstance())) - } else { - env.generateSequence(1, numVertices).flatMap[Edge[Long, NullValue]]( - (key: Long, out: Collector[Edge[Long, NullValue]]) => { - val numOutEdges: Int = (Math.random() * (numVertices / 2)).toInt - for ( i <- 0 to numOutEdges ) { - val target: Long = ((Math.random() * numVertices) + 1).toLong - new Edge[Long, NullValue](key, target, NullValue.getInstance()) - } - }) - } - } - - private var fileOutput: Boolean = false - private var edgesPath: String = null - private var outputPath: String = null - private val numVertices = 100 -} diff --git a/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/library/metric/directed/EdgeMetrics.java b/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/library/metric/directed/EdgeMetrics.java index 167e31c1882dd..b3e1e30f58a27 100644 --- a/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/library/metric/directed/EdgeMetrics.java +++ b/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/library/metric/directed/EdgeMetrics.java @@ -366,6 +366,15 @@ public long getNumberOfEdges() { return edgeCount; } + /** + * Get the average degree. + * + * @return average degree + */ + public float getAverageDegree() { + return edgeCount / (float)vertexCount; + } + /** * Get the number of triangle triplets. * @@ -453,6 +462,7 @@ public String toString() { return "vertex count: " + nf.format(vertexCount) + "; edge count: " + nf.format(edgeCount) + + "; average degree: " + nf.format(getAverageDegree()) + "; triangle triplet count: " + nf.format(triangleTripletCount) + "; rectangle triplet count: " + nf.format(rectangleTripletCount) + "; triplet count: " + nf.format(tripletCount) diff --git a/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/library/metric/directed/VertexMetrics.java b/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/library/metric/directed/VertexMetrics.java index 22f7733909706..909eea534170b 100644 --- a/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/library/metric/directed/VertexMetrics.java +++ b/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/library/metric/directed/VertexMetrics.java @@ -227,6 +227,15 @@ public long getNumberOfEdges() { return edgeCount; } + /** + * Get the average degree. + * + * @return average degree + */ + public float getAverageDegree() { + return edgeCount / (float)vertexCount; + } + /** * Get the number of triplets. * @@ -278,6 +287,7 @@ public String toString() { return "vertex count: " + nf.format(vertexCount) + "; edge count: " + nf.format(edgeCount) + + "; average degree: " + nf.format(getAverageDegree()) + "; triplet count: " + nf.format(tripletCount) + "; maximum degree: " + nf.format(maximumDegree) + "; maximum out degree: " + nf.format(maximumOutDegree) diff --git a/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/library/metric/undirected/EdgeMetrics.java b/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/library/metric/undirected/EdgeMetrics.java index 1d5b66402337e..6bce42c9ef303 100644 --- a/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/library/metric/undirected/EdgeMetrics.java +++ b/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/library/metric/undirected/EdgeMetrics.java @@ -328,6 +328,15 @@ public long getNumberOfEdges() { return edgeCount; } + /** + * Get the average degree. + * + * @return average degree + */ + public float getAverageDegree() { + return edgeCount / (float)vertexCount; + } + /** * Get the number of triangle triplets. * @@ -397,6 +406,7 @@ public String toString() { return "vertex count: " + nf.format(vertexCount) + "; edge count: " + nf.format(edgeCount) + + "; average degree: " + nf.format(getAverageDegree()) + "; triangle triplet count: " + nf.format(triangleTripletCount) + "; rectangle triplet count: " + nf.format(rectangleTripletCount) + "; triplet count: " + nf.format(tripletCount) diff --git a/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/library/metric/undirected/VertexMetrics.java b/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/library/metric/undirected/VertexMetrics.java index d04fa7bfe1af5..80126056fa1ba 100644 --- a/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/library/metric/undirected/VertexMetrics.java +++ b/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/library/metric/undirected/VertexMetrics.java @@ -229,6 +229,15 @@ public long getNumberOfEdges() { return edgeCount; } + /** + * Get the average degree. + * + * @return average degree + */ + public float getAverageDegree() { + return edgeCount / (float)vertexCount; + } + /** * Get the number of triplets. * @@ -262,6 +271,7 @@ public String toString() { return "vertex count: " + nf.format(vertexCount) + "; edge count: " + nf.format(edgeCount) + + "; average degree: " + nf.format(getAverageDegree()) + "; triplet count: " + nf.format(tripletCount) + "; maximum degree: " + nf.format(maximumDegree) + "; maximum triplets: " + nf.format(maximumTriplets); From 760d879f85e03537fdd3823852a5208e4cd8545b Mon Sep 17 00:00:00 2001 From: Greg Hogan Date: Mon, 24 Oct 2016 11:46:47 -0400 Subject: [PATCH 3/4] Update driver "usage" statements to align with FLINK-4824 --- .../graph/drivers/ClusteringCoefficient.java | 55 ++++++++++--------- .../apache/flink/graph/drivers/Graph500.java | 50 ++++++++++------- .../flink/graph/drivers/GraphMetrics.java | 32 ++++++----- .../org/apache/flink/graph/drivers/HITS.java | 44 ++++++++------- .../flink/graph/drivers/JaccardIndex.java | 6 +- .../flink/graph/drivers/TriangleListing.java | 49 +++++++++-------- 6 files changed, 127 insertions(+), 109 deletions(-) diff --git a/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/ClusteringCoefficient.java b/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/ClusteringCoefficient.java index 2d4ba7d55ad58..18b0406e9db4d 100644 --- a/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/ClusteringCoefficient.java +++ b/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/ClusteringCoefficient.java @@ -19,6 +19,7 @@ package org.apache.flink.graph.drivers; import org.apache.commons.lang3.StringEscapeUtils; +import org.apache.commons.lang3.text.StrBuilder; import org.apache.commons.lang3.text.WordUtils; import org.apache.commons.math3.random.JDKRandomGenerator; import org.apache.flink.api.common.JobExecutionResult; @@ -27,11 +28,12 @@ import org.apache.flink.api.java.io.CsvOutputFormat; import org.apache.flink.api.java.utils.DataSetUtils; import org.apache.flink.api.java.utils.ParameterTool; +import org.apache.flink.client.program.ProgramParametrizationException; import org.apache.flink.graph.Graph; import org.apache.flink.graph.GraphAnalytic; import org.apache.flink.graph.GraphCsvReader; -import org.apache.flink.graph.asm.translate.translators.LongValueToUnsignedIntValue; import org.apache.flink.graph.asm.translate.TranslateGraphIds; +import org.apache.flink.graph.asm.translate.translators.LongValueToUnsignedIntValue; import org.apache.flink.graph.generator.RMatGraph; import org.apache.flink.graph.generator.random.JDKRandomGeneratorFactory; import org.apache.flink.graph.generator.random.RandomGenerableFactory; @@ -65,24 +67,27 @@ public class ClusteringCoefficient { private static final boolean DEFAULT_CLIP_AND_FLIP = true; - private static void printUsage() { - System.out.println(WordUtils.wrap("The local clustering coefficient measures the connectedness of each" + - " vertex's neighborhood and the global clustering coefficient measures the connectedness of the graph." + - " Scores range from 0.0 (no edges between neighbors or vertices) to 1.0 (neighborhood or graph" + - " is a clique).", 80)); - System.out.println(); - System.out.println(WordUtils.wrap("This algorithm returns tuples containing the vertex ID, the degree of" + - " the vertex, and the number of edges between vertex neighbors.", 80)); - System.out.println(); - System.out.println("usage: ClusteringCoefficient --directed --input --output "); - System.out.println(); - System.out.println("options:"); - System.out.println(" --input csv --type [--simplify ] --input_filename FILENAME [--input_line_delimiter LINE_DELIMITER] [--input_field_delimiter FIELD_DELIMITER]"); - System.out.println(" --input rmat [--scale SCALE] [--edge_factor EDGE_FACTOR]"); - System.out.println(); - System.out.println(" --output print"); - System.out.println(" --output hash"); - System.out.println(" --output csv --output_filename FILENAME [--output_line_delimiter LINE_DELIMITER] [--output_field_delimiter FIELD_DELIMITER]"); + private static String getUsage(String message) { + return new StrBuilder() + .appendNewLine() + .appendln(WordUtils.wrap("The local clustering coefficient measures the connectedness of each" + + " vertex's neighborhood and the global clustering coefficient measures the connectedness of the graph." + + " Scores range from 0.0 (no edges between neighbors or vertices) to 1.0 (neighborhood or graph" + + " is a clique).", 80)) + .appendNewLine() + .appendln(WordUtils.wrap("This algorithm returns tuples containing the vertex ID, the degree of" + + " the vertex, and the number of edges between vertex neighbors.", 80)) + .appendNewLine() + .appendln("usage: ClusteringCoefficient --directed --input --output ") + .appendNewLine() + .appendln("options:") + .appendln(" --input csv --type [--simplify ] --input_filename FILENAME [--input_line_delimiter LINE_DELIMITER] [--input_field_delimiter FIELD_DELIMITER]") + .appendln(" --input rmat [--scale SCALE] [--edge_factor EDGE_FACTOR]") + .appendNewLine() + .appendln(" --output print") + .appendln(" --output hash") + .appendln(" --output csv --output_filename FILENAME [--output_line_delimiter LINE_DELIMITER] [--output_field_delimiter FIELD_DELIMITER]") + .toString(); } public static void main(String[] args) throws Exception { @@ -93,8 +98,7 @@ public static void main(String[] args) throws Exception { ParameterTool parameters = ParameterTool.fromArgs(args); if (! parameters.has("directed")) { - printUsage(); - return; + throw new ProgramParametrizationException(getUsage("must declare execution mode as '--directed true' or '--directed false'")); } boolean directedAlgorithm = parameters.getBoolean("directed"); @@ -195,8 +199,7 @@ public static void main(String[] args) throws Exception { } break; default: - printUsage(); - return; + throw new ProgramParametrizationException(getUsage("invalid CSV type")); } } break; @@ -287,8 +290,7 @@ public static void main(String[] args) throws Exception { } break; default: - printUsage(); - return; + throw new ProgramParametrizationException(getUsage("invalid input type")); } switch (parameters.get("output", "")) { @@ -327,8 +329,7 @@ public static void main(String[] args) throws Exception { break; default: - printUsage(); - return; + throw new ProgramParametrizationException(getUsage("invalid output type")); } System.out.println(gcc.getResult()); diff --git a/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/Graph500.java b/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/Graph500.java index d0c8717a7ea02..8f9a54a3a3741 100644 --- a/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/Graph500.java +++ b/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/Graph500.java @@ -18,6 +18,9 @@ package org.apache.flink.graph.drivers; +import org.apache.commons.lang3.StringEscapeUtils; +import org.apache.commons.lang3.text.StrBuilder; +import org.apache.commons.lang3.text.WordUtils; import org.apache.commons.math3.random.JDKRandomGenerator; import org.apache.flink.api.common.JobExecutionResult; import org.apache.flink.api.java.DataSet; @@ -26,6 +29,7 @@ import org.apache.flink.api.java.tuple.Tuple2; import org.apache.flink.api.java.utils.DataSetUtils; import org.apache.flink.api.java.utils.ParameterTool; +import org.apache.flink.client.program.ProgramParametrizationException; import org.apache.flink.graph.Graph; import org.apache.flink.graph.asm.simple.undirected.Simplify; import org.apache.flink.graph.generator.RMatGraph; @@ -53,6 +57,25 @@ public class Graph500 { private static final boolean DEFAULT_CLIP_AND_FLIP = true; + private static String getUsage(String message) { + return new StrBuilder() + .appendNewLine() + .appendln("A Graph500 generator using the Recursive Matrix (RMat) graph generator.") + .appendNewLine() + .appendln(WordUtils.wrap("The graph matrix contains 2^scale vertices although not every vertex will" + + " be represented in an edge. The number of edges is edge_factor * 2^scale edges" + + " although some edges may be duplicates.", 80)) + .appendNewLine() + .appendln("Note: this does not yet implement permutation of vertex labels or edges.") + .appendNewLine() + .appendln(" --output print") + .appendln(" --output hash") + .appendln(" --output csv --output_filename FILENAME [--output_line_delimiter LINE_DELIMITER] [--output_field_delimiter FIELD_DELIMITER]") + .appendNewLine() + .appendln("Usage error: " + message) + .toString(); + } + public static void main(String[] args) throws Exception { // Set up the execution environment final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment(); @@ -96,29 +119,18 @@ public static void main(String[] args) throws Exception { case "csv": String filename = parameters.get("filename"); - String row_delimiter = parameters.get("row_delimiter", CsvOutputFormat.DEFAULT_LINE_DELIMITER); - String field_delimiter = parameters.get("field_delimiter", CsvOutputFormat.DEFAULT_FIELD_DELIMITER); + String lineDelimiter = StringEscapeUtils.unescapeJava( + parameters.get("output_line_delimiter", CsvOutputFormat.DEFAULT_LINE_DELIMITER)); + + String fieldDelimiter = StringEscapeUtils.unescapeJava( + parameters.get("output_field_delimiter", CsvOutputFormat.DEFAULT_FIELD_DELIMITER)); - edges.writeAsCsv(filename, row_delimiter, field_delimiter); + edges.writeAsCsv(filename, lineDelimiter, fieldDelimiter); - env.execute(); + env.execute("Graph500"); break; default: - System.out.println("A Graph500 generator using the Recursive Matrix (RMat) graph generator."); - System.out.println(); - System.out.println("The graph matrix contains 2^scale vertices although not every vertex will"); - System.out.println("be represented in an edge. The number of edges is edge_factor * 2^scale edges"); - System.out.println("although some edges may be duplicates."); - System.out.println(); - System.out.println("Note: this does not yet implement permutation of vertex labels or edges."); - System.out.println(); - System.out.println("usage:"); - System.out.println(" Graph500 [--scale SCALE] [--edge_factor EDGE_FACTOR] --output print"); - System.out.println(" Graph500 [--scale SCALE] [--edge_factor EDGE_FACTOR] --output hash"); - System.out.println(" Graph500 [--scale SCALE] [--edge_factor EDGE_FACTOR] --output csv" + - " --filename FILENAME [--row_delimiter ROW_DELIMITER] [--field_delimiter FIELD_DELIMITER]"); - - return; + throw new ProgramParametrizationException(getUsage("invalid output type")); } JobExecutionResult result = env.getLastJobExecutionResult(); diff --git a/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/GraphMetrics.java b/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/GraphMetrics.java index e1e06aefdf3a1..4fb11c3ac44c7 100644 --- a/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/GraphMetrics.java +++ b/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/GraphMetrics.java @@ -19,17 +19,19 @@ package org.apache.flink.graph.drivers; import org.apache.commons.lang3.StringEscapeUtils; +import org.apache.commons.lang3.text.StrBuilder; import org.apache.commons.lang3.text.WordUtils; import org.apache.commons.math3.random.JDKRandomGenerator; import org.apache.flink.api.common.JobExecutionResult; import org.apache.flink.api.java.ExecutionEnvironment; import org.apache.flink.api.java.io.CsvOutputFormat; import org.apache.flink.api.java.utils.ParameterTool; +import org.apache.flink.client.program.ProgramParametrizationException; import org.apache.flink.graph.Graph; import org.apache.flink.graph.GraphAnalytic; import org.apache.flink.graph.GraphCsvReader; -import org.apache.flink.graph.asm.translate.translators.LongValueToUnsignedIntValue; import org.apache.flink.graph.asm.translate.TranslateGraphIds; +import org.apache.flink.graph.asm.translate.translators.LongValueToUnsignedIntValue; import org.apache.flink.graph.generator.RMatGraph; import org.apache.flink.graph.generator.random.JDKRandomGeneratorFactory; import org.apache.flink.graph.generator.random.RandomGenerableFactory; @@ -56,14 +58,17 @@ public class GraphMetrics { private static final boolean DEFAULT_CLIP_AND_FLIP = true; - private static void printUsage() { - System.out.println(WordUtils.wrap("Computes vertex and edge metrics on a directed or undirected graph.", 80)); - System.out.println(); - System.out.println("usage: GraphMetrics --directed --input "); - System.out.println(); - System.out.println("options:"); - System.out.println(" --input csv --type [--simplify ] --input_filename FILENAME [--input_line_delimiter LINE_DELIMITER] [--input_field_delimiter FIELD_DELIMITER]"); - System.out.println(" --input rmat [--scale SCALE] [--edge_factor EDGE_FACTOR]"); + private static String getUsage(String message) { + return new StrBuilder() + .appendNewLine() + .appendln(WordUtils.wrap("Computes vertex and edge metrics on a directed or undirected graph.", 80)) + .appendNewLine() + .appendln("usage: GraphMetrics --directed --input ") + .appendNewLine() + .appendln("options:") + .appendln(" --input csv --type [--simplify ] --input_filename FILENAME [--input_line_delimiter LINE_DELIMITER] [--input_field_delimiter FIELD_DELIMITER]") + .appendln(" --input rmat [--scale SCALE] [--edge_factor EDGE_FACTOR]") + .toString(); } public static void main(String[] args) throws Exception { @@ -73,8 +78,7 @@ public static void main(String[] args) throws Exception { ParameterTool parameters = ParameterTool.fromArgs(args); if (! parameters.has("directed")) { - printUsage(); - return; + throw new ProgramParametrizationException(getUsage("must declare execution mode as '--directed true' or '--directed false'")); } boolean directedAlgorithm = parameters.getBoolean("directed"); @@ -151,8 +155,7 @@ public static void main(String[] args) throws Exception { } break; default: - printUsage(); - return; + throw new ProgramParametrizationException(getUsage("invalid CSV type")); } } break; @@ -213,8 +216,7 @@ public static void main(String[] args) throws Exception { } break; default: - printUsage(); - return; + throw new ProgramParametrizationException(getUsage("invalid input type")); } env.execute("Graph Metrics"); diff --git a/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/HITS.java b/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/HITS.java index fe792437450c4..e0a233ab7ada2 100644 --- a/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/HITS.java +++ b/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/HITS.java @@ -19,6 +19,7 @@ package org.apache.flink.graph.drivers; import org.apache.commons.lang3.StringEscapeUtils; +import org.apache.commons.lang3.text.StrBuilder; import org.apache.commons.lang3.text.WordUtils; import org.apache.commons.math3.random.JDKRandomGenerator; import org.apache.flink.api.common.JobExecutionResult; @@ -27,11 +28,12 @@ import org.apache.flink.api.java.io.CsvOutputFormat; import org.apache.flink.api.java.utils.DataSetUtils; import org.apache.flink.api.java.utils.ParameterTool; +import org.apache.flink.client.program.ProgramParametrizationException; import org.apache.flink.graph.Graph; import org.apache.flink.graph.GraphCsvReader; import org.apache.flink.graph.asm.simple.directed.Simplify; -import org.apache.flink.graph.asm.translate.translators.LongValueToUnsignedIntValue; import org.apache.flink.graph.asm.translate.TranslateGraphIds; +import org.apache.flink.graph.asm.translate.translators.LongValueToUnsignedIntValue; import org.apache.flink.graph.generator.RMatGraph; import org.apache.flink.graph.generator.random.JDKRandomGeneratorFactory; import org.apache.flink.graph.generator.random.RandomGenerableFactory; @@ -60,20 +62,23 @@ public class HITS { private static final int DEFAULT_EDGE_FACTOR = 16; - private static void printUsage() { - System.out.println(WordUtils.wrap("Hyperlink-Induced Topic Search computes two interdependent" + - " scores for every vertex in a directed graph. A good \"hub\" links to good \"authorities\"" + - " and good \"authorities\" are linked from good \"hubs\".", 80)); - System.out.println(); - System.out.println("usage: HITS --input --output "); - System.out.println(); - System.out.println("options:"); - System.out.println(" --input csv --type --input_filename FILENAME [--input_line_delimiter LINE_DELIMITER] [--input_field_delimiter FIELD_DELIMITER]"); - System.out.println(" --input rmat [--scale SCALE] [--edge_factor EDGE_FACTOR]"); - System.out.println(); - System.out.println(" --output print"); - System.out.println(" --output hash"); - System.out.println(" --output csv --output_filename FILENAME [--output_line_delimiter LINE_DELIMITER] [--output_field_delimiter FIELD_DELIMITER]"); + private static String getUsage(String message) { + return new StrBuilder() + .appendNewLine() + .appendln(WordUtils.wrap("Hyperlink-Induced Topic Search computes two interdependent" + + " scores for every vertex in a directed graph. A good \"hub\" links to good \"authorities\"" + + " and good \"authorities\" are linked from good \"hubs\".", 80)) + .appendNewLine() + .appendln("usage: HITS --input --output ") + .appendNewLine() + .appendln("options:") + .appendln(" --input csv --type --input_filename FILENAME [--input_line_delimiter LINE_DELIMITER] [--input_field_delimiter FIELD_DELIMITER]") + .appendln(" --input rmat [--scale SCALE] [--edge_factor EDGE_FACTOR]") + .appendNewLine() + .appendln(" --output print") + .appendln(" --output hash") + .appendln(" --output csv --output_filename FILENAME [--output_line_delimiter LINE_DELIMITER] [--output_field_delimiter FIELD_DELIMITER]") + .toString(); } public static void main(String[] args) throws Exception { @@ -114,8 +119,7 @@ public static void main(String[] args) throws Exception { } break; default: - printUsage(); - return; + throw new ProgramParametrizationException(getUsage("invalid CSV type")); } } break; @@ -144,8 +148,7 @@ public static void main(String[] args) throws Exception { } break; default: - printUsage(); - return; + throw new ProgramParametrizationException(getUsage("invalid input type")); } switch (parameters.get("output", "")) { @@ -173,8 +176,7 @@ public static void main(String[] args) throws Exception { env.execute(); break; default: - printUsage(); - return; + throw new ProgramParametrizationException(getUsage("invalid output type")); } JobExecutionResult result = env.getLastJobExecutionResult(); diff --git a/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/JaccardIndex.java b/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/JaccardIndex.java index b9f0b417a3583..5c173e02c1ab2 100644 --- a/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/JaccardIndex.java +++ b/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/JaccardIndex.java @@ -58,11 +58,11 @@ */ public class JaccardIndex { - public static final int DEFAULT_SCALE = 10; + private static final int DEFAULT_SCALE = 10; - public static final int DEFAULT_EDGE_FACTOR = 16; + private static final int DEFAULT_EDGE_FACTOR = 16; - public static final boolean DEFAULT_CLIP_AND_FLIP = true; + private static final boolean DEFAULT_CLIP_AND_FLIP = true; private static String getUsage(String message) { return new StrBuilder() diff --git a/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/TriangleListing.java b/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/TriangleListing.java index 0b1e48f16c740..954f732316416 100644 --- a/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/TriangleListing.java +++ b/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/drivers/TriangleListing.java @@ -19,6 +19,7 @@ package org.apache.flink.graph.drivers; import org.apache.commons.lang3.StringEscapeUtils; +import org.apache.commons.lang3.text.StrBuilder; import org.apache.commons.lang3.text.WordUtils; import org.apache.commons.math3.random.JDKRandomGenerator; import org.apache.flink.api.common.JobExecutionResult; @@ -27,11 +28,12 @@ import org.apache.flink.api.java.io.CsvOutputFormat; import org.apache.flink.api.java.utils.DataSetUtils; import org.apache.flink.api.java.utils.ParameterTool; +import org.apache.flink.client.program.ProgramParametrizationException; import org.apache.flink.graph.Graph; import org.apache.flink.graph.GraphCsvReader; import org.apache.flink.graph.asm.simple.undirected.Simplify; -import org.apache.flink.graph.asm.translate.translators.LongValueToUnsignedIntValue; import org.apache.flink.graph.asm.translate.TranslateGraphIds; +import org.apache.flink.graph.asm.translate.translators.LongValueToUnsignedIntValue; import org.apache.flink.graph.generator.RMatGraph; import org.apache.flink.graph.generator.random.JDKRandomGeneratorFactory; import org.apache.flink.graph.generator.random.RandomGenerableFactory; @@ -60,21 +62,24 @@ public class TriangleListing { private static final boolean DEFAULT_CLIP_AND_FLIP = true; - private static void printUsage() { - System.out.println(WordUtils.wrap("Lists all triangles in a graph.", 80)); - System.out.println(); - System.out.println(WordUtils.wrap("This algorithm returns tuples containing the vertex IDs for each triangle and" + - " for directed graphs a bitmask indicating the presence of the six potential connecting edges.", 80)); - System.out.println(); - System.out.println("usage: TriangleListing --directed --input --output "); - System.out.println(); - System.out.println("options:"); - System.out.println(" --input csv --type [--simplify ] --input_filename FILENAME [--input_line_delimiter LINE_DELIMITER] [--input_field_delimiter FIELD_DELIMITER]"); - System.out.println(" --input rmat [--scale SCALE] [--edge_factor EDGE_FACTOR]"); - System.out.println(); - System.out.println(" --output print"); - System.out.println(" --output hash"); - System.out.println(" --output csv --output_filename FILENAME [--output_line_delimiter LINE_DELIMITER] [--output_field_delimiter FIELD_DELIMITER]"); + private static String getUsage(String message) { + return new StrBuilder() + .appendNewLine() + .appendln(WordUtils.wrap("Lists all triangles in a graph.", 80)) + .appendNewLine() + .appendln(WordUtils.wrap("This algorithm returns tuples containing the vertex IDs for each triangle and" + + " for directed graphs a bitmask indicating the presence of the six potential connecting edges.", 80)) + .appendNewLine() + .appendln("usage: TriangleListing --directed --input --output ") + .appendNewLine() + .appendln("options:") + .appendln(" --input csv --type [--simplify ] --input_filename FILENAME [--input_line_delimiter LINE_DELIMITER] [--input_field_delimiter FIELD_DELIMITER]") + .appendln(" --input rmat [--scale SCALE] [--edge_factor EDGE_FACTOR]") + .appendNewLine() + .appendln(" --output print") + .appendln(" --output hash") + .appendln(" --output csv --output_filename FILENAME [--output_line_delimiter LINE_DELIMITER] [--output_field_delimiter FIELD_DELIMITER]") + .toString(); } public static void main(String[] args) throws Exception { @@ -84,8 +89,7 @@ public static void main(String[] args) throws Exception { ParameterTool parameters = ParameterTool.fromArgs(args); if (! parameters.has("directed")) { - printUsage(); - return; + throw new ProgramParametrizationException(getUsage("must declare execution mode as '--directed true' or '--directed false'")); } boolean directedAlgorithm = parameters.getBoolean("directed"); @@ -153,8 +157,7 @@ public static void main(String[] args) throws Exception { } break; default: - printUsage(); - return; + throw new ProgramParametrizationException(getUsage("invalid CSV type")); } @@ -203,8 +206,7 @@ public static void main(String[] args) throws Exception { } break; default: - printUsage(); - return; + throw new ProgramParametrizationException(getUsage("invalid input type")); } switch (parameters.get("output", "")) { @@ -238,8 +240,7 @@ public static void main(String[] args) throws Exception { env.execute(); break; default: - printUsage(); - return; + throw new ProgramParametrizationException(getUsage("invalid output type")); } JobExecutionResult result = env.getLastJobExecutionResult(); From 673cf2c8b55e0a183f9a4270ea9c09118df51cab Mon Sep 17 00:00:00 2001 From: Greg Hogan Date: Mon, 24 Oct 2016 12:11:49 -0400 Subject: [PATCH 4/4] Small cleanup to web documentation usage for running Gelly examples --- docs/dev/libs/gelly/index.md | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/docs/dev/libs/gelly/index.md b/docs/dev/libs/gelly/index.md index 0d9bdb592e7a8..db7073fbe531d 100644 --- a/docs/dev/libs/gelly/index.md +++ b/docs/dev/libs/gelly/index.md @@ -69,8 +69,8 @@ The remaining sections provide a description of available methods and present se Running Gelly Examples ---------------------- -The Gelly library and examples jars are [provided](https://flink.apache.org/downloads.html "Apache Flink: Downloads") -in the Flink distribution at **opt/lib/gelly** (for versions older than Flink 1.2 these can be manually downloaded from +The Gelly library and examples jars are provided in the [Flink distribution](https://flink.apache.org/downloads.html "Apache Flink: Downloads") +in the folder **opt/lib/gelly** (for versions older than Flink 1.2 these can be manually downloaded from [Maven Central](http://search.maven.org/#search|ga|1|flink%20gelly). To run the Gelly examples the **flink-gelly** (for Java) or **flink-gelly-scala** (for Scala) jar must be copied to @@ -81,7 +81,7 @@ cp opt/lib/gelly/flink-gelly_*.jar lib/ cp opt/lib/gelly/flink-gelly-scala_*.jar lib/ ~~~ -Gelly's example jar includes both drivers for the library methods as well as additional example algorithms. After +Gelly's examples jar includes both drivers for the library methods as well as additional example algorithms. After configuring and starting the cluster, list the available algorithm classes: ~~~bash @@ -121,7 +121,7 @@ wget -O - http://snap.stanford.edu/data/bigdata/communities/com-lj.ungraph.txt.g --output hash ~~~ -Please submit feature requests or issue reports through the user [mailing list](https://flink.apache.org/community.html#mailing-lists) +Please submit feature requests and report issues on the user [mailing list](https://flink.apache.org/community.html#mailing-lists) or [Flink Jira](https://issues.apache.org/jira/browse/FLINK). We welcome suggestions for new algorithms and features as well as [code contributions](https://flink.apache.org/contribute-code.html).