PageRank

另一篇介绍的很好的博客：http://blog.jobbole.com/71431/

一、什么是PageRank

PageRank 是对搜索引擎的搜索网页进行排序的算法。

过去的排序算法是比如使用网页名字，关键词出现的次数，人工等方法，但是这种方法一方面搜索结果不准确，另一方面搜索结果容易被人为因素影响。

所以，PageRank应运而生。

PageRank算法计算每一个网页的PageRank值，然后根据这个值的大小对网页的重要性进行排序。它的思想是模拟一个悠闲的上网者，上网者首先随机选择一个网页打开，然后在这个网页上呆了几分钟后，跳转到该网页

所指向的链接，这样无所事事、漫无目的地在网页上跳来跳去，PageRank就是估计这个悠闲的上网者分布在各个网页上的概率。

PageRank背后的两个基本假设：

• 数量假设：更重要的网页更可能被更多的网页链接到。
• 质量假设：有更高的PageRank的网页将会传递更高的权重。(类似大佬转发的网页就比我转发的网页更重要)

二、PageRank模型

互联网中网页和网页之间的链接关系组成一个有向图，其中网页是节点，网页之间的链接为有相边。如上图所示的模型就表示四个网页，A指向B的箭头表示A中存在指向B的链接。

可以使用转移矩阵来表示这样一个有向图。上图所对应的模型即可用转移矩阵表示为：

我们看到，A网页发出三个链接，那么到任何一个网页的概率都均等，所以我们上边的矩阵A这一列对用的BCD三个网页概率都为1/3。

有了上面的转移矩阵，我们从两个角度来思考：

一方面，一个网页X如果有许多网页指向他，那么转移矩阵的X行就会有许多的非零元素，转移之后这个网页就会获得更大的PageRank，这刚好和我们上文提到的数量假设相对应。

另一方面，假设X网页具有很高的重要性，那么X所转移到的那些网页将会由于X的高重要性而获得相对更高的重要性，这刚好和我们上文提到的质量假设相对应。

那么，怎样表示一个网页的重要性呢？

我们使用PageRank Matrix来表示。上述有向图对应一个4 * 1的PageRank矩阵。

为什么初始化为均分初始化呢？我们可以这样思考，一个无聊的上网这开始想要来打开一个网页，那么他打开这一堆网页中的任何一个在最开始都是风概率的。(当然不考虑个人信息，那是个性化搜索的范畴)

好了，我们现在有了两个矩阵，转移矩阵和初始的PageRank矩阵PR0, 我们接下来就要模拟网页间的跳转。

如图，一次跳转之后，网页A的PageRank就应该由可以跳到A的网页和他们各自的PageRank相乘之后的和来表示。上图的矩阵乘法就可以表示一次跳转之后的各个网页的PageRank值。

同理，PR2 = Transition Matrix * PR1。 PRn =  Transition Matrix * PR(n - 1)。可以证明，最终的PR是收敛的。（马尔科夫过程，图是强连通的）

三、Dead Ends 终止点问题

有些网页不存在指向其他网页的链接，那么多次迭代之后，导致所有网页的PageRank都变为0.

上图网页c没有指向外部的链接，最终将会导致各个网页的PageRank都变为0.

四、Spider traps 陷阱问题

陷阱问题是有些网页只存在指向自己的链接，那么多次迭代之后，这将导致这个网页的PageRank为1,而其他网页的PageRank为0.

 

五、解决终止点问题和陷阱问题

我们现在回到开始是所说的那个无聊的上网者，假如人类遇到这个问题的话，采取的解决办法就是关闭当前网页，重新打开一个网页。所以为了避免上述两个问题，我们呢对上边的转移公式做一个小小的修正。（有概率随机打开一个网页，这时候打开所有网页的概率均等）

回到之前的例子：

六、Map-Reduce计算PageRank

6.1输入数据

上边分析的时候数据似乎是存储在转移矩阵中，但实际不是这样的，这样一方面太浪费空间了，另一方面，插入和删除数据都很麻烦。

实际的数据输入格式如下：

我们先对数据进行预处理，类似于邻接表的存储方式，存储网页和该网页包含的链接所指向的网页。

我们使用Relations.txt来生成转移矩阵，然后在进行转移矩阵和PageRank矩阵的相乘操作。

6.2矩阵相乘

接下来就设计到Map-Reduce处理矩阵的乘法操作，我们不能直接来进行两个矩阵的乘法操作。因为如果这样做的话，我们需要转移矩阵的一行数据都来齐之后才可进行，这就需要我们 in-memory 存储整个矩阵，很容易造成out of mempry 而且速度还慢。

所以我们的处理方法是用Traverse MAtrix的每一列和PR相成，最后把所有元素相加即可。这样我们就不需要in-memery那么多东西。

其实对于一个网页A，可以思考为，你A B C D分别可以为我贡献多少，然后把ABCD的贡献全都加起来，就是A的PageRank。矩阵直接相乘和我们的相乘相加的方法的其别就是，直接相乘是计算A的PageRank 的时候，把所以对A有贡献的网页都拿来，计算出贡献加起来作为A的PageRank;相乘相加是首先计算出A可以为ABCD贡献多少，然后B可以为ABCD贡献多少，然后C..D..。最后再把这些值加起来，分别作为ABCD的PageRank。

我们的流程图如下：

容易知道，我们使用两个Map-Reduce来实现。

MR1 --- Mapper1 从Relations.txt中读取网页并计算跳转到的网页及概率

  

