[
https://issues.apache.org/jira/browse/FLINK-3945?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15300093#comment-15300093
]
ASF GitHub Bot commented on FLINK-3945:
---------------------------------------
Github user vasia commented on a diff in the pull request:
https://github.com/apache/flink/pull/2021#discussion_r64577865
--- Diff:
flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/asm/degree/annotate/directed/EdgeDegreesPair.java
---
@@ -0,0 +1,81 @@
+/*
+ * 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.asm.degree.annotate.directed;
+
+import org.apache.flink.api.common.ExecutionConfig;
+import
org.apache.flink.api.common.operators.base.JoinOperatorBase.JoinHint;
+import org.apache.flink.api.java.DataSet;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.api.java.tuple.Tuple3;
+import org.apache.flink.graph.Edge;
+import org.apache.flink.graph.Graph;
+import org.apache.flink.graph.GraphAlgorithm;
+import org.apache.flink.graph.Vertex;
+import
org.apache.flink.graph.asm.degree.annotate.DegreeAnnotationFunctions.JoinEdgeDegreeWithVertexDegree;
+import
org.apache.flink.graph.asm.degree.annotate.directed.VertexDegrees.Degrees;
+
+/**
+ * Annotates edges of a directed graph with the degree, out-degree, and
+ * in-degree of both the source and target vertices.
+ *
+ * @param <K> ID type
+ * @param <VV> vertex value type
+ * @param <EV> edge value type
+ */
+public class EdgeDegreesPair<K, VV, EV>
+implements GraphAlgorithm<K, VV, EV, DataSet<Edge<K, Tuple3<EV, Degrees,
Degrees>>>> {
+
+ // Optional configuration
+ private int parallelism = ExecutionConfig.PARALLELISM_UNKNOWN;
+
+ /**
+ * Override the operator parallelism.
+ *
+ * @param parallelism operator parallelism
+ * @return this
+ */
+ public EdgeDegreesPair<K, VV, EV> setParallelism(int parallelism) {
+ this.parallelism = parallelism;
+
+ return this;
+ }
+
+ @Override
+ public DataSet<Edge<K, Tuple3<EV, Degrees, Degrees>>> run(Graph<K, VV,
EV> input)
+ throws Exception {
+ // s, t, d(s)
+ DataSet<Edge<K, Tuple2<EV, Degrees>>> edgeSourceDegrees = input
+ .run(new EdgeSourceDegrees<K, VV, EV>()
+ .setParallelism(parallelism));
+
+ // t, d(t)
+ DataSet<Vertex<K, Degrees>> vertexDegrees = input
+ .run(new VertexDegrees<K, VV, EV>()
+ .setParallelism(parallelism));
+
+ // s, t, (d(s), d(t))
+ return edgeSourceDegrees
+ .join(vertexDegrees, JoinHint.REPARTITION_HASH_SECOND)
--- End diff --
I'm not sure whether it's a good idea to always provide a join hint here.
It's true that the vertex set is _usually_ much smaller than the edge set, but
what if that's not true? Can we at least provide a way to override this
configuration?
> Degree annotation for directed graphs
> -------------------------------------
>
> Key: FLINK-3945
> URL: https://issues.apache.org/jira/browse/FLINK-3945
> Project: Flink
> Issue Type: Improvement
> Components: Gelly
> Affects Versions: 1.1.0
> Reporter: Greg Hogan
> Assignee: Greg Hogan
> Priority: Minor
> Fix For: 1.1.0
>
>
> There is a third degree count for vertices in directed graphs which is the
> distinct count of out- and in-neighbors. This also adds edge annotation of
> the vertex degrees for directed graphs.
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
