/**

 *  numPages缺省15个测试页面

 *

 *  EDGES表示从一个pageId指向相连的另外一个pageId

 */

public class PageRankData {

    public static final Object[][] EDGES = {

            {1L, 2L},

            {1L, 15L},

            {2L, 3L},

            {2L, 4L},

            {2L, 5L},

            {2L, 6L},

            {2L, 7L},

            {3L, 13L},

            {4L, 2L},

            {5L, 11L},

            {5L, 12L},

            {6L, 1L},

            {6L, 7L},

            {6L, 8L},

            {7L, 1L},

            {7L, 8L},

            {8L, 1L},

            {8L, 9L},

            {8L, 10L},

            {9L, 14L},

            {9L, 1L},

            {10L, 1L},

            {10L, 13L},

            {11L, 12L},

            {11L, 1L},

            {12L, 1L},

            {13L, 14L},

            {14L, 12L},

            {15L, 1L},

    };

    private static int numPages = 15;

    public static DataSet<Tuple2<Long, Long>> getDefaultEdgeDataSet(ExecutionEnvironment env) {

        List<Tuple2<Long, Long>> edges = new ArrayList<Tuple2<Long, Long>>();

        for (Object[] e : EDGES) {

            edges.add(new Tuple2<Long, Long>((Long) e[0], (Long) e[1]));

        }

        return env.fromCollection(edges);

    }

    public static DataSet<Long> getDefaultPagesDataSet(ExecutionEnvironment env) {

        return env.generateSequence(1, 15);

    }

    public static int getNumberOfPages() {

        return numPages;

    }

}

/**

 * @Description: 使用批量迭代的页面排名算法的基本实现。

 * 此实现需要一组页面和一组有向链接作为输入，并按如下方式工作。

 * 在每次迭代中，每个页面的等级均匀分布到它指向的所有页面。每个页面收集指向它的所有页面的部分等级，对它们求和，并对总和应用阻尼因子。结果是页面的新排名。使用所有页面的新等级开始新的迭代。该实现在固定次数的迭代之后终止。

 * 这是页面排名算法的维基百科条目。

 *

 * 输入文件是纯文本文件，必须格式如下：

 *

 * 页面表示为由新行字符分隔的（长）ID。

 * 例如，"1\n2\n12\n42\n63"给出五个页面ID为1,2,12,42和63的页面。

 * 链接表示为页面ID对，由空格字符分隔。链接由换行符分隔。

 * 例如，"1 2\n2 12\n1 12\n42 63"给出四个（定向）链接（1） - >（2），（2） - >（12），（1） - >（12）和（42） - >（63）。

 * 对于这个简单的实现，要求每个页面至少有一个传入链接和一个传出链接（页面可以指向自身）。

 * 用法：PageRankBasic --pages <path> --links <path> --output <path> --numPages <n> --iterations <n>

 *  如果未提供参数，则使用{@link PageRankData}中的默认数据和10次迭代运行程序。

 *

 **/

public class PageRank {

    //阻尼系数

    private static final double DAMPENING_FACTOR = 0.85;

    //收敛阈值.

    private static final double EPSILON = 0.0001;

    private static DataSet<Long> getPagesDataSet(ExecutionEnvironment env, ParameterTool params) {

        if (params.has("pages")) {

            return env.readCsvFile(params.get("params"))

                    .fieldDelimiter(" ")

                    .lineDelimiter("\n")

                    .types(Long.class)

                    .map(new MapFunction<Tuple1<Long>, Long>() {

                        @Override

                        public Long map(Tuple1<Long> value) throws Exception {

                            return value.f0;

                        }

                    });

        } else {

            System.out.println("Executing PageRank example with default pages data set.");

            System.out.println("Use --pages to specify file input.");

            return PageRankData.getDefaultPagesDataSet(env);

        }

    }

    private static DataSet<Tuple2<Long, Long>> getLinksDataSet(ExecutionEnvironment env, ParameterTool params) {

        if (params.has("links")) {

            return env.readCsvFile(params.get("links"))

                    .fieldDelimiter(" ")

                    .lineDelimiter("\n")

                    .types(Long.class, Long.class);

        } else {

            System.out.println("Executing PageRank example with default links data set.");

            System.out.println("Use --links to specify file input.");

            return PageRankData.getDefaultEdgeDataSet(env);

        }

    }

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

        ParameterTool params = ParameterTool.fromArgs(args);

        final int numPages = params.getInt("numPages", PageRankData.getNumberOfPages());

        final int maxIterations = params.getInt("iterations", 10);

        // set up execution environment

        final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

        // make the parameters available to the web ui

        env.getConfig().setGlobalJobParameters(params);

        // get input data

        DataSet<Long> pagesInput = getPagesDataSet(env, params);

        DataSet<Tuple2<Long, Long>> linksInput = getLinksDataSet(env, params);

        // assign initial rank to pages

        DataSet<Tuple2<Long, Double>> pagesWithRanks = pagesInput

                .map(new RankAssigner(1.0d / numPages));

        // build adjacency list from link input

        DataSet<Tuple2<Long, Long[]>> adjacencyListInput = linksInput

                .groupBy(0)

                .reduceGroup(new BuildOutgoingEdgeList());

        // set iterative data set

        IterativeDataSet<Tuple2<Long, Double>> iteration = pagesWithRanks.iterate(maxIterations);

        DataSet<Tuple2<Long, Double>> newRanks = iteration.join(adjacencyListInput)

                .where(0).equalTo(0)

                //迭代算子