MR1 --- Mapper2 从PR.txt中读取网页并均分概率

MR1 --- Reducer 两个Mapper过来的key就代表是慨率网页key转移，输出key应该为转移到的网页，value为网页key对这个网页做了多少贡献。

MR2 --- Mapper 很简单，就是从前一个Map-Reduce生成的文件中读取数据

MR2 --- Reducer Mapper按照网页作key之后分类的sum,得到各个网页的PageRank

 七、主要代码

Driver

public class Driver {

    public static void main(String[] args) throws Exception {
        UnitMultiplication multiplication = new UnitMultiplication();
        UnitSum sum = new UnitSum();

        //args0: dir of transition.txt
        //args1: dir of PageRank.txt
        //args2: dir of unitMultiplication result
        //args3: times of convergence
        String transitionMatrix = args[0];
        String prMatrix = args[1];
        String unitState = args[2];
        int count = Integer.parseInt(args[3]);
        for(int i=0;  i<count;  i++) {
            String[] args1 = {transitionMatrix, prMatrix+i, unitState+i};
            multiplication.main(args1);
            String[] args2 = {unitState + i, prMatrix+(i+1)};
            sum.main(args2);
        }
    }
}

UnitMultiplication

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.InputFormat;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.chain.ChainMapper;
import org.apache.hadoop.mapreduce.lib.input.MultipleInputs;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

public class UnitMultiplication {

    public static class TransitionMapper extends Mapper<Object, Text, Text, Text> {

        @Override
        public void map(Object key, Text value, Context context) throws IOException, InterruptedException {
            String line = value.toString().trim();
            String[] fromTo = line.split("	");

            if(fromTo.length == 1 || fromTo[1].trim().equals("")) {
                return;
            }
            String from = fromTo[0];
            String[] tos = fromTo[1].split(",");
            for (String to: tos) {
                context.write(new Text(from), new Text(to + "=" + (double)1/tos.length));
            }
        }
    }

    public static class PRMapper extends Mapper<Object, Text, Text, Text> {

        @Override
        public void map(Object key, Text value, Context context) throws IOException, InterruptedException {
            String[] pr = value.toString().trim().split("	");
            context.write(new Text(pr[0]), new Text(pr[1]));
        }
    }

    public static class MultiplicationReducer extends Reducer<Text, Text, Text, Text> {


        @Override
        public void reduce(Text key, Iterable<Text> values, Context context)
                throws IOException, InterruptedException {
            List<String> transitionUnit = new ArrayList<String>();
            double prUnit = 0;
            for (Text value: values) {
                if(value.toString().contains("=")) {
                    transitionUnit.add(value.toString());
                }
                else {
                    prUnit = Double.parseDouble(value.toString());
                }
            }
            for (String unit: transitionUnit) {
                String outputKey = unit.split("=")[0];
                double relation = Double.parseDouble(unit.split("=")[1]);
                String outputValue = String.valueOf(relation * prUnit);
                context.write(new Text(outputKey), new Text(outputValue));
            }
        }
    }

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

        Configuration conf = new Configuration();
        Job job = Job.getInstance(conf);
        job.setJarByClass(UnitMultiplication.class);

        ChainMapper.addMapper(job, TransitionMapper.class, Object.class, Text.class, Text.class, Text.class, conf);
        ChainMapper.addMapper(job, PRMapper.class, Object.class, Text.class, Text.class, Text.class, conf);

        job.setReducerClass(MultiplicationReducer.class);

        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(Text.class);

        MultipleInputs.addInputPath(job, new Path(args[0]), TextInputFormat.class, TransitionMapper.class);
        MultipleInputs.addInputPath(job, new Path(args[1]), TextInputFormat.class, PRMapper.class);

        FileOutputFormat.setOutputPath(job, new Path(args[2]));
        job.waitForCompletion(true);
    }

}

UnitSum

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.DoubleWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.IOException;
import java.text.DecimalFormat;

public class UnitSum {
    public static class PassMapper extends Mapper<Object, Text, Text, DoubleWritable> {

        @Override
        public void map(Object key, Text value, Context context) throws IOException, InterruptedException {
           String[] pageSubrank = value.toString().split("	");
            double subRank = Double.parseDouble(pageSubrank[1]);
            context.write(new Text(pageSubrank[0]), new DoubleWritable(subRank));
        }
    }

    public static class SumReducer extends Reducer<Text, DoubleWritable, Text, DoubleWritable> {


        @Override
        public void reduce(Text key, Iterable<DoubleWritable> values, Context context)
                throws IOException, InterruptedException {

            double sum = 0;
            for (DoubleWritable value: values) {
                sum += value.get();
            }
            DecimalFormat df = new DecimalFormat("#.0000");
            sum = Double.valueOf(df.format(sum));
            context.write(key, new DoubleWritable(sum));
        }
    }

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

        Configuration conf = new Configuration();
        Job job = Job.getInstance(conf);
        job.setJarByClass(UnitSum.class);
        job.setMapperClass(PassMapper.class);
        job.setReducerClass(SumReducer.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(DoubleWritable.class);
        FileInputFormat.addInputPath(job, new Path(args[0]));
        FileOutputFormat.setOutputPath(job, new Path(args[1]));
        job.waitForCompletion(true);
    }
}

 考虑到dead ends 和 spider traps的话，只是前边的矩阵cell相乘的时候乘以beta，后边再用一个mapper来加上beta倍的pagerank即可。