You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by se...@apache.org on 2016/08/25 18:49:02 UTC
[59/89] [abbrv] flink git commit: [FLINK-4264] [gelly] New
GraphMetrics driver
[FLINK-4264] [gelly] New GraphMetrics driver
Updates VertexMetrics analytic, adds directed and undirected
EdgeMetric analytics, and includes a new GraphMetrics driver.
This closes #2295
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/58850f29
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/58850f29
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/58850f29
Branch: refs/heads/flip-6
Commit: 58850f29243559c67bdc07c737bc5772c3f098db
Parents: ad8e665
Author: Greg Hogan <co...@greghogan.com>
Authored: Wed Jul 20 15:44:54 2016 -0400
Committer: Greg Hogan <co...@greghogan.com>
Committed: Wed Aug 24 10:05:12 2016 -0400
----------------------------------------------------------------------
.../apache/flink/graph/driver/GraphMetrics.java | 232 +++++++++
.../graph/examples/ClusteringCoefficient.java | 14 +-
.../org/apache/flink/graph/examples/HITS.java | 2 +-
.../flink/graph/examples/JaccardIndex.java | 2 +-
.../flink/graph/examples/TriangleListing.java | 2 +-
.../degree/annotate/directed/VertexDegrees.java | 2 +-
.../directed/LocalClusteringCoefficient.java | 32 +-
.../undirected/LocalClusteringCoefficient.java | 31 +-
.../library/metric/directed/EdgeMetrics.java | 507 +++++++++++++++++++
.../library/metric/directed/VertexMetrics.java | 102 +++-
.../library/metric/undirected/EdgeMetrics.java | 445 ++++++++++++++++
.../metric/undirected/VertexMetrics.java | 63 ++-
.../metric/directed/EdgeMetricsTest.java | 90 ++++
.../metric/directed/VertexMetricsTest.java | 19 +-
.../metric/undirected/EdgeMetricsTest.java | 89 ++++
.../metric/undirected/VertexMetricsTest.java | 14 +-
16 files changed, 1600 insertions(+), 46 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/58850f29/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/driver/GraphMetrics.java
----------------------------------------------------------------------
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/driver/GraphMetrics.java
new file mode 100644
index 0000000..cc265bb
--- /dev/null
+++ b/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/driver/GraphMetrics.java
@@ -0,0 +1,232 @@
+/*
+ * 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.driver;
+
+import org.apache.commons.lang3.StringEscapeUtils;
+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.graph.Graph;
+import org.apache.flink.graph.GraphAnalytic;
+import org.apache.flink.graph.GraphCsvReader;
+import org.apache.flink.graph.asm.translate.LongValueToIntValue;
+import org.apache.flink.graph.asm.translate.TranslateGraphIds;
+import org.apache.flink.graph.generator.RMatGraph;
+import org.apache.flink.graph.generator.random.JDKRandomGeneratorFactory;
+import org.apache.flink.graph.generator.random.RandomGenerableFactory;
+import org.apache.flink.types.IntValue;
+import org.apache.flink.types.LongValue;
+import org.apache.flink.types.NullValue;
+import org.apache.flink.types.StringValue;
+
+import java.text.NumberFormat;
+
+/**
+ * Computes vertex and edge metrics on a directed or undirected graph.
+ *
+ * @see org.apache.flink.graph.library.metric.directed.EdgeMetrics
+ * @see org.apache.flink.graph.library.metric.directed.VertexMetrics
+ * @see org.apache.flink.graph.library.metric.undirected.EdgeMetrics
+ * @see org.apache.flink.graph.library.metric.undirected.VertexMetrics
+ */
+public class GraphMetrics {
+
+ public static final int DEFAULT_SCALE = 10;
+
+ public static final int DEFAULT_EDGE_FACTOR = 16;
+
+ public 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 <true | false> --input <csv | rmat [options]>");
+ System.out.println();
+ System.out.println("options:");
+ System.out.println(" --input csv --type <integer | string> [--simplify <true | false>] --input_filename FILENAME [--input_line_delimiter LINE_DELIMITER] [--input_field_delimiter FIELD_DELIMITER]");
+ System.out.println(" --input rmat [--scale SCALE] [--edge_factor EDGE_FACTOR]");
+ }
+
+ public static void main(String[] args) throws Exception {
+ // Set up the execution environment
+ final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+ env.getConfig().enableObjectReuse();
+
+ ParameterTool parameters = ParameterTool.fromArgs(args);
+ if (! parameters.has("directed")) {
+ printUsage();
+ return;
+ }
+ boolean directedAlgorithm = parameters.getBoolean("directed");
+
+ GraphAnalytic vm;
+ GraphAnalytic em;
+
+ switch (parameters.get("input", "")) {
+ case "csv": {
+ String lineDelimiter = StringEscapeUtils.unescapeJava(
+ parameters.get("input_line_delimiter", CsvOutputFormat.DEFAULT_LINE_DELIMITER));
+
+ String fieldDelimiter = StringEscapeUtils.unescapeJava(
+ parameters.get("input_field_delimiter", CsvOutputFormat.DEFAULT_FIELD_DELIMITER));
+
+ GraphCsvReader reader = Graph
+ .fromCsvReader(parameters.get("input_filename"), env)
+ .ignoreCommentsEdges("#")
+ .lineDelimiterEdges(lineDelimiter)
+ .fieldDelimiterEdges(fieldDelimiter);
+
+ switch (parameters.get("type", "")) {
+ case "integer": {
+ Graph<LongValue, NullValue, NullValue> graph = reader
+ .keyType(LongValue.class);
+
+ if (directedAlgorithm) {
+ if (parameters.getBoolean("simplify", false)) {
+ graph = graph
+ .run(new org.apache.flink.graph.asm.simple.directed.Simplify<LongValue, NullValue, NullValue>());
+ }
+
+ vm = graph
+ .run(new org.apache.flink.graph.library.metric.directed.VertexMetrics<LongValue, NullValue, NullValue>());
+ em = graph
+ .run(new org.apache.flink.graph.library.metric.directed.EdgeMetrics<LongValue, NullValue, NullValue>());
+ } else {
+ if (parameters.getBoolean("simplify", false)) {
+ graph = graph
+ .run(new org.apache.flink.graph.asm.simple.undirected.Simplify<LongValue, NullValue, NullValue>(false));
+ }
+
+ vm = graph
+ .run(new org.apache.flink.graph.library.metric.undirected.VertexMetrics<LongValue, NullValue, NullValue>());
+ em = graph
+ .run(new org.apache.flink.graph.library.metric.undirected.EdgeMetrics<LongValue, NullValue, NullValue>());
+ }
+ } break;
+
+ case "string": {
+ Graph<StringValue, NullValue, NullValue> graph = reader
+ .keyType(StringValue.class);
+
+ if (directedAlgorithm) {
+ if (parameters.getBoolean("simplify", false)) {
+ graph = graph
+ .run(new org.apache.flink.graph.asm.simple.directed.Simplify<StringValue, NullValue, NullValue>());
+ }
+
+ vm = graph
+ .run(new org.apache.flink.graph.library.metric.directed.VertexMetrics<StringValue, NullValue, NullValue>());
+ em = graph
+ .run(new org.apache.flink.graph.library.metric.directed.EdgeMetrics<StringValue, NullValue, NullValue>());
+ } else {
+ if (parameters.getBoolean("simplify", false)) {
+ graph = graph
+ .run(new org.apache.flink.graph.asm.simple.undirected.Simplify<StringValue, NullValue, NullValue>(false));
+ }
+
+ vm = graph
+ .run(new org.apache.flink.graph.library.metric.undirected.VertexMetrics<StringValue, NullValue, NullValue>());
+ em = graph
+ .run(new org.apache.flink.graph.library.metric.undirected.EdgeMetrics<StringValue, NullValue, NullValue>());
+ }
+ } break;
+
+ default:
+ printUsage();
+ return;
+ }
+ } break;
+
+ case "rmat": {
+ int scale = parameters.getInt("scale", DEFAULT_SCALE);
+ int edgeFactor = parameters.getInt("edge_factor", DEFAULT_EDGE_FACTOR);
+
+ RandomGenerableFactory<JDKRandomGenerator> rnd = new JDKRandomGeneratorFactory();
+
+ long vertexCount = 1L << scale;
+ long edgeCount = vertexCount * edgeFactor;
+
+
+ Graph<LongValue, NullValue, NullValue> graph = new RMatGraph<>(env, rnd, vertexCount, edgeCount)
+ .generate();
+
+ if (directedAlgorithm) {
+ if (scale > 32) {
+ Graph<LongValue, NullValue, NullValue> newGraph = graph
+ .run(new org.apache.flink.graph.asm.simple.directed.Simplify<LongValue, NullValue, NullValue>());
+
+ vm = newGraph
+ .run(new org.apache.flink.graph.library.metric.directed.VertexMetrics<LongValue, NullValue, NullValue>());
+ em = newGraph
+ .run(new org.apache.flink.graph.library.metric.directed.EdgeMetrics<LongValue, NullValue, NullValue>());
+ } else {
+ Graph<IntValue, NullValue, NullValue> newGraph = graph
+ .run(new TranslateGraphIds<LongValue, IntValue, NullValue, NullValue>(new LongValueToIntValue()))
+ .run(new org.apache.flink.graph.asm.simple.directed.Simplify<IntValue, NullValue, NullValue>());
+
+ vm = newGraph
+ .run(new org.apache.flink.graph.library.metric.directed.VertexMetrics<IntValue, NullValue, NullValue>());
+ em = newGraph
+ .run(new org.apache.flink.graph.library.metric.directed.EdgeMetrics<IntValue, NullValue, NullValue>());
+ }
+ } else {
+ boolean clipAndFlip = parameters.getBoolean("clip_and_flip", DEFAULT_CLIP_AND_FLIP);
+
+ if (scale > 32) {
+ Graph<LongValue, NullValue, NullValue> newGraph = graph
+ .run(new org.apache.flink.graph.asm.simple.undirected.Simplify<LongValue, NullValue, NullValue>(clipAndFlip));
+
+ vm = newGraph
+ .run(new org.apache.flink.graph.library.metric.undirected.VertexMetrics<LongValue, NullValue, NullValue>());
+ em = newGraph
+ .run(new org.apache.flink.graph.library.metric.undirected.EdgeMetrics<LongValue, NullValue, NullValue>());
+ } else {
+ Graph<IntValue, NullValue, NullValue> newGraph = graph
+ .run(new TranslateGraphIds<LongValue, IntValue, NullValue, NullValue>(new LongValueToIntValue()))
+ .run(new org.apache.flink.graph.asm.simple.undirected.Simplify<IntValue, NullValue, NullValue>(clipAndFlip));
+
+ vm = newGraph
+ .run(new org.apache.flink.graph.library.metric.undirected.VertexMetrics<IntValue, NullValue, NullValue>());
+ em = newGraph
+ .run(new org.apache.flink.graph.library.metric.undirected.EdgeMetrics<IntValue, NullValue, NullValue>());
+ }
+ }
+ } break;
+
+ default:
+ printUsage();
+ return;
+ }
+
+ env.execute("Graph Metrics");
+
+ System.out.print("Vertex metrics:\n ");
+ System.out.println(vm.getResult().toString().replace(";", "\n "));
+ System.out.print("\nEdge metrics:\n ");
+ System.out.println(em.getResult().toString().replace(";", "\n "));
+
+ JobExecutionResult result = env.getLastJobExecutionResult();
+
+ NumberFormat nf = NumberFormat.getInstance();
+ System.out.println("\nExecution runtime: " + nf.format(result.getNetRuntime()) + " ms");
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/58850f29/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/examples/ClusteringCoefficient.java
----------------------------------------------------------------------
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/examples/ClusteringCoefficient.java
index 8641428..e099e2b 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/examples/ClusteringCoefficient.java
@@ -72,7 +72,7 @@ public class ClusteringCoefficient {
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 <true | false> --input <csv | rmat [options]> --output <print | hash | csv [options]");
+ System.out.println("usage: ClusteringCoefficient --directed <true | false> --input <csv | rmat [options]> --output <print | hash | csv [options]>");
System.out.println();
System.out.println("options:");
System.out.println(" --input csv --type <integer | string> --input_filename FILENAME [--input_line_delimiter LINE_DELIMITER] [--input_field_delimiter FIELD_DELIMITER]");
@@ -174,7 +174,8 @@ public class ClusteringCoefficient {
gcc = newGraph
.run(new org.apache.flink.graph.library.clustering.directed.GlobalClusteringCoefficient<LongValue, NullValue, NullValue>());
lcc = newGraph
- .run(new org.apache.flink.graph.library.clustering.directed.LocalClusteringCoefficient<LongValue, NullValue, NullValue>());
+ .run(new org.apache.flink.graph.library.clustering.directed.LocalClusteringCoefficient<LongValue, NullValue, NullValue>()
+ .setIncludeZeroDegreeVertices(false));
} else {
Graph<IntValue, NullValue, NullValue> newGraph = graph
.run(new TranslateGraphIds<LongValue, IntValue, NullValue, NullValue>(new LongValueToIntValue()))
@@ -183,7 +184,8 @@ public class ClusteringCoefficient {
gcc = newGraph
.run(new org.apache.flink.graph.library.clustering.directed.GlobalClusteringCoefficient<IntValue, NullValue, NullValue>());
lcc = newGraph
- .run(new org.apache.flink.graph.library.clustering.directed.LocalClusteringCoefficient<IntValue, NullValue, NullValue>());
+ .run(new org.apache.flink.graph.library.clustering.directed.LocalClusteringCoefficient<IntValue, NullValue, NullValue>()
+ .setIncludeZeroDegreeVertices(false));
}
} else {
boolean clipAndFlip = parameters.getBoolean("clip_and_flip", DEFAULT_CLIP_AND_FLIP);
@@ -195,7 +197,8 @@ public class ClusteringCoefficient {
gcc = newGraph
.run(new org.apache.flink.graph.library.clustering.undirected.GlobalClusteringCoefficient<LongValue, NullValue, NullValue>());
lcc = newGraph
- .run(new org.apache.flink.graph.library.clustering.undirected.LocalClusteringCoefficient<LongValue, NullValue, NullValue>());
+ .run(new org.apache.flink.graph.library.clustering.undirected.LocalClusteringCoefficient<LongValue, NullValue, NullValue>()
+ .setIncludeZeroDegreeVertices(false));
} else {
Graph<IntValue, NullValue, NullValue> newGraph = graph
.run(new TranslateGraphIds<LongValue, IntValue, NullValue, NullValue>(new LongValueToIntValue()))
@@ -204,7 +207,8 @@ public class ClusteringCoefficient {
gcc = newGraph
.run(new org.apache.flink.graph.library.clustering.undirected.GlobalClusteringCoefficient<IntValue, NullValue, NullValue>());
lcc = newGraph
- .run(new org.apache.flink.graph.library.clustering.undirected.LocalClusteringCoefficient<IntValue, NullValue, NullValue>());
+ .run(new org.apache.flink.graph.library.clustering.undirected.LocalClusteringCoefficient<IntValue, NullValue, NullValue>()
+ .setIncludeZeroDegreeVertices(false));
}
}
} break;
http://git-wip-us.apache.org/repos/asf/flink/blob/58850f29/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/examples/HITS.java
----------------------------------------------------------------------
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/examples/HITS.java
index c772a3a..59612d9 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/examples/HITS.java
@@ -65,7 +65,7 @@ public class HITS {
" 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 <csv | rmat [options]> --output <print | hash | csv [options]");
+ System.out.println("usage: HITS --input <csv | rmat [options]> --output <print | hash | csv [options]>");
System.out.println();
System.out.println("options:");
System.out.println(" --input csv --type <integer | string> --input_filename FILENAME [--input_line_delimiter LINE_DELIMITER] [--input_field_delimiter FIELD_DELIMITER]");
http://git-wip-us.apache.org/repos/asf/flink/blob/58850f29/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/examples/JaccardIndex.java
----------------------------------------------------------------------
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/examples/JaccardIndex.java
index 2158fa2..824aab7 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/examples/JaccardIndex.java
@@ -69,7 +69,7 @@ public class JaccardIndex {
System.out.println(WordUtils.wrap("This algorithm returns 4-tuples containing two vertex IDs, the" +
" number of shared neighbors, and the number of distinct neighbors.", 80));
System.out.println();
- System.out.println("usage: JaccardIndex --input <csv | rmat [options]> --output <print | hash | csv [options]");
+ System.out.println("usage: JaccardIndex --input <csv | rmat [options]> --output <print | hash | csv [options]>");
System.out.println();
System.out.println("options:");
System.out.println(" --input csv --type <integer | string> --input_filename FILENAME [--input_line_delimiter LINE_DELIMITER] [--input_field_delimiter FIELD_DELIMITER]");
http://git-wip-us.apache.org/repos/asf/flink/blob/58850f29/flink-libraries/flink-gelly-examples/src/main/java/org/apache/flink/graph/examples/TriangleListing.java
----------------------------------------------------------------------
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/examples/TriangleListing.java
index cd06dde..f3ce708 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/examples/TriangleListing.java
@@ -66,7 +66,7 @@ public class TriangleListing {
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 <true | false> --input <csv | rmat [options]> --output <print | hash | csv [options]");
+ System.out.println("usage: TriangleListing --directed <true | false> --input <csv | rmat [options]> --output <print | hash | csv [options]>");
System.out.println();
System.out.println("options:");
System.out.println(" --input csv --type <integer | string> --input_filename FILENAME [--input_line_delimiter LINE_DELIMITER] [--input_field_delimiter FIELD_DELIMITER]");
http://git-wip-us.apache.org/repos/asf/flink/blob/58850f29/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/asm/degree/annotate/directed/VertexDegrees.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/asm/degree/annotate/directed/VertexDegrees.java b/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/asm/degree/annotate/directed/VertexDegrees.java
index 9fef221..84873bc 100644
--- a/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/asm/degree/annotate/directed/VertexDegrees.java
+++ b/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/asm/degree/annotate/directed/VertexDegrees.java
@@ -86,7 +86,7 @@ extends GraphAlgorithmDelegatingDataSet<K, VV, EV, Vertex<K, Degrees>> {
@Override
protected String getAlgorithmName() {
- return VertexOutDegree.class.getName();
+ return VertexDegrees.class.getName();
}
@Override
http://git-wip-us.apache.org/repos/asf/flink/blob/58850f29/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/library/clustering/directed/LocalClusteringCoefficient.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/library/clustering/directed/LocalClusteringCoefficient.java b/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/library/clustering/directed/LocalClusteringCoefficient.java
index e0defcd..22c8b41 100644
--- a/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/library/clustering/directed/LocalClusteringCoefficient.java
+++ b/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/library/clustering/directed/LocalClusteringCoefficient.java
@@ -32,6 +32,7 @@ import org.apache.flink.graph.asm.degree.annotate.directed.VertexDegrees.Degrees
import org.apache.flink.graph.library.clustering.directed.LocalClusteringCoefficient.Result;
import org.apache.flink.graph.utils.Murmur3_32;
import org.apache.flink.graph.utils.proxy.GraphAlgorithmDelegatingDataSet;
+import org.apache.flink.graph.utils.proxy.OptionalBoolean;
import org.apache.flink.types.CopyableValue;
import org.apache.flink.types.LongValue;
import org.apache.flink.util.Collector;
@@ -59,9 +60,26 @@ public class LocalClusteringCoefficient<K extends Comparable<K> & CopyableValue<
extends GraphAlgorithmDelegatingDataSet<K, VV, EV, Result<K>> {
// Optional configuration
+ private OptionalBoolean includeZeroDegreeVertices = new OptionalBoolean(true, true);
+
private int littleParallelism = PARALLELISM_DEFAULT;
/**
+ * By default the vertex set is checked for zero degree vertices. When this
+ * flag is disabled only clustering coefficient scores for vertices with
+ * a degree of a least one will be produced.
+ *
+ * @param includeZeroDegreeVertices whether to output scores for vertices
+ * with a degree of zero
+ * @return this
+ */
+ public LocalClusteringCoefficient<K, VV, EV> setIncludeZeroDegreeVertices(boolean includeZeroDegreeVertices) {
+ this.includeZeroDegreeVertices.set(includeZeroDegreeVertices);
+
+ return this;
+ }
+
+ /**
* Override the parallelism of operators processing small amounts of data.
*
* @param littleParallelism operator parallelism
@@ -90,6 +108,16 @@ extends GraphAlgorithmDelegatingDataSet<K, VV, EV, Result<K>> {
LocalClusteringCoefficient rhs = (LocalClusteringCoefficient) other;
+ // verify that configurations can be merged
+
+ if (includeZeroDegreeVertices.conflictsWith(rhs.includeZeroDegreeVertices)) {
+ return false;
+ }
+
+ // merge configurations
+
+ includeZeroDegreeVertices.mergeWith(rhs.includeZeroDegreeVertices);
+
littleParallelism = Math.min(littleParallelism, rhs.littleParallelism);
return true;
@@ -128,8 +156,8 @@ extends GraphAlgorithmDelegatingDataSet<K, VV, EV, Result<K>> {
// u, deg(u)
DataSet<Vertex<K, Degrees>> vertexDegree = input
.run(new VertexDegrees<K, VV, EV>()
- .setParallelism(littleParallelism)
- .setIncludeZeroDegreeVertices(true));
+ .setIncludeZeroDegreeVertices(includeZeroDegreeVertices.get())
+ .setParallelism(littleParallelism));
// u, deg(u), triangle count
return vertexDegree
http://git-wip-us.apache.org/repos/asf/flink/blob/58850f29/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/library/clustering/undirected/LocalClusteringCoefficient.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/library/clustering/undirected/LocalClusteringCoefficient.java b/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/library/clustering/undirected/LocalClusteringCoefficient.java
index cd859d9..4b4bf07 100644
--- a/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/library/clustering/undirected/LocalClusteringCoefficient.java
+++ b/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/library/clustering/undirected/LocalClusteringCoefficient.java
@@ -32,6 +32,7 @@ import org.apache.flink.graph.asm.degree.annotate.undirected.VertexDegree;
import org.apache.flink.graph.library.clustering.undirected.LocalClusteringCoefficient.Result;
import org.apache.flink.graph.utils.Murmur3_32;
import org.apache.flink.graph.utils.proxy.GraphAlgorithmDelegatingDataSet;
+import org.apache.flink.graph.utils.proxy.OptionalBoolean;
import org.apache.flink.types.CopyableValue;
import org.apache.flink.types.LongValue;
import org.apache.flink.util.Collector;
@@ -59,9 +60,26 @@ public class LocalClusteringCoefficient<K extends Comparable<K> & CopyableValue<
extends GraphAlgorithmDelegatingDataSet<K, VV, EV, Result<K>> {
// Optional configuration
+ private OptionalBoolean includeZeroDegreeVertices = new OptionalBoolean(true, true);
+
private int littleParallelism = PARALLELISM_DEFAULT;
/**
+ * By default the vertex set is checked for zero degree vertices. When this
+ * flag is disabled only clustering coefficient scores for vertices with
+ * a degree of a least one will be produced.
+ *
+ * @param includeZeroDegreeVertices whether to output scores for vertices
+ * with a degree of zero
+ * @return this
+ */
+ public LocalClusteringCoefficient<K, VV, EV> setIncludeZeroDegreeVertices(boolean includeZeroDegreeVertices) {
+ this.includeZeroDegreeVertices.set(includeZeroDegreeVertices);
+
+ return this;
+ }
+
+ /**
* Override the parallelism of operators processing small amounts of data.
*
* @param littleParallelism operator parallelism
@@ -91,6 +109,15 @@ extends GraphAlgorithmDelegatingDataSet<K, VV, EV, Result<K>> {
LocalClusteringCoefficient rhs = (LocalClusteringCoefficient) other;
+ // verify that configurations can be merged
+
+ if (includeZeroDegreeVertices.conflictsWith(rhs.includeZeroDegreeVertices)) {
+ return false;
+ }
+
+ // merge configurations
+
+ includeZeroDegreeVertices.mergeWith(rhs.includeZeroDegreeVertices);
littleParallelism = Math.min(littleParallelism, rhs.littleParallelism);
return true;
@@ -129,8 +156,8 @@ extends GraphAlgorithmDelegatingDataSet<K, VV, EV, Result<K>> {
// u, deg(u)
DataSet<Vertex<K, LongValue>> vertexDegree = input
.run(new VertexDegree<K, VV, EV>()
- .setParallelism(littleParallelism)
- .setIncludeZeroDegreeVertices(true));
+ .setIncludeZeroDegreeVertices(includeZeroDegreeVertices.get())
+ .setParallelism(littleParallelism));
// u, deg(u), triangle count
return vertexDegree
http://git-wip-us.apache.org/repos/asf/flink/blob/58850f29/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/library/metric/directed/EdgeMetrics.java
----------------------------------------------------------------------
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
new file mode 100644
index 0000000..167e31c
--- /dev/null
+++ b/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/library/metric/directed/EdgeMetrics.java
@@ -0,0 +1,507 @@
+/*
+ * 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.library.metric.directed;
+
+import org.apache.commons.lang3.builder.EqualsBuilder;
+import org.apache.commons.lang3.builder.HashCodeBuilder;
+import org.apache.flink.api.common.JobExecutionResult;
+import org.apache.flink.api.common.accumulators.LongCounter;
+import org.apache.flink.api.common.accumulators.LongMaximum;
+import org.apache.flink.api.common.functions.FlatMapFunction;
+import org.apache.flink.api.common.functions.GroupReduceFunction;
+import org.apache.flink.api.common.functions.ReduceFunction;
+import org.apache.flink.api.common.io.RichOutputFormat;
+import org.apache.flink.api.common.operators.base.ReduceOperatorBase.CombineHint;
+import org.apache.flink.api.java.DataSet;
+import org.apache.flink.api.java.functions.FunctionAnnotation.ForwardedFields;
+import org.apache.flink.api.java.tuple.Tuple3;
+import org.apache.flink.api.java.tuple.Tuple4;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.graph.AbstractGraphAnalytic;
+import org.apache.flink.graph.Edge;
+import org.apache.flink.graph.Graph;
+import org.apache.flink.graph.asm.degree.annotate.directed.EdgeDegreesPair;
+import org.apache.flink.graph.asm.degree.annotate.directed.VertexDegrees.Degrees;
+import org.apache.flink.graph.library.metric.directed.EdgeMetrics.Result;
+import org.apache.flink.types.CopyableValue;
+import org.apache.flink.types.LongValue;
+import org.apache.flink.util.AbstractID;
+import org.apache.flink.util.Collector;
+
+import java.io.IOException;
+import java.text.NumberFormat;
+
+import static org.apache.flink.api.common.ExecutionConfig.PARALLELISM_DEFAULT;
+
+/**
+ * Compute the following edge metrics in a directed graph:
+ * - number of vertices
+ * - number of edges
+ * - number of triangle triplets
+ * - number of rectangle triplets
+ * - number of triplets
+ * - maximum degree
+ * - maximum out degree
+ * - maximum in degree
+ * - maximum number of triangle triplets
+ * - maximum number of rectangle triplets
+ * - maximum number of triplets
+ *
+ * @param <K> graph ID type
+ * @param <VV> vertex value type
+ * @param <EV> edge value type
+ */
+public class EdgeMetrics<K extends Comparable<K> & CopyableValue<K>, VV, EV>
+extends AbstractGraphAnalytic<K, VV, EV, Result> {
+
+ private String id = new AbstractID().toString();
+
+ private int parallelism = PARALLELISM_DEFAULT;
+
+ /**
+ * Override the operator parallelism.
+ *
+ * @param parallelism operator parallelism
+ * @return this
+ */
+ public EdgeMetrics<K, VV, EV> setParallelism(int parallelism) {
+ this.parallelism = parallelism;
+
+ return this;
+ }
+
+ /*
+ * Implementation notes:
+ *
+ * Use aggregator to replace SumEdgeStats when aggregators are rewritten to use
+ * a hash-combineable hashable-reduce.
+ *
+ * Use distinct to replace ReduceEdgeStats when the combiner can be disabled
+ * with a sorted-reduce forced.
+ */
+
+ @Override
+ public EdgeMetrics<K, VV, EV> run(Graph<K, VV, EV> input)
+ throws Exception {
+ super.run(input);
+
+ // s, t, (d(s), d(t))
+ DataSet<Edge<K, Tuple3<EV, Degrees, Degrees>>> edgeDegreesPair = input
+ .run(new EdgeDegreesPair<K, VV, EV>()
+ .setParallelism(parallelism));
+
+ // s, d(s), count of (u, v) where deg(u) < deg(v) or (deg(u) == deg(v) and u < v)
+ DataSet<Tuple3<K, Degrees, LongValue>> edgeStats = edgeDegreesPair
+ .flatMap(new EdgeStats<K, EV>())
+ .setParallelism(parallelism)
+ .name("Edge stats")
+ .groupBy(0, 1)
+ .reduceGroup(new ReduceEdgeStats<K>())
+ .setParallelism(parallelism)
+ .name("Reduce edge stats")
+ .groupBy(0)
+ .reduce(new SumEdgeStats<K>())
+ .setCombineHint(CombineHint.HASH)
+ .setParallelism(parallelism)
+ .name("Sum edge stats");
+
+ edgeStats
+ .output(new EdgeMetricsHelper<K, EV>(id))
+ .setParallelism(parallelism)
+ .name("Edge metrics");
+
+ return this;
+ }
+
+ @Override
+ public Result getResult() {
+ JobExecutionResult res = env.getLastJobExecutionResult();
+
+ long vertexCount = res.getAccumulatorResult(id + "-0");
+ long edgeCount = res.getAccumulatorResult(id + "-1");
+ long triangleTripletCount = res.getAccumulatorResult(id + "-2");
+ long rectangleTripletCount = res.getAccumulatorResult(id + "-3");
+ long tripletCount = res.getAccumulatorResult(id + "-4");
+ long maximumDegree = res.getAccumulatorResult(id + "-5");
+ long maximumOutDegree = res.getAccumulatorResult(id + "-6");
+ long maximumInDegree = res.getAccumulatorResult(id + "-7");
+ long maximumTriangleTriplets = res.getAccumulatorResult(id + "-8");
+ long maximumRectangleTriplets = res.getAccumulatorResult(id + "-9");
+ long maximumTriplets = res.getAccumulatorResult(id + "-a");
+
+ return new Result(vertexCount, edgeCount, triangleTripletCount, rectangleTripletCount, tripletCount,
+ maximumDegree, maximumOutDegree, maximumInDegree,
+ maximumTriangleTriplets, maximumRectangleTriplets, maximumTriplets);
+ }
+
+ /**
+ * Produces a pair of tuples. The first tuple contains the source vertex ID,
+ * the target vertex ID, the source degrees, and the low-order count. The
+ * second tuple is the same with the source and target roles reversed.
+ *
+ * The low-order count is one if the source vertex degree is less than the
+ * target vertex degree or if the degrees are equal and the source vertex
+ * ID compares lower than the target vertex ID; otherwise the low-order
+ * count is zero.
+ *
+ * @param <T> ID type
+ * @param <ET> edge value type
+ */
+ private static final class EdgeStats<T extends Comparable<T>, ET>
+ implements FlatMapFunction<Edge<T, Tuple3<ET, Degrees, Degrees>>, Tuple4<T, T, Degrees, LongValue>> {
+ private LongValue zero = new LongValue(0);
+
+ private LongValue one = new LongValue(1);
+
+ private Tuple4<T, T, Degrees, LongValue> output = new Tuple4<>();
+
+ @Override
+ public void flatMap(Edge<T, Tuple3<ET, Degrees, Degrees>> edge, Collector<Tuple4<T, T, Degrees, LongValue>> out)
+ throws Exception {
+ Tuple3<ET, Degrees, Degrees> degrees = edge.f2;
+ long sourceDegree = degrees.f1.getDegree().getValue();
+ long targetDegree = degrees.f2.getDegree().getValue();
+
+ boolean ordered = (sourceDegree < targetDegree
+ || (sourceDegree == targetDegree && edge.f0.compareTo(edge.f1) < 0));
+
+ output.f0 = edge.f0;
+ output.f1 = edge.f1;
+ output.f2 = edge.f2.f1;
+ output.f3 = ordered ? one : zero;
+ out.collect(output);
+
+ output.f0 = edge.f1;
+ output.f1 = edge.f0;
+ output.f2 = edge.f2.f2;
+ output.f3 = ordered ? zero : one;
+ out.collect(output);
+ }
+ }
+
+ /**
+ * Produces a distinct value for each edge.
+ *
+ * @param <T> ID type
+ */
+ @ForwardedFields("0")
+ private static final class ReduceEdgeStats<T>
+ implements GroupReduceFunction<Tuple4<T, T, Degrees, LongValue>, Tuple3<T, Degrees, LongValue>> {
+ Tuple3<T, Degrees, LongValue> output = new Tuple3<>();
+
+ @Override
+ public void reduce(Iterable<Tuple4<T, T, Degrees, LongValue>> values, Collector<Tuple3<T, Degrees, LongValue>> out)
+ throws Exception {
+ Tuple4<T, T, Degrees, LongValue> value = values.iterator().next();
+
+ output.f0 = value.f0;
+ output.f1 = value.f2;
+ output.f2 = value.f3;
+
+ out.collect(output);
+ }
+ }
+
+ /**
+ * Sums the low-order counts.
+ *
+ * @param <T> ID type
+ */
+ private static class SumEdgeStats<T>
+ implements ReduceFunction<Tuple3<T, Degrees, LongValue>> {
+ @Override
+ public Tuple3<T, Degrees, LongValue> reduce(Tuple3<T, Degrees, LongValue> value1, Tuple3<T, Degrees, LongValue> value2)
+ throws Exception {
+ value1.f2.setValue(value1.f2.getValue() + value2.f2.getValue());
+ return value1;
+ }
+ }
+
+ /**
+ * Helper class to collect edge metrics.
+ *
+ * @param <T> ID type
+ */
+ private static class EdgeMetricsHelper<T extends Comparable<T>, ET>
+ extends RichOutputFormat<Tuple3<T, Degrees, LongValue>> {
+ private final String id;
+
+ private long vertexCount;
+ private long edgeCount;
+ private long triangleTripletCount;
+ private long rectangleTripletCount;
+ private long tripletCount;
+ private long maximumDegree;
+ private long maximumOutDegree;
+ private long maximumInDegree;
+ private long maximumTriangleTriplets;
+ private long maximumRectangleTriplets;
+ private long maximumTriplets;
+
+ /**
+ * This helper class collects edge metrics by scanning over and
+ * discarding elements from the given DataSet.
+ *
+ * The unique id is required because Flink's accumulator namespace is
+ * among all operators.
+ *
+ * @param id unique string used for accumulator names
+ */
+ public EdgeMetricsHelper(String id) {
+ this.id = id;
+ }
+
+ @Override
+ public void configure(Configuration parameters) {}
+
+ @Override
+ public void open(int taskNumber, int numTasks) throws IOException {}
+
+ @Override
+ public void writeRecord(Tuple3<T, Degrees, LongValue> record) throws IOException {
+ Degrees degrees = record.f1;
+ long degree = degrees.getDegree().getValue();
+ long outDegree = degrees.getOutDegree().getValue();
+ long inDegree = degrees.getInDegree().getValue();
+
+ long lowDegree = record.f2.getValue();
+ long highDegree = degree - lowDegree;
+
+ long triangleTriplets = lowDegree * (lowDegree - 1) / 2;
+ long rectangleTriplets = triangleTriplets + lowDegree * highDegree;
+ long triplets = degree * (degree - 1) / 2;
+
+ vertexCount++;
+ edgeCount += outDegree;
+ triangleTripletCount += triangleTriplets;
+ rectangleTripletCount += rectangleTriplets;
+ tripletCount += triplets;
+ maximumDegree = Math.max(maximumDegree, degree);
+ maximumOutDegree = Math.max(maximumOutDegree, outDegree);
+ maximumInDegree = Math.max(maximumInDegree, inDegree);
+ maximumTriangleTriplets = Math.max(maximumTriangleTriplets, triangleTriplets);
+ maximumRectangleTriplets = Math.max(maximumRectangleTriplets, rectangleTriplets);
+ maximumTriplets = Math.max(maximumTriplets, triplets);
+ }
+
+ @Override
+ public void close() throws IOException {
+ getRuntimeContext().addAccumulator(id + "-0", new LongCounter(vertexCount));
+ getRuntimeContext().addAccumulator(id + "-1", new LongCounter(edgeCount));
+ getRuntimeContext().addAccumulator(id + "-2", new LongCounter(triangleTripletCount));
+ getRuntimeContext().addAccumulator(id + "-3", new LongCounter(rectangleTripletCount));
+ getRuntimeContext().addAccumulator(id + "-4", new LongCounter(tripletCount));
+ getRuntimeContext().addAccumulator(id + "-5", new LongMaximum(maximumDegree));
+ getRuntimeContext().addAccumulator(id + "-6", new LongMaximum(maximumOutDegree));
+ getRuntimeContext().addAccumulator(id + "-7", new LongMaximum(maximumInDegree));
+ getRuntimeContext().addAccumulator(id + "-8", new LongMaximum(maximumTriangleTriplets));
+ getRuntimeContext().addAccumulator(id + "-9", new LongMaximum(maximumRectangleTriplets));
+ getRuntimeContext().addAccumulator(id + "-a", new LongMaximum(maximumTriplets));
+ }
+ }
+
+ /**
+ * Wraps edge metrics.
+ */
+ public static class Result {
+ private long vertexCount;
+ private long edgeCount;
+ private long triangleTripletCount;
+ private long rectangleTripletCount;
+ private long tripletCount;
+ private long maximumDegree;
+ private long maximumOutDegree;
+ private long maximumInDegree;
+ private long maximumTriangleTriplets;
+ private long maximumRectangleTriplets;
+ private long maximumTriplets;
+
+ public Result(long vertexCount, long edgeCount, long triangleTripletCount, long rectangleTripletCount, long tripletCount,
+ long maximumDegree, long maximumOutDegree, long maximumInDegree,
+ long maximumTriangleTriplets, long maximumRectangleTriplets, long maximumTriplets) {
+ this.vertexCount = vertexCount;
+ this.edgeCount = edgeCount;
+ this.triangleTripletCount = triangleTripletCount;
+ this.rectangleTripletCount = rectangleTripletCount;
+ this.tripletCount = tripletCount;
+ this.maximumDegree = maximumDegree;
+ this.maximumOutDegree = maximumOutDegree;
+ this.maximumInDegree = maximumInDegree;
+ this.maximumTriangleTriplets = maximumTriangleTriplets;
+ this.maximumRectangleTriplets = maximumRectangleTriplets;
+ this.maximumTriplets = maximumTriplets;
+ }
+
+ /**
+ * Get the number of vertices.
+ *
+ * @return number of vertices
+ */
+ public long getNumberOfVertices() {
+ return vertexCount;
+ }
+
+ /**
+ * Get the number of edges.
+ *
+ * @return number of edges
+ */
+ public long getNumberOfEdges() {
+ return edgeCount;
+ }
+
+ /**
+ * Get the number of triangle triplets.
+ *
+ * @return number of triangle triplets
+ */
+ public long getNumberOfTriangleTriplets() {
+ return triangleTripletCount;
+ }
+
+ /**
+ * Get the number of rectangle triplets.
+ *
+ * @return number of rectangle triplets
+ */
+ public long getNumberOfRectangleTriplets() {
+ return rectangleTripletCount;
+ }
+
+ /**
+ * Get the number of triplets.
+ *
+ * @return number of triplets
+ */
+ public long getNumberOfTriplets() {
+ return tripletCount;
+ }
+
+ /**
+ * Get the maximum degree.
+ *
+ * @return maximum degree
+ */
+ public long getMaximumDegree() {
+ return maximumDegree;
+ }
+
+ /**
+ * Get the maximum out degree.
+ *
+ * @return maximum out degree
+ */
+ public long getMaximumOutDegree() {
+ return maximumOutDegree;
+ }
+
+ /**
+ * Get the maximum in degree.
+ *
+ * @return maximum in degree
+ */
+ public long getMaximumInDegree() {
+ return maximumInDegree;
+ }
+
+ /**
+ * Get the maximum triangle triplets.
+ *
+ * @return maximum triangle triplets
+ */
+ public long getMaximumTriangleTriplets() {
+ return maximumTriangleTriplets;
+ }
+
+ /**
+ * Get the maximum rectangle triplets.
+ *
+ * @return maximum rectangle triplets
+ */
+ public long getMaximumRectangleTriplets() {
+ return maximumRectangleTriplets;
+ }
+
+ /**
+ * Get the maximum triplets.
+ *
+ * @return maximum triplets
+ */
+ public long getMaximumTriplets() {
+ return maximumTriplets;
+ }
+
+ @Override
+ public String toString() {
+ NumberFormat nf = NumberFormat.getInstance();
+
+ return "vertex count: " + nf.format(vertexCount)
+ + "; edge count: " + nf.format(edgeCount)
+ + "; triangle triplet count: " + nf.format(triangleTripletCount)
+ + "; rectangle triplet count: " + nf.format(rectangleTripletCount)
+ + "; triplet count: " + nf.format(tripletCount)
+ + "; maximum degree: " + nf.format(maximumDegree)
+ + "; maximum out degree: " + nf.format(maximumOutDegree)
+ + "; maximum in degree: " + nf.format(maximumInDegree)
+ + "; maximum triangle triplets: " + nf.format(maximumTriangleTriplets)
+ + "; maximum rectangle triplets: " + nf.format(maximumRectangleTriplets)
+ + "; maximum triplets: " + nf.format(maximumTriplets);
+ }
+
+ @Override
+ public int hashCode() {
+ return new HashCodeBuilder()
+ .append(vertexCount)
+ .append(edgeCount)
+ .append(triangleTripletCount)
+ .append(rectangleTripletCount)
+ .append(tripletCount)
+ .append(maximumDegree)
+ .append(maximumOutDegree)
+ .append(maximumInDegree)
+ .append(maximumTriangleTriplets)
+ .append(maximumRectangleTriplets)
+ .append(maximumTriplets)
+ .hashCode();
+ }
+
+ @Override
+ public boolean equals(Object obj) {
+ if (obj == null) { return false; }
+ if (obj == this) { return true; }
+ if (obj.getClass() != getClass()) { return false; }
+
+ Result rhs = (Result)obj;
+
+ return new EqualsBuilder()
+ .append(vertexCount, rhs.vertexCount)
+ .append(edgeCount, rhs.edgeCount)
+ .append(triangleTripletCount, rhs.triangleTripletCount)
+ .append(rectangleTripletCount, rhs.rectangleTripletCount)
+ .append(tripletCount, rhs.tripletCount)
+ .append(maximumDegree, rhs.maximumDegree)
+ .append(maximumOutDegree, rhs.maximumOutDegree)
+ .append(maximumInDegree, rhs.maximumInDegree)
+ .append(maximumTriangleTriplets, rhs.maximumTriangleTriplets)
+ .append(maximumRectangleTriplets, rhs.maximumRectangleTriplets)
+ .append(maximumTriplets, rhs.maximumTriplets)
+ .isEquals();
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/58850f29/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/library/metric/directed/VertexMetrics.java
----------------------------------------------------------------------
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 434bd28..22f7733 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
@@ -22,6 +22,7 @@ import org.apache.commons.lang3.builder.EqualsBuilder;
import org.apache.commons.lang3.builder.HashCodeBuilder;
import org.apache.flink.api.common.JobExecutionResult;
import org.apache.flink.api.common.accumulators.LongCounter;
+import org.apache.flink.api.common.accumulators.LongMaximum;
import org.apache.flink.api.common.io.RichOutputFormat;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.configuration.Configuration;
@@ -35,12 +36,19 @@ import org.apache.flink.types.CopyableValue;
import org.apache.flink.util.AbstractID;
import java.io.IOException;
+import java.text.NumberFormat;
import static org.apache.flink.api.common.ExecutionConfig.PARALLELISM_DEFAULT;
/**
- * Compute the number of vertices, number of edges, and number of triplets in
- * a directed graph.
+ * Compute the following vertex metrics in a directed graph:
+ * - number of vertices
+ * - number of edges
+ * - number of triplets
+ * - maximum degree
+ * - maximum out degree
+ * - maximum in degree
+ * - maximum number of triplets
*
* @param <K> graph ID type
* @param <VV> vertex value type
@@ -107,8 +115,12 @@ extends AbstractGraphAnalytic<K, VV, EV, Result> {
long vertexCount = res.getAccumulatorResult(id + "-0");
long edgeCount = res.getAccumulatorResult(id + "-1");
long tripletCount = res.getAccumulatorResult(id + "-2");
+ long maximumDegree = res.getAccumulatorResult(id + "-3");
+ long maximumOutDegree = res.getAccumulatorResult(id + "-4");
+ long maximumInDegree = res.getAccumulatorResult(id + "-5");
+ long maximumTriplets = res.getAccumulatorResult(id + "-6");
- return new Result(vertexCount, edgeCount / 2, tripletCount);
+ return new Result(vertexCount, edgeCount, tripletCount, maximumDegree, maximumOutDegree, maximumInDegree, maximumTriplets);
}
/**
@@ -123,13 +135,17 @@ extends AbstractGraphAnalytic<K, VV, EV, Result> {
private long vertexCount;
private long edgeCount;
private long tripletCount;
+ private long maximumDegree;
+ private long maximumOutDegree;
+ private long maximumInDegree;
+ private long maximumTriplets;
/**
* This helper class collects vertex metrics by scanning over and
* discarding elements from the given DataSet.
*
* The unique id is required because Flink's accumulator namespace is
- * among all operators.
+ * shared among all operators.
*
* @param id unique string used for accumulator names
*/
@@ -147,10 +163,16 @@ extends AbstractGraphAnalytic<K, VV, EV, Result> {
public void writeRecord(Vertex<T, Degrees> record) throws IOException {
long degree = record.f1.getDegree().getValue();
long outDegree = record.f1.getOutDegree().getValue();
+ long inDegree = record.f1.getInDegree().getValue();
+ long triplets = degree * (degree - 1) / 2;
vertexCount++;
edgeCount += outDegree;
- tripletCount += degree * (degree - 1) / 2;
+ tripletCount += triplets;
+ maximumDegree = Math.max(maximumDegree, degree);
+ maximumOutDegree = Math.max(maximumOutDegree, outDegree);
+ maximumInDegree = Math.max(maximumInDegree, inDegree);
+ maximumTriplets = Math.max(maximumTriplets, triplets);
}
@Override
@@ -158,6 +180,10 @@ extends AbstractGraphAnalytic<K, VV, EV, Result> {
getRuntimeContext().addAccumulator(id + "-0", new LongCounter(vertexCount));
getRuntimeContext().addAccumulator(id + "-1", new LongCounter(edgeCount));
getRuntimeContext().addAccumulator(id + "-2", new LongCounter(tripletCount));
+ getRuntimeContext().addAccumulator(id + "-3", new LongMaximum(maximumDegree));
+ getRuntimeContext().addAccumulator(id + "-4", new LongMaximum(maximumOutDegree));
+ getRuntimeContext().addAccumulator(id + "-5", new LongMaximum(maximumInDegree));
+ getRuntimeContext().addAccumulator(id + "-6", new LongMaximum(maximumTriplets));
}
}
@@ -168,11 +194,19 @@ extends AbstractGraphAnalytic<K, VV, EV, Result> {
private long vertexCount;
private long edgeCount;
private long tripletCount;
+ private long maximumDegree;
+ private long maximumOutDegree;
+ private long maximumInDegree;
+ private long maximumTriplets;
- public Result(long vertexCount, long edgeCount, long tripletCount) {
+ public Result(long vertexCount, long edgeCount, long tripletCount, long maximumDegree, long maximumOutDegree, long maximumInDegree, long maximumTriplets) {
this.vertexCount = vertexCount;
this.edgeCount = edgeCount;
this.tripletCount = tripletCount;
+ this.maximumDegree = maximumDegree;
+ this.maximumOutDegree = maximumOutDegree;
+ this.maximumInDegree = maximumInDegree;
+ this.maximumTriplets = maximumTriplets;
}
/**
@@ -202,11 +236,53 @@ extends AbstractGraphAnalytic<K, VV, EV, Result> {
return tripletCount;
}
+ /**
+ * Get the maximum degree.
+ *
+ * @return maximum degree
+ */
+ public long getMaximumDegree() {
+ return maximumDegree;
+ }
+
+ /**
+ * Get the maximum out degree.
+ *
+ * @return maximum out degree
+ */
+ public long getMaximumOutDegree() {
+ return maximumOutDegree;
+ }
+
+ /**
+ * Get the maximum in degree.
+ *
+ * @return maximum in degree
+ */
+ public long getMaximumInDegree() {
+ return maximumInDegree;
+ }
+
+ /**
+ * Get the maximum triplets.
+ *
+ * @return maximum triplets
+ */
+ public long getMaximumTriplets() {
+ return maximumTriplets;
+ }
+
@Override
public String toString() {
- return "vertex count: " + vertexCount
- + ", edge count:" + edgeCount
- + ", triplet count: " + tripletCount;
+ NumberFormat nf = NumberFormat.getInstance();
+
+ return "vertex count: " + nf.format(vertexCount)
+ + "; edge count: " + nf.format(edgeCount)
+ + "; triplet count: " + nf.format(tripletCount)
+ + "; maximum degree: " + nf.format(maximumDegree)
+ + "; maximum out degree: " + nf.format(maximumOutDegree)
+ + "; maximum in degree: " + nf.format(maximumInDegree)
+ + "; maximum triplets: " + nf.format(maximumTriplets);
}
@Override
@@ -215,6 +291,10 @@ extends AbstractGraphAnalytic<K, VV, EV, Result> {
.append(vertexCount)
.append(edgeCount)
.append(tripletCount)
+ .append(maximumDegree)
+ .append(maximumOutDegree)
+ .append(maximumInDegree)
+ .append(maximumTriplets)
.hashCode();
}
@@ -230,6 +310,10 @@ extends AbstractGraphAnalytic<K, VV, EV, Result> {
.append(vertexCount, rhs.vertexCount)
.append(edgeCount, rhs.edgeCount)
.append(tripletCount, rhs.tripletCount)
+ .append(maximumDegree, rhs.maximumDegree)
+ .append(maximumOutDegree, rhs.maximumOutDegree)
+ .append(maximumInDegree, rhs.maximumInDegree)
+ .append(maximumTriplets, rhs.maximumTriplets)
.isEquals();
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/58850f29/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/library/metric/undirected/EdgeMetrics.java
----------------------------------------------------------------------
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
new file mode 100644
index 0000000..1d5b664
--- /dev/null
+++ b/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/library/metric/undirected/EdgeMetrics.java
@@ -0,0 +1,445 @@
+/*
+ * 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.library.metric.undirected;
+
+import org.apache.commons.lang3.builder.EqualsBuilder;
+import org.apache.commons.lang3.builder.HashCodeBuilder;
+import org.apache.flink.api.common.JobExecutionResult;
+import org.apache.flink.api.common.accumulators.LongCounter;
+import org.apache.flink.api.common.accumulators.LongMaximum;
+import org.apache.flink.api.common.functions.MapFunction;
+import org.apache.flink.api.common.functions.ReduceFunction;
+import org.apache.flink.api.common.io.RichOutputFormat;
+import org.apache.flink.api.common.operators.base.ReduceOperatorBase.CombineHint;
+import org.apache.flink.api.java.DataSet;
+import org.apache.flink.api.java.functions.FunctionAnnotation;
+import org.apache.flink.api.java.tuple.Tuple3;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.graph.AbstractGraphAnalytic;
+import org.apache.flink.graph.Edge;
+import org.apache.flink.graph.Graph;
+import org.apache.flink.graph.asm.degree.annotate.undirected.EdgeDegreePair;
+import org.apache.flink.graph.library.metric.undirected.EdgeMetrics.Result;
+import org.apache.flink.types.CopyableValue;
+import org.apache.flink.types.LongValue;
+import org.apache.flink.util.AbstractID;
+
+import java.io.IOException;
+import java.text.NumberFormat;
+
+import static org.apache.flink.api.common.ExecutionConfig.PARALLELISM_DEFAULT;
+
+/**
+ * Compute the following edge metrics in an undirected graph:
+ * - number of vertices
+ * - number of edges
+ * - number of triangle triplets
+ * - number of rectangle triplets
+ * - number of triplets
+ * - maximum degree
+ * - maximum number of triangle triplets
+ * - maximum number of rectangle triplets
+ * - maximum number of triplets
+ *
+ * @param <K> graph ID type
+ * @param <VV> vertex value type
+ * @param <EV> edge value type
+ */
+public class EdgeMetrics<K extends Comparable<K> & CopyableValue<K>, VV, EV>
+extends AbstractGraphAnalytic<K, VV, EV, Result> {
+
+ private String id = new AbstractID().toString();
+
+ // Optional configuration
+ private boolean reduceOnTargetId = false;
+
+ private int parallelism = PARALLELISM_DEFAULT;
+
+ /**
+ * The degree can be counted from either the edge source or target IDs.
+ * By default the source IDs are counted. Reducing on target IDs may
+ * optimize the algorithm if the input edge list is sorted by target ID.
+ *
+ * @param reduceOnTargetId set to {@code true} if the input edge list
+ * is sorted by target ID
+ * @return this
+ */
+ public EdgeMetrics<K, VV, EV> setReduceOnTargetId(boolean reduceOnTargetId) {
+ this.reduceOnTargetId = reduceOnTargetId;
+
+ return this;
+ }
+
+ /**
+ * Override the operator parallelism.
+ *
+ * @param parallelism operator parallelism
+ * @return this
+ */
+ public EdgeMetrics<K, VV, EV> setParallelism(int parallelism) {
+ this.parallelism = parallelism;
+
+ return this;
+ }
+
+ /*
+ * Implementation notes:
+ *
+ * Use aggregator to replace SumEdgeStats when aggregators are rewritten to use
+ * a hash-combineable hashed-reduce.
+ */
+
+ @Override
+ public EdgeMetrics<K, VV, EV> run(Graph<K, VV, EV> input)
+ throws Exception {
+ super.run(input);
+
+ // s, t, (d(s), d(t))
+ DataSet<Edge<K, Tuple3<EV, LongValue, LongValue>>> edgeDegreePair = input
+ .run(new EdgeDegreePair<K, VV, EV>()
+ .setReduceOnTargetId(reduceOnTargetId)
+ .setParallelism(parallelism));
+
+ // s, d(s), count of (u, v) where deg(u) < deg(v) or (deg(u) == deg(v) and u < v)
+ DataSet<Tuple3<K, LongValue, LongValue>> edgeStats = edgeDegreePair
+ .map(new EdgeStats<K, EV>())
+ .setParallelism(parallelism)
+ .name("Edge stats")
+ .groupBy(0)
+ .reduce(new SumEdgeStats<K>())
+ .setCombineHint(CombineHint.HASH)
+ .setParallelism(parallelism)
+ .name("Sum edge stats");
+
+ edgeStats
+ .output(new EdgeMetricsHelper<K, EV>(id))
+ .setParallelism(parallelism)
+ .name("Edge metrics");
+
+ return this;
+ }
+
+ @Override
+ public Result getResult() {
+ JobExecutionResult res = env.getLastJobExecutionResult();
+
+ long vertexCount = res.getAccumulatorResult(id + "-0");
+ long edgeCount = res.getAccumulatorResult(id + "-1");
+ long triangleTripletCount = res.getAccumulatorResult(id + "-2");
+ long rectangleTripletCount = res.getAccumulatorResult(id + "-3");
+ long tripletCount = res.getAccumulatorResult(id + "-4");
+ long maximumDegree = res.getAccumulatorResult(id + "-5");
+ long maximumTriangleTriplets = res.getAccumulatorResult(id + "-6");
+ long maximumRectangleTriplets = res.getAccumulatorResult(id + "-7");
+ long maximumTriplets = res.getAccumulatorResult(id + "-8");
+
+ return new Result(vertexCount, edgeCount / 2, triangleTripletCount, rectangleTripletCount, tripletCount,
+ maximumDegree, maximumTriangleTriplets, maximumRectangleTriplets, maximumTriplets);
+ }
+
+ /**
+ * Evaluates each edge and emits a tuple containing the source vertex ID,
+ * the source vertex degree, and a value of zero or one indicating the
+ * low-order count. The low-order count is one if the source vertex degree
+ * is less than the target vertex degree or if the degrees are equal and
+ * the source vertex ID compares lower than the target vertex ID; otherwise
+ * the low-order count is zero.
+ *
+ * @param <T> ID type
+ * @param <ET> edge value type
+ */
+ @FunctionAnnotation.ForwardedFields("0; 2.1->1")
+ private static class EdgeStats<T extends Comparable<T>, ET>
+ implements MapFunction<Edge<T, Tuple3<ET, LongValue, LongValue>>, Tuple3<T, LongValue, LongValue>> {
+ private LongValue zero = new LongValue(0);
+
+ private LongValue one = new LongValue(1);
+
+ private Tuple3<T, LongValue, LongValue> output = new Tuple3<>();
+
+ @Override
+ public Tuple3<T, LongValue, LongValue> map(Edge<T, Tuple3<ET, LongValue, LongValue>> edge)
+ throws Exception {
+ Tuple3<ET, LongValue, LongValue> degrees = edge.f2;
+
+ output.f0 = edge.f0;
+ output.f1 = degrees.f1;
+
+ long sourceDegree = degrees.f1.getValue();
+ long targetDegree = degrees.f2.getValue();
+
+ if (sourceDegree < targetDegree ||
+ (sourceDegree == targetDegree && edge.f0.compareTo(edge.f1) < 0)) {
+ output.f2 = one;
+ } else {
+ output.f2 = zero;
+ }
+
+ return output;
+ }
+ }
+
+ /**
+ * Sums the low-order counts.
+ *
+ * @param <T> ID type
+ */
+ private static class SumEdgeStats<T>
+ implements ReduceFunction<Tuple3<T, LongValue, LongValue>> {
+ @Override
+ public Tuple3<T, LongValue, LongValue> reduce(Tuple3<T, LongValue, LongValue> value1, Tuple3<T, LongValue, LongValue> value2)
+ throws Exception {
+ value1.f2.setValue(value1.f2.getValue() + value2.f2.getValue());
+ return value1;
+ }
+ }
+
+ /**
+ * Helper class to collect edge metrics.
+ *
+ * @param <T> ID type
+ */
+ private static class EdgeMetricsHelper<T extends Comparable<T>, ET>
+ extends RichOutputFormat<Tuple3<T, LongValue, LongValue>> {
+ private final String id;
+
+ private long vertexCount;
+ private long edgeCount;
+ private long triangleTripletCount;
+ private long rectangleTripletCount;
+ private long tripletCount;
+ private long maximumDegree;
+ private long maximumTriangleTriplets;
+ private long maximumRectangleTriplets;
+ private long maximumTriplets;
+
+ /**
+ * This helper class collects edge metrics by scanning over and
+ * discarding elements from the given DataSet.
+ *
+ * The unique id is required because Flink's accumulator namespace is
+ * among all operators.
+ *
+ * @param id unique string used for accumulator names
+ */
+ public EdgeMetricsHelper(String id) {
+ this.id = id;
+ }
+
+ @Override
+ public void configure(Configuration parameters) {}
+
+ @Override
+ public void open(int taskNumber, int numTasks) throws IOException {}
+
+ @Override
+ public void writeRecord(Tuple3<T, LongValue, LongValue> record) throws IOException {
+ long degree = record.f1.getValue();
+ long lowDegree = record.f2.getValue();
+ long highDegree = degree - lowDegree;
+
+ long triangleTriplets = lowDegree * (lowDegree - 1) / 2;
+ long rectangleTriplets = triangleTriplets + lowDegree * highDegree;
+ long triplets = degree * (degree - 1) / 2;
+
+ vertexCount++;
+ edgeCount += degree;
+ triangleTripletCount += triangleTriplets;
+ rectangleTripletCount += rectangleTriplets;
+ tripletCount += triplets;
+ maximumDegree = Math.max(maximumDegree, degree);
+ maximumTriangleTriplets = Math.max(maximumTriangleTriplets, triangleTriplets);
+ maximumRectangleTriplets = Math.max(maximumRectangleTriplets, rectangleTriplets);
+ maximumTriplets = Math.max(maximumTriplets, triplets);
+ }
+
+ @Override
+ public void close() throws IOException {
+ getRuntimeContext().addAccumulator(id + "-0", new LongCounter(vertexCount));
+ getRuntimeContext().addAccumulator(id + "-1", new LongCounter(edgeCount));
+ getRuntimeContext().addAccumulator(id + "-2", new LongCounter(triangleTripletCount));
+ getRuntimeContext().addAccumulator(id + "-3", new LongCounter(rectangleTripletCount));
+ getRuntimeContext().addAccumulator(id + "-4", new LongCounter(tripletCount));
+ getRuntimeContext().addAccumulator(id + "-5", new LongMaximum(maximumDegree));
+ getRuntimeContext().addAccumulator(id + "-6", new LongMaximum(maximumTriangleTriplets));
+ getRuntimeContext().addAccumulator(id + "-7", new LongMaximum(maximumRectangleTriplets));
+ getRuntimeContext().addAccumulator(id + "-8", new LongMaximum(maximumTriplets));
+ }
+ }
+
+ /**
+ * Wraps edge metrics.
+ */
+ public static class Result {
+ private long vertexCount;
+ private long edgeCount;
+ private long triangleTripletCount;
+ private long rectangleTripletCount;
+ private long tripletCount;
+ private long maximumDegree;
+ private long maximumTriangleTriplets;
+ private long maximumRectangleTriplets;
+ private long maximumTriplets;
+
+ public Result(long vertexCount, long edgeCount, long triangleTripletCount, long rectangleTripletCount, long tripletCount,
+ long maximumDegree, long maximumTriangleTriplets, long maximumRectangleTriplets, long maximumTriplets) {
+ this.vertexCount = vertexCount;
+ this.edgeCount = edgeCount;
+ this.triangleTripletCount = triangleTripletCount;
+ this.rectangleTripletCount = rectangleTripletCount;
+ this.tripletCount = tripletCount;
+ this.maximumDegree = maximumDegree;
+ this.maximumTriangleTriplets = maximumTriangleTriplets;
+ this.maximumRectangleTriplets = maximumRectangleTriplets;
+ this.maximumTriplets = maximumTriplets;
+ }
+
+ /**
+ * Get the number of vertices.
+ *
+ * @return number of vertices
+ */
+ public long getNumberOfVertices() {
+ return vertexCount;
+ }
+
+ /**
+ * Get the number of edges.
+ *
+ * @return number of edges
+ */
+ public long getNumberOfEdges() {
+ return edgeCount;
+ }
+
+ /**
+ * Get the number of triangle triplets.
+ *
+ * @return number of triangle triplets
+ */
+ public long getNumberOfTriangleTriplets() {
+ return triangleTripletCount;
+ }
+
+ /**
+ * Get the number of rectangle triplets.
+ *
+ * @return number of rectangle triplets
+ */
+ public long getNumberOfRectangleTriplets() {
+ return rectangleTripletCount;
+ }
+
+ /**
+ * Get the number of triplets.
+ *
+ * @return number of triplets
+ */
+ public long getNumberOfTriplets() {
+ return tripletCount;
+ }
+
+ /**
+ * Get the maximum degree.
+ *
+ * @return maximum degree
+ */
+ public long getMaximumDegree() {
+ return maximumDegree;
+ }
+
+ /**
+ * Get the maximum triangle triplets.
+ *
+ * @return maximum triangle triplets
+ */
+ public long getMaximumTriangleTriplets() {
+ return maximumTriangleTriplets;
+ }
+
+ /**
+ * Get the maximum rectangle triplets.
+ *
+ * @return maximum rectangle triplets
+ */
+ public long getMaximumRectangleTriplets() {
+ return maximumRectangleTriplets;
+ }
+
+ /**
+ * Get the maximum triplets.
+ *
+ * @return maximum triplets
+ */
+ public long getMaximumTriplets() {
+ return maximumTriplets;
+ }
+
+ @Override
+ public String toString() {
+ NumberFormat nf = NumberFormat.getInstance();
+
+ return "vertex count: " + nf.format(vertexCount)
+ + "; edge count: " + nf.format(edgeCount)
+ + "; triangle triplet count: " + nf.format(triangleTripletCount)
+ + "; rectangle triplet count: " + nf.format(rectangleTripletCount)
+ + "; triplet count: " + nf.format(tripletCount)
+ + "; maximum degree: " + nf.format(maximumDegree)
+ + "; maximum triangle triplets: " + nf.format(maximumTriangleTriplets)
+ + "; maximum rectangle triplets: " + nf.format(maximumRectangleTriplets)
+ + "; maximum triplets: " + nf.format(maximumTriplets);
+ }
+
+ @Override
+ public int hashCode() {
+ return new HashCodeBuilder()
+ .append(vertexCount)
+ .append(edgeCount)
+ .append(triangleTripletCount)
+ .append(rectangleTripletCount)
+ .append(tripletCount)
+ .append(maximumDegree)
+ .append(maximumTriangleTriplets)
+ .append(maximumRectangleTriplets)
+ .append(maximumTriplets)
+ .hashCode();
+ }
+
+ @Override
+ public boolean equals(Object obj) {
+ if (obj == null) { return false; }
+ if (obj == this) { return true; }
+ if (obj.getClass() != getClass()) { return false; }
+
+ Result rhs = (Result)obj;
+
+ return new EqualsBuilder()
+ .append(vertexCount, rhs.vertexCount)
+ .append(edgeCount, rhs.edgeCount)
+ .append(triangleTripletCount, rhs.triangleTripletCount)
+ .append(rectangleTripletCount, rhs.rectangleTripletCount)
+ .append(tripletCount, rhs.tripletCount)
+ .append(maximumDegree, rhs.maximumDegree)
+ .append(maximumTriangleTriplets, rhs.maximumTriangleTriplets)
+ .append(maximumRectangleTriplets, rhs.maximumRectangleTriplets)
+ .append(maximumTriplets, rhs.maximumTriplets)
+ .isEquals();
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/58850f29/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/library/metric/undirected/VertexMetrics.java
----------------------------------------------------------------------
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 3c26e43..d04fa7b 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
@@ -22,6 +22,7 @@ import org.apache.commons.lang3.builder.EqualsBuilder;
import org.apache.commons.lang3.builder.HashCodeBuilder;
import org.apache.flink.api.common.JobExecutionResult;
import org.apache.flink.api.common.accumulators.LongCounter;
+import org.apache.flink.api.common.accumulators.LongMaximum;
import org.apache.flink.api.common.io.RichOutputFormat;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.configuration.Configuration;
@@ -35,12 +36,17 @@ import org.apache.flink.types.LongValue;
import org.apache.flink.util.AbstractID;
import java.io.IOException;
+import java.text.NumberFormat;
import static org.apache.flink.api.common.ExecutionConfig.PARALLELISM_DEFAULT;
/**
- * Compute the number of vertices, number of edges, and number of triplets in
- * an undirected graph.
+ * Compute the following vertex metrics in an undirected graph:
+ * - number of vertices
+ * - number of edges
+ * - number of triplets
+ * - maximum degree
+ * - maximum number of triplets
*
* @param <K> graph ID type
* @param <VV> vertex value type
@@ -125,8 +131,10 @@ extends AbstractGraphAnalytic<K, VV, EV, Result> {
long vertexCount = res.getAccumulatorResult(id + "-0");
long edgeCount = res.getAccumulatorResult(id + "-1");
long tripletCount = res.getAccumulatorResult(id + "-2");
+ long maximumDegree = res.getAccumulatorResult(id + "-3");
+ long maximumTriplets = res.getAccumulatorResult(id + "-4");
- return new Result(vertexCount, edgeCount / 2, tripletCount);
+ return new Result(vertexCount, edgeCount / 2, tripletCount, maximumDegree, maximumTriplets);
}
/**
@@ -141,13 +149,15 @@ extends AbstractGraphAnalytic<K, VV, EV, Result> {
private long vertexCount;
private long edgeCount;
private long tripletCount;
+ private long maximumDegree;
+ private long maximumTriplets;
/**
* This helper class collects vertex metrics by scanning over and
* discarding elements from the given DataSet.
*
* The unique id is required because Flink's accumulator namespace is
- * among all operators.
+ * shared among all operators.
*
* @param id unique string used for accumulator names
*/
@@ -164,10 +174,13 @@ extends AbstractGraphAnalytic<K, VV, EV, Result> {
@Override
public void writeRecord(Vertex<T, LongValue> record) throws IOException {
long degree = record.f1.getValue();
+ long triplets = degree * (degree - 1) / 2;
vertexCount++;
edgeCount += degree;
- tripletCount += degree * (degree - 1) / 2;
+ tripletCount += triplets;
+ maximumDegree = Math.max(maximumDegree, degree);
+ maximumTriplets = Math.max(maximumTriplets, triplets);
}
@Override
@@ -175,6 +188,8 @@ extends AbstractGraphAnalytic<K, VV, EV, Result> {
getRuntimeContext().addAccumulator(id + "-0", new LongCounter(vertexCount));
getRuntimeContext().addAccumulator(id + "-1", new LongCounter(edgeCount));
getRuntimeContext().addAccumulator(id + "-2", new LongCounter(tripletCount));
+ getRuntimeContext().addAccumulator(id + "-3", new LongMaximum(maximumDegree));
+ getRuntimeContext().addAccumulator(id + "-4", new LongMaximum(maximumTriplets));
}
}
@@ -185,11 +200,15 @@ extends AbstractGraphAnalytic<K, VV, EV, Result> {
private long vertexCount;
private long edgeCount;
private long tripletCount;
+ private long maximumDegree;
+ private long maximumTriplets;
- public Result(long vertexCount, long edgeCount, long tripletCount) {
+ public Result(long vertexCount, long edgeCount, long tripletCount, long maximumDegree, long maximumTriplets) {
this.vertexCount = vertexCount;
this.edgeCount = edgeCount;
this.tripletCount = tripletCount;
+ this.maximumDegree = maximumDegree;
+ this.maximumTriplets = maximumTriplets;
}
/**
@@ -219,11 +238,33 @@ extends AbstractGraphAnalytic<K, VV, EV, Result> {
return tripletCount;
}
+ /**
+ * Get the maximum degree.
+ *
+ * @return maximum degree
+ */
+ public long getMaximumDegree() {
+ return maximumDegree;
+ }
+
+ /**
+ * Get the maximum triplets.
+ *
+ * @return maximum triplets
+ */
+ public long getMaximumTriplets() {
+ return maximumTriplets;
+ }
+
@Override
public String toString() {
- return "vertex count: " + vertexCount
- + ", edge count:" + edgeCount
- + ", triplet count: " + tripletCount;
+ NumberFormat nf = NumberFormat.getInstance();
+
+ return "vertex count: " + nf.format(vertexCount)
+ + "; edge count: " + nf.format(edgeCount)
+ + "; triplet count: " + nf.format(tripletCount)
+ + "; maximum degree: " + nf.format(maximumDegree)
+ + "; maximum triplets: " + nf.format(maximumTriplets);
}
@Override
@@ -232,6 +273,8 @@ extends AbstractGraphAnalytic<K, VV, EV, Result> {
.append(vertexCount)
.append(edgeCount)
.append(tripletCount)
+ .append(maximumDegree)
+ .append(maximumTriplets)
.hashCode();
}
@@ -247,6 +290,8 @@ extends AbstractGraphAnalytic<K, VV, EV, Result> {
.append(vertexCount, rhs.vertexCount)
.append(edgeCount, rhs.edgeCount)
.append(tripletCount, rhs.tripletCount)
+ .append(maximumDegree, rhs.maximumDegree)
+ .append(maximumTriplets, rhs.maximumTriplets)
.isEquals();
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/58850f29/flink-libraries/flink-gelly/src/test/java/org/apache/flink/graph/library/metric/directed/EdgeMetricsTest.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-gelly/src/test/java/org/apache/flink/graph/library/metric/directed/EdgeMetricsTest.java b/flink-libraries/flink-gelly/src/test/java/org/apache/flink/graph/library/metric/directed/EdgeMetricsTest.java
new file mode 100644
index 0000000..af5a154
--- /dev/null
+++ b/flink-libraries/flink-gelly/src/test/java/org/apache/flink/graph/library/metric/directed/EdgeMetricsTest.java
@@ -0,0 +1,90 @@
+/*
+ * 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.library.metric.directed;
+
+import org.apache.commons.math3.util.CombinatoricsUtils;
+import org.apache.flink.graph.asm.AsmTestBase;
+import org.apache.flink.graph.library.metric.directed.EdgeMetrics.Result;
+import org.apache.flink.types.IntValue;
+import org.apache.flink.types.LongValue;
+import org.apache.flink.types.NullValue;
+import org.junit.Test;
+
+import static org.junit.Assert.assertEquals;
+
+public class EdgeMetricsTest
+extends AsmTestBase {
+
+ @Test
+ public void testWithSimpleGraph()
+ throws Exception {
+ Result expectedResult = new Result(6, 7, 2, 6, 13, 4, 2, 3, 1, 3, 6);
+
+ Result edgeMetrics = new EdgeMetrics<IntValue, NullValue, NullValue>()
+ .run(directedSimpleGraph)
+ .execute();
+
+ assertEquals(expectedResult, edgeMetrics);
+ }
+
+ @Test
+ public void testWithCompleteGraph()
+ throws Exception {
+ long expectedDegree = completeGraphVertexCount - 1;
+ long expectedEdges = completeGraphVertexCount * expectedDegree;
+ long expectedMaximumTriplets = CombinatoricsUtils.binomialCoefficient((int)expectedDegree, 2);
+ long expectedTriplets = completeGraphVertexCount * expectedMaximumTriplets;
+
+ Result expectedResult = new Result(completeGraphVertexCount, expectedEdges, expectedTriplets / 3, 2 * expectedTriplets / 3, expectedTriplets,
+ expectedDegree, expectedDegree, expectedDegree,
+ expectedMaximumTriplets, expectedMaximumTriplets, expectedMaximumTriplets);
+
+ Result edgeMetrics = new EdgeMetrics<LongValue, NullValue, NullValue>()
+ .run(completeGraph)
+ .execute();
+
+ assertEquals(expectedResult, edgeMetrics);
+ }
+
+ @Test
+ public void testWithEmptyGraph()
+ throws Exception {
+ Result expectedResult;
+
+ expectedResult = new Result(0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0);
+
+ Result withoutZeroDegreeVertices = new EdgeMetrics<LongValue, NullValue, NullValue>()
+ .run(emptyGraph)
+ .execute();
+
+ assertEquals(withoutZeroDegreeVertices, expectedResult);
+ }
+
+ @Test
+ public void testWithRMatGraph()
+ throws Exception {
+ Result expectedResult = new Result(902, 12009, 107817, 315537, 1003442, 463, 334, 342, 820, 3822, 106953);
+
+ Result withoutZeroDegreeVertices = new EdgeMetrics<LongValue, NullValue, NullValue>()
+ .run(directedRMatGraph)
+ .execute();
+
+ assertEquals(expectedResult, withoutZeroDegreeVertices);
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/58850f29/flink-libraries/flink-gelly/src/test/java/org/apache/flink/graph/library/metric/directed/VertexMetricsTest.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-gelly/src/test/java/org/apache/flink/graph/library/metric/directed/VertexMetricsTest.java b/flink-libraries/flink-gelly/src/test/java/org/apache/flink/graph/library/metric/directed/VertexMetricsTest.java
index 8145abd..e4362c0 100644
--- a/flink-libraries/flink-gelly/src/test/java/org/apache/flink/graph/library/metric/directed/VertexMetricsTest.java
+++ b/flink-libraries/flink-gelly/src/test/java/org/apache/flink/graph/library/metric/directed/VertexMetricsTest.java
@@ -34,10 +34,10 @@ extends AsmTestBase {
@Test
public void testWithSimpleGraph()
throws Exception {
- Result expectedResult = new Result(6, 7, 13);
+ Result expectedResult = new Result(6, 7, 13, 4, 2, 3, 6);
Result vertexMetrics = new VertexMetrics<IntValue, NullValue, NullValue>()
- .run(undirectedSimpleGraph)
+ .run(directedSimpleGraph)
.execute();
assertEquals(expectedResult, vertexMetrics);
@@ -47,10 +47,11 @@ extends AsmTestBase {
public void testWithCompleteGraph()
throws Exception {
long expectedDegree = completeGraphVertexCount - 1;
- long expectedEdges = completeGraphVertexCount * expectedDegree / 2;
- long expectedTriplets = completeGraphVertexCount * CombinatoricsUtils.binomialCoefficient((int)expectedDegree, 2);
+ long expectedEdges = completeGraphVertexCount * expectedDegree;
+ long expectedMaximumTriplets = CombinatoricsUtils.binomialCoefficient((int)expectedDegree, 2);
+ long expectedTriplets = completeGraphVertexCount * expectedMaximumTriplets;
- Result expectedResult = new Result(completeGraphVertexCount, expectedEdges, expectedTriplets);
+ Result expectedResult = new Result(completeGraphVertexCount, expectedEdges, expectedTriplets, expectedDegree, expectedDegree, expectedDegree, expectedMaximumTriplets);
Result vertexMetrics = new VertexMetrics<LongValue, NullValue, NullValue>()
.run(completeGraph)
@@ -64,7 +65,7 @@ extends AsmTestBase {
throws Exception {
Result expectedResult;
- expectedResult = new Result(0, 0, 0);
+ expectedResult = new Result(0, 0, 0, 0, 0, 0, 0);
Result withoutZeroDegreeVertices = new VertexMetrics<LongValue, NullValue, NullValue>()
.setIncludeZeroDegreeVertices(false)
@@ -73,7 +74,7 @@ extends AsmTestBase {
assertEquals(withoutZeroDegreeVertices, expectedResult);
- expectedResult = new Result(3, 0, 0);
+ expectedResult = new Result(3, 0, 0, 0, 0, 0, 0);
Result withZeroDegreeVertices = new VertexMetrics<LongValue, NullValue, NullValue>()
.setIncludeZeroDegreeVertices(true)
@@ -86,10 +87,10 @@ extends AsmTestBase {
@Test
public void testWithRMatGraph()
throws Exception {
- Result expectedResult = new Result(902, 10442, 1003442);
+ Result expectedResult = new Result(902, 12009, 1003442, 463, 334, 342, 106953);
Result withoutZeroDegreeVertices = new VertexMetrics<LongValue, NullValue, NullValue>()
- .run(undirectedRMatGraph)
+ .run(directedRMatGraph)
.execute();
assertEquals(expectedResult, withoutZeroDegreeVertices);
http://git-wip-us.apache.org/repos/asf/flink/blob/58850f29/flink-libraries/flink-gelly/src/test/java/org/apache/flink/graph/library/metric/undirected/EdgeMetricsTest.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-gelly/src/test/java/org/apache/flink/graph/library/metric/undirected/EdgeMetricsTest.java b/flink-libraries/flink-gelly/src/test/java/org/apache/flink/graph/library/metric/undirected/EdgeMetricsTest.java
new file mode 100644
index 0000000..b300d66
--- /dev/null
+++ b/flink-libraries/flink-gelly/src/test/java/org/apache/flink/graph/library/metric/undirected/EdgeMetricsTest.java
@@ -0,0 +1,89 @@
+/*
+ * 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.library.metric.undirected;
+
+import org.apache.commons.math3.util.CombinatoricsUtils;
+import org.apache.flink.graph.asm.AsmTestBase;
+import org.apache.flink.graph.library.metric.undirected.EdgeMetrics.Result;
+import org.apache.flink.types.IntValue;
+import org.apache.flink.types.LongValue;
+import org.apache.flink.types.NullValue;
+import org.junit.Test;
+
+import static org.junit.Assert.assertEquals;
+
+public class EdgeMetricsTest
+extends AsmTestBase {
+
+ @Test
+ public void testWithSimpleGraph()
+ throws Exception {
+ Result expectedResult = new Result(6, 7, 2, 6, 13, 4, 1, 3, 6);
+
+ Result edgeMetrics = new EdgeMetrics<IntValue, NullValue, NullValue>()
+ .run(undirectedSimpleGraph)
+ .execute();
+
+ assertEquals(expectedResult, edgeMetrics);
+ }
+
+ @Test
+ public void testWithCompleteGraph()
+ throws Exception {
+ long expectedDegree = completeGraphVertexCount - 1;
+ long expectedEdges = completeGraphVertexCount * expectedDegree / 2;
+ long expectedMaximumTriplets = CombinatoricsUtils.binomialCoefficient((int)expectedDegree, 2);
+ long expectedTriplets = completeGraphVertexCount * expectedMaximumTriplets;
+
+ Result expectedResult = new Result(completeGraphVertexCount, expectedEdges, expectedTriplets / 3, 2 * expectedTriplets / 3, expectedTriplets,
+ expectedDegree, expectedMaximumTriplets, expectedMaximumTriplets, expectedMaximumTriplets);
+
+ Result edgeMetrics = new EdgeMetrics<LongValue, NullValue, NullValue>()
+ .run(completeGraph)
+ .execute();
+
+ assertEquals(expectedResult, edgeMetrics);
+ }
+
+ @Test
+ public void testWithEmptyGraph()
+ throws Exception {
+ Result expectedResult;
+
+ expectedResult = new Result(0, 0, 0, 0, 0, 0, 0, 0, 0);
+
+ Result withoutZeroDegreeVertices = new EdgeMetrics<LongValue, NullValue, NullValue>()
+ .run(emptyGraph)
+ .execute();
+
+ assertEquals(withoutZeroDegreeVertices, expectedResult);
+ }
+
+ @Test
+ public void testWithRMatGraph()
+ throws Exception {
+ Result expectedResult = new Result(902, 10442, 107817, 315537, 1003442, 463, 820, 3822, 106953);
+
+ Result withoutZeroDegreeVertices = new EdgeMetrics<LongValue, NullValue, NullValue>()
+ .run(undirectedRMatGraph)
+ .execute();
+
+ assertEquals(expectedResult, withoutZeroDegreeVertices);
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/58850f29/flink-libraries/flink-gelly/src/test/java/org/apache/flink/graph/library/metric/undirected/VertexMetricsTest.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-gelly/src/test/java/org/apache/flink/graph/library/metric/undirected/VertexMetricsTest.java b/flink-libraries/flink-gelly/src/test/java/org/apache/flink/graph/library/metric/undirected/VertexMetricsTest.java
index a36ca94..8f7e1da 100644
--- a/flink-libraries/flink-gelly/src/test/java/org/apache/flink/graph/library/metric/undirected/VertexMetricsTest.java
+++ b/flink-libraries/flink-gelly/src/test/java/org/apache/flink/graph/library/metric/undirected/VertexMetricsTest.java
@@ -34,7 +34,7 @@ extends AsmTestBase {
@Test
public void testWithSimpleGraph()
throws Exception {
- Result expectedResult = new Result(6, 7, 13);
+ Result expectedResult = new Result(6, 7, 13, 4, 6);
Result vertexMetrics = new VertexMetrics<IntValue, NullValue, NullValue>()
.run(undirectedSimpleGraph)
@@ -48,9 +48,11 @@ extends AsmTestBase {
throws Exception {
long expectedDegree = completeGraphVertexCount - 1;
long expectedEdges = completeGraphVertexCount * expectedDegree / 2;
- long expectedTriplets = completeGraphVertexCount * CombinatoricsUtils.binomialCoefficient((int)expectedDegree, 2);
+ long expectedMaximumTriplets = CombinatoricsUtils.binomialCoefficient((int)expectedDegree, 2);
+ long expectedTriplets = completeGraphVertexCount * expectedMaximumTriplets;
- Result expectedResult = new Result(completeGraphVertexCount, expectedEdges, expectedTriplets);
+ Result expectedResult = new Result(completeGraphVertexCount, expectedEdges, expectedTriplets,
+ expectedDegree, expectedMaximumTriplets);
Result vertexMetrics = new VertexMetrics<LongValue, NullValue, NullValue>()
.run(completeGraph)
@@ -64,7 +66,7 @@ extends AsmTestBase {
throws Exception {
Result expectedResult;
- expectedResult = new Result(0, 0, 0);
+ expectedResult = new Result(0, 0, 0, 0, 0);
Result withoutZeroDegreeVertices = new VertexMetrics<LongValue, NullValue, NullValue>()
.setIncludeZeroDegreeVertices(false)
@@ -73,7 +75,7 @@ extends AsmTestBase {
assertEquals(withoutZeroDegreeVertices, expectedResult);
- expectedResult = new Result(3, 0, 0);
+ expectedResult = new Result(3, 0, 0, 0, 0);
Result withZeroDegreeVertices = new VertexMetrics<LongValue, NullValue, NullValue>()
.setIncludeZeroDegreeVertices(true)
@@ -86,7 +88,7 @@ extends AsmTestBase {
@Test
public void testWithRMatGraph()
throws Exception {
- Result expectedResult = new Result(902, 10442, 1003442);
+ Result expectedResult = new Result(902, 10442, 1003442, 463, 106953);
Result withoutZeroDegreeVertices = new VertexMetrics<LongValue, NullValue, NullValue>()
.run(undirectedRMatGraph)