                .flatMap(new JoinVertexWithEdgesMatch())

                // collect and sum ranks

                .groupBy(0)

                .aggregate(Aggregations.SUM, 1)

                // apply dampening factor

                .map(new Dampener(PageRank.DAMPENING_FACTOR, numPages));

        //如果没有达到收敛条件，循环10次后结束

        DataSet<Tuple2<Long, Double>> finalPageRanks = iteration.closeWith(

                newRanks,

                newRanks.join(iteration).where(0).equalTo(0)

                        // termination condition

                        .filter(new EpsilonFilter()));

        // emit result

        if (params.has("output")) {

            finalPageRanks.writeAsCsv(params.get("output"), "\n", " ");

            // execute program

            env.execute("Basic Page Rank Example");

        } else {

            System.out.println("Printing result to stdout. Use --output to specify output path.");

            finalPageRanks.print();

        }

    }

    /**

     * A map function that assigns an initial rank to all pages.

     */

    public static final class RankAssigner implements MapFunction<Long, Tuple2<Long, Double>> {

        Tuple2<Long, Double> outPageWithRank;

        public RankAssigner(double rank) {

            this.outPageWithRank = new Tuple2<>(-1L, rank);

        }

        @Override

        public Tuple2<Long, Double> map(Long value) throws Exception {

            outPageWithRank.f0 = value;

            return outPageWithRank;

        }

    }

    /**

     * A reduce function that takes a sequence of edges and builds the adjacency list for the vertex where the edges

     * originate. Run as a pre-processing step.

     * 将分组后的links数据按Id放入Tuple2<Long, Long[]>

     * 与hadoop的mapreduce的reduce部分类似，groupby就是shuffle

     */

    @FunctionAnnotation.ForwardedFields("0")

    public static final class BuildOutgoingEdgeList implements GroupReduceFunction<Tuple2<Long, Long>, Tuple2<Long, Long[]>> {

        private final ArrayList<Long> neighbors = new ArrayList<Long>();

        @Override

        public void reduce(Iterable<Tuple2<Long, Long>> values, Collector<Tuple2<Long, Long[]>> out) {

            neighbors.clear();

            Long id = 0L;

            for (Tuple2<Long, Long> n : values) {

                id = n.f0;

                neighbors.add(n.f1);

                System.out.println("id: " + id + " ,neighbors: " + n.f1);

            }

            out.collect(new Tuple2<Long, Long[]>(id, neighbors.toArray(new Long[neighbors.size()])));

        }

    }

    /**

     * Join function that distributes a fraction of a vertex's rank to all neighbors.

     * 按照页面id以及其连接页面的数量，重新计算相邻点的rank，迭代10次

     * rankToDistribute就是计算了顶点（id）指向边（neighbors）的贡献值，neighbors最终的rank值需要合计后才能算出

     */

    public static final class JoinVertexWithEdgesMatch implements FlatMapFunction<Tuple2<Tuple2<Long, Double>, Tuple2<Long, Long[]>>, Tuple2<Long, Double>> {

        @Override

        public void flatMap(Tuple2<Tuple2<Long, Double>, Tuple2<Long, Long[]>> value, Collector<Tuple2<Long, Double>> out) {

            System.out.println("before:" + value);

            Long[] neighbors = value.f1.f1;

            double rank = value.f0.f1;

            double rankToDistribute = rank / ((double) neighbors.length);

            for (Long neighbor : neighbors) {

//                System.out.println("neighbor:" + neighbor + ",rankToDistribute:" + rankToDistribute);

                out.collect(new Tuple2<Long, Double>(neighbor, rankToDistribute));

            }

        }

    }

    /**

     * The function that applies the page rank dampening formula.

     * 阻尼系数公式:PR(A)=(1-d)/N + d(PR(T1)/C(T1)+ ... +PR(Tn)/C(Tn))

     * PR(A) 是页面A的PR值

     * PR(Ti)是页面Ti的PR值，在这里，页面Ti是指向A的所有页面中的某个页面

     * C(Ti)是页面Ti的出度，也就是Ti指向其他页面的边的个数

     * d 为阻尼系数，其意义是，在任意时刻，用户到达某页面后并继续向后浏览的概率，

     * 该数值是根据上网者使用浏览器书签的平均频率估算而得，通常d=0.85

     */

    @FunctionAnnotation.ForwardedFields("0")

    public static final class Dampener implements MapFunction<Tuple2<Long, Double>, Tuple2<Long, Double>> {

        private final double dampening;

        private final double randomJump;

        public Dampener(double dampening, double numVertices) {

            this.dampening = dampening;

            this.randomJump = (1 - dampening) / numVertices;

        }

        @Override

        public Tuple2<Long, Double> map(Tuple2<Long, Double> value) {

            value.f1 = (value.f1 * dampening) + randomJump;

            return value;

        }

    }

    /**

     * Filter that filters vertices where the rank difference is below a threshold.

     * 如果迭代之间的参数之和低于此EPSILON，我们将会收敛

     * 每次迭代后都会调用Filter判断是否要退出迭代

     */

    public static final class EpsilonFilter implements FilterFunction<Tuple2<Tuple2<Long, Double>, Tuple2<Long, Double>>> {

        @Override

        public boolean filter(Tuple2<Tuple2<Long, Double>, Tuple2<Long, Double>> value) {

            System.out.println("value:"+value+",math:"+Math.abs(value.f0.f1 - value.f1.f1));

            return Math.abs(value.f0.f1 - value.f1.f1) > EPSILON;

        }

    }

}