[ 
https://issues.apache.org/jira/browse/FLINK-1528?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=14351544#comment-14351544
 ] 

ASF GitHub Bot commented on FLINK-1528:
---------------------------------------

Github user vasia commented on a diff in the pull request:

    https://github.com/apache/flink/pull/420#discussion_r25995666
  
    --- Diff: 
flink-staging/flink-gelly/src/main/java/org/apache/flink/graph/example/LocalClusteringCoefficientExample.java
 ---
    @@ -0,0 +1,205 @@
    +/*
    + * 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.example;
    +
    +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.tuple.Tuple2;
    +import org.apache.flink.graph.Edge;
    +import org.apache.flink.graph.EdgeDirection;
    +import org.apache.flink.graph.EdgesFunction;
    +import org.apache.flink.graph.Graph;
    +import org.apache.flink.graph.NeighborsFunctionWithVertexValue;
    +import org.apache.flink.graph.Vertex;
    +import org.apache.flink.graph.example.utils.LocalClusteringCoefficientData;
    +import org.apache.flink.types.NullValue;
    +
    +import java.util.HashSet;
    +
    +public class LocalClusteringCoefficientExample implements 
ProgramDescription {
    +
    +   // 
--------------------------------------------------------------------------------------------
    +   //  Program
    +   // 
--------------------------------------------------------------------------------------------
    +
    +   public static void main (String [] args) throws Exception {
    +
    +           if(!parseParameters(args)) {
    +                   return;
    +           }
    +
    +           ExecutionEnvironment env = 
ExecutionEnvironment.getExecutionEnvironment();
    +
    +           DataSet<Edge<Long, NullValue>> edges = getEdgeDataSet(env);
    +           Graph<Long, NullValue, NullValue> graph = 
Graph.fromDataSet(edges, env);
    +
    +           // Get the neighbors of each vertex in a HashSet
    +           DataSet<Tuple2<Long, HashSet<Long>>> neighborhoods = graph
    +                           .reduceOnEdges(new NeighborhoodEdgesFunction(), 
EdgeDirection.OUT);
    +
    +           // Construct a new graph where the neighborhood is the vertex 
value
    +           Graph<Long, HashSet<Long>, NullValue> newGraph = graph
    +                           .mapVertices(new EmptyVertexMapFunction())
    +                           .joinWithVertices(neighborhoods, new 
NeighborhoodVertexMapFunction());
    +
    +           // Calculate clustering coefficient
    +           DataSet<Tuple2<Long, Double>> clusteringCoefficients = newGraph
    +                           .reduceOnNeighbors(new 
ClusteringCoefficientNeighborsFunction(), EdgeDirection.OUT);
    +
    +           // Emit results
    +           if(fileOutput) {
    +                   clusteringCoefficients.writeAsCsv(outputPath, "\n", 
",");
    +           } else {
    +                   clusteringCoefficients.print();
    +           }
    +
    +           env.execute("Local Clustering Coefficient Example");
    +   }
    +
    +   // 
--------------------------------------------------------------------------------------------
    +   //  Clustering Coefficient Functions
    +   // 
--------------------------------------------------------------------------------------------
    +
    +   private static final class NeighborhoodEdgesFunction
    +                   implements EdgesFunction<Long, NullValue, Tuple2<Long, 
HashSet<Long>>> {
    +
    +           @Override
    +           public Tuple2<Long, HashSet<Long>> iterateEdges(
    +                           Iterable<Tuple2<Long, Edge<Long, NullValue>>> 
edges) throws Exception {
    +
    +                   Long vertexId = null;
    +                   HashSet<Long> neighbors = new HashSet<Long>();
    +
    +                   for (Tuple2<Long, Edge<Long, NullValue>> edge : edges) {
    +                           vertexId = edge.f0;
    +                           neighbors.add(edge.f1.getTarget());
    +                   }
    +
    +                   return new Tuple2<Long, HashSet<Long>>(vertexId, 
neighbors);
    +           }
    +   }
    +
    +   private static final class EmptyVertexMapFunction
    +                   implements MapFunction<Vertex<Long, NullValue>, 
HashSet<Long>> {
    +
    +           @Override
    +           public HashSet<Long> map(Vertex<Long, NullValue> arg) throws 
Exception {
    +                   return new HashSet<Long>();
    +           }
    +   }
    +
    +   private static final class NeighborhoodVertexMapFunction
    +                   implements MapFunction<Tuple2<HashSet<Long>, 
HashSet<Long>>, HashSet<Long>> {
    +
    +           @Override
    +           public HashSet<Long> map(Tuple2<HashSet<Long>, HashSet<Long>> 
arg) throws Exception {
    +                   return arg.f1;
    +           }
    +   }
    +
    +   private static final class ClusteringCoefficientNeighborsFunction
    +                   implements NeighborsFunctionWithVertexValue<Long, 
HashSet<Long>, NullValue, Tuple2<Long, Double>> {
    +
    +           @Override
    +           public Tuple2<Long, Double> iterateNeighbors(Vertex<Long, 
HashSet<Long>> vertex,
    +                           Iterable<Tuple2<Edge<Long, NullValue>, 
Vertex<Long, HashSet<Long>>>> neighbors) throws Exception {
    +
    +                   int deg = vertex.getValue().size();
    +                   int e = 0;
    +
    +                   // Calculate common neighbor count (e)
    +                   for (Tuple2<Edge<Long, NullValue>, Vertex<Long, 
HashSet<Long>>> neighbor : neighbors) {
    +                           // Iterate neighbor's neighbors
    +                           for (Long nn : neighbor.f1.getValue()) {
    +                                   if (vertex.getValue().contains(nn)) {
    +                                           e++;
    +                                   }
    +                           }
    +                   }
    +
    +                   // We assume an undirected graph, so we need to divide 
e here
    +                   e /= 2;
    --- End diff --
    
    I don't get the reasoning behind this.. Do you assume an undirected graph 
as input? or you convert the input to undirected? in any case, I think we 
should add a comment in the beginning of the example to define the behavior.


> Add local clustering coefficient library method and example
> -----------------------------------------------------------
>
>                 Key: FLINK-1528
>                 URL: https://issues.apache.org/jira/browse/FLINK-1528
>             Project: Flink
>          Issue Type: Task
>          Components: Gelly
>            Reporter: Vasia Kalavri
>            Assignee: Daniel Bali
>
> Add a gelly library method and example to compute the local clustering 
> coefficient.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

Reply via email to