[jira] [Comment Edited] (FLINK-2044) Implementation of Gelly Algorithm HITS

["Ahamd Javid (JIRA)" <ji...@apache.org>]

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

Ahamd Javid edited comment on FLINK-2044 at 5/19/15 11:38 AM:
--------------------------------------------------------------

We are two Students part of the DIMA Big data Project and would like to implement HITS algorithm 
and we are new with flink Gelly API and we have created a small piece of code.
If anyone could give us comments warmly welcome.
[This is our github issue link](https://github.com/JavidMayar/flink/commit/724be153d751d409e149a874fbd962e0fd52e782)

this is a sample of our code :

public class Hits_Class {

    static Graph< Long, Double, Double > mainGraph;

    static int iterationValue;


    public Hits_Class( Graph< Long, Double, Double > myGraph, int iteration ){

        mainGraph = myGraph;
        iterationValue = iteration;

    }

    /**
     * This function compute the hub and authority values from graph
     * and update the graph in each iteration with hubValues and authorityValues
     *
     * return Dataset< Tuple3 > with IDs, Hub Values, Authority Values
     *
     */
    public static DataSet<Tuple3<Long, Double, Double>> run(){

        DataSet<Tuple2<Long, Double>> authorityValues = null;
        DataSet<Tuple2<Long, Double>> hubValues = null;
        Graph<Long, Double, Double> updatedGraph = mainGraph;

        for(int i = 0; i < iterationValue ; i++ ) {
            authorityValues = authorityUpdateRule(updatedGraph);


            updatedGraph = updateGraph(updatedGraph, authorityValues);

            hubValues = hubUpdateRule(updatedGraph);


            updatedGraph = updateGraph(updatedGraph, hubValues);

        }


        DataSet<Tuple3<Long, Double, Double> > HubAuthory=hubValues.join(authorityValues).where(0).equalTo(0).map(new HubAuthorityScore());
        return  HubAuthory;


    }

    /**
     * This Class have the function of map which join the hub and authority DataSets to one Tuple3 dataSet Example < ID , HubValue, AuthorityValue >
     * @param1 HubValueDataSet
     * @param2 AuthorityValueDatSet
     *
     * return Tuple3DataSet
     */

    public static class HubAuthorityScore implements MapFunction<Tuple2<Tuple2<Long,Double>,Tuple2<Long,Double>>,Tuple3<Long, Double,Double>>{

        @Override
        public Tuple3<Long, Double, Double> map(Tuple2<Tuple2<Long, Double>, Tuple2<Long, Double>> value) throws Exception {
            return new Tuple3<Long, Double, Double>(value.f0.f0,value.f0.f1,value.f1.f1);
        }
    }

    /**
     * This function create the authority DataSet from given graph
     * @param1 graph
     *
     * return DataSet<Tuple2> with Authority Values for each node.
     */

    public static DataSet<Tuple2<Long, Double>> authorityUpdateRule( Graph< Long, Double, Double> myGraph ){

        DataSet<Tuple2<Long, Double>> verticesWithSum = myGraph.reduceOnNeighbors(new SumValues(), EdgeDirection.IN);
        return verticesWithSum;

    }

    /**
     * This function create the hubValues DataSet from given graph
     * @param1 graph
     *
     * return DataSet<Tuple2> with Hub Values for each node.
     */

    public static DataSet<Tuple2<Long, Double>> hubUpdateRule( Graph< Long, Double, Double> myGraph ){

        DataSet<Tuple2<Long, Double>> verticesWithSum = myGraph.reduceOnNeighbors(new SumValues(), EdgeDirection.OUT);
        return verticesWithSum;
    }

    /**
     * This function update the graph vertices values by give DatSet<Tuple2>
     * @param1 graph
     *
     * return graph with updated vertices Values.
     */

    public static Graph<Long, Double, Double > updateGraph( Graph<Long, Double, Double> myGraph, DataSet<Tuple2< Long, Double >> myDataSet ){

        Graph<Long, Double, Double> updatedGraph = myGraph.joinWithVertices(myDataSet,new MapFunction<Tuple2<Double, Double>, Double>() {

            public Double map(Tuple2<Double, Double> value) {
                return value.f0 = value.f1;
            }
        });

        return updatedGraph;

    }

    // function to sum the neighbor values
    static final class SumValues implements ReduceNeighborsFunction<Double> {

        @Override
        public Double reduceNeighbors(Double firstNeighbor, Double secondNeighbor) {
            return firstNeighbor + secondNeighbor;
        }
    }
}



our main class with test data:
public class HitsMain {

    public static void main(String args [] ) throws Exception {

        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

        DataSet<Vertex<Long, Double>> vertices = getVertices(env);

        DataSet<Edge<Long, Double>> edges = getEdges(env);

        //create a graph from vertices(list of vertices created with getVertices function)
        // and edges(list of Edges created with getEdges function).
        Graph<Long, Double, Double> graph = Graph.fromDataSet(vertices, edges, env);

        // add  graph to Hits_Class with iteration value.
        DataSet<Tuple3<Long, Double, Double>> hubAndAuthority = new Hits_Class(graph,3).run();


        // print the retrieved data set.
        hubAndAuthority.print();

        env.execute("Hits Example");

    }

    /**
     * this function create a set of links (Edges).
     * @param en (environment variable)
     * @return Datset of Edges.
     */

    public static DataSet<Edge<Long, Double> > getEdges(ExecutionEnvironment en){
        return en.generateSequence(1,1).flatMap(new FlatMapFunction<Long, Edge<Long,Double>>() {
            @Override
            public void flatMap(Long value, Collector<Edge<Long,Double>> out) throws Exception {

                out.collect(new Edge<Long, Double>(1l,2l, 0.0));
                out.collect(new Edge<Long, Double>(2l,1l, 0.0));
                out.collect(new Edge<Long, Double>(2l,3l, 0.0));
                out.collect(new Edge<Long, Double>(3l,1l, 0.0));
                out.collect(new Edge<Long, Double>(3l,2l, 0.0));
            }
        });

    }

    /**
     * this function create a set of nodes (Vertices).
     * @param en (environment variable)
     * @return DataSet of Vertices.
     */
    public static DataSet<Vertex<Long, Double>> getVertices(ExecutionEnvironment en){
        return en.generateSequence(1,1).flatMap(new FlatMapFunction<Long, Vertex<Long,Double>>() {
            @Override
            public void flatMap(Long value, Collector<Vertex<Long,Double>> out) throws Exception {

                out.collect(new Vertex<Long, Double>(1l, 1.0));
                out.collect(new Vertex<Long, Double>(2l, 1.0));
                out.collect(new Vertex<Long, Double>(3l, 1.0));
            }
        });

    }
}








was (Author: javidmayar):
We are two Students part of the DIMA Big data Project and would like to implement HITS algorithm 
and we are new with flink Gelly API and we have created a small piece of code.
If anyone could give us comments warmly welcome.
[This is our github issue link](https://github.com/JavidMayar/flink/commit/724be153d751d409e149a874fbd962e0fd52e782)




> Implementation of Gelly Algorithm HITS
> --------------------------------------
>
>                 Key: FLINK-2044
>                 URL: https://issues.apache.org/jira/browse/FLINK-2044
>             Project: Flink
>          Issue Type: Improvement
>          Components: Gelly
>            Reporter: Ahamd Javid
>            Priority: Minor
>
> Implementation of Hits Algorithm in Gelly API using Java 



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