PageRank在Hadoop和spark下的实现以及对比

关于PageRank的地位，不必多说。

主要思想：对于每个网页，用户都有可能点击网页上的某个链接，例如

A:B,C,D
B:A,D
C:A
D:B,C

由这个我们可以得到网页的转移矩阵

A B C D

A 0 1/2 1 0

B 1/3 0 0 0

C 1/3 1/2 0 0

D 1/3 0 0 1/2

Aij表示网页j到网页i的转移概率。假设起始状态每个用户对ABCD四个网站的点击概率相同都是0.25，那么各个网站第一次被访问的概率为（0.25，0.25，0.25，0.25），第二次访问考虑到在页面跳转，利用转移矩阵对于网站A的概率为（0，1/2,1,0）*(0.25,0.25,0.25,0.25)T,一次类推，经过若干次迭代会收敛到某个值。但是考虑到有些链接是单链即没有别的链接只想他，他也不指向别的链接，以及有些链接是自己指向自己，那么上述的方式将无法收敛。所以后面加了一个阻尼系数一般取0.85，至于为什么是这样，挺复杂的证明。

最后的公式为alaph=factor*matrix*(alaph)T+(1-facotr)/n*

详细的介绍可以参考：http://blog.jobbole.com/71431/

接下来便是对比Hadoop和spark了。这里只是单纯的讨论两个环境下编程的效率，不讨论性能。

Hadoop：

输入的文件：

A 0.25:B,C,D
B 0.25:A,D
C 0.25:A
D 0.25:B,C

这里得先说一句，之所以加了0.25是因为初始的概率为1/n，而n为网站数，这里统计网站数又得需要一个MapReduce来实现，所以作罢，权当n是手工输入的。

由于每次迭代后的结果只能放在文件中，所以这里花了很多时间在规范如何输出，以及map和reduce之间如何传值的问题。

在map中，我们要做的是从输入文件中获取alaph和每个网站的转移概率。例如

A 0.25:B,C,D

B的转移概率为1/3而且是从A转向B的，所以输出的是<"B","link:A 0.333">link表示这是个转移概率，A表示是从A出发的

alaph的表示：<"B","alaph: A 0.25">这里的A表示这个alaph值对应这A。

由于我们这里迭代后的输入文件都是从输出文件中获取，所以我们需要将输出文件搞的和一开始输入文件一样，所以在map阶段需要输出<"A","content:B,C,D">方便reduce输出和输入文件一样格式的输出。

在reduce阶段，此时对于键值B而言，会收到如下

<"B","link:A 0.333">

<"B","link:D 0.5">

<"B","alaph: A 0.25">

<"B","alaph: D 0.25">

<"B","content:A,D">

我们根据不同的单词，将value整合。这的alaph=0.333*0.25+0.5*0.25,接着再加上阻尼系数等，得到最后的alaph值。然后利用content对应的value，最后输出<"B：0.375","A,D">

这样迭代若干次。

附上代码：

 package org.apache.hadoop.PageRank;

 import java.util.ArrayList;

 import org.apache.hadoop.conf.Configuration;

 import org.apache.hadoop.fs.FileSystem;

 import org.apache.hadoop.fs.Path;

 import org.apache.hadoop.io.Text;

 import org.apache.hadoop.mapreduce.Job;

 import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;

 import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

 public class PageRank {

     public static void run(){

     }

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

         double factor=0;

         if(args.length>1){

             factor=Double.parseDouble(args[0]);

         }else{

             factor=0.85;

         }

         String input="hdfs://10.107.8.110:9000/PageRank_input";

         String output="hdfs://10.107.8.110:9000/PageRank/output";

         ArrayList<String> pathList=new ArrayList<String>();

         for(int i=0;i<20;i++){

             Configuration conf = new Configuration();

             conf.set("num","4");

             conf.set("factor",String.valueOf(factor));

             Job job = Job.getInstance(conf, "PageRank");

             job.setJarByClass(org.apache.hadoop.PageRank.PageRank.class);

             job.setMapperClass(MyMapper.class);

             job.setReducerClass(MyReducer.class);

             job.setOutputKeyClass(Text.class);

             job.setOutputValueClass(Text.class);

             FileInputFormat.setInputPaths(job, new Path(input));

             FileOutputFormat.setOutputPath(job, new Path(output));

             input=output;

             pathList.add(output);

             output=output+1;

             System.out.println("the "+i+"th iterator is finished");

             job.waitForCompletion(true);

         }

         for(int i=0;i<pathList.size()-1;i++){

             Configuration conf=new Configuration();

             Path path=new Path(pathList.get(i));

             FileSystem fs=path.getFileSystem(conf);

             fs.delete(path,true);

         }

     }

 }

 package org.apache.hadoop.PageRank;

 import java.io.IOException;

 import java.util.HashMap;

 import java.util.Map;

 import org.apache.hadoop.io.LongWritable;

 import org.apache.hadoop.io.Text;

 import org.apache.hadoop.mapreduce.Mapper;

 public class MyMapper extends Mapper<LongWritable, Text, Text, Text> {

     public void map(LongWritable ikey, Text ivalue, Context context)

             throws IOException, InterruptedException {

         String[] line=ivalue.toString().split(":");

         String content=line[1];

         int num=content.split(",").length;

         String word=line[0].split("    ")[0];

         String alaph=line[0].split("    ")[1];

         context.write(new Text(word),new Text("content:"+content));

         for(String w:content.split(",")){

             context.write(new Text(w),new Text("link:"+word+" "+String.valueOf(1.0/num)));

             context.write(new Text(w),new Text("alaph:"+word+" "+alaph));

         }

     }

 }

 package org.apache.hadoop.PageRank;

 import java.io.IOException;

 import java.util.HashMap;

 import java.util.Map;

 import org.apache.hadoop.conf.Configuration;

 import org.apache.hadoop.io.Text;

 import org.apache.hadoop.mapreduce.Reducer;

 public class MyReducer extends Reducer<Text, Text, Text, Text> {

     public void reduce(Text _key, Iterable<Text> values, Context context)

             throws IOException, InterruptedException {

         // process values

         Configuration conf=context.getConfiguration();

         double factor=Double.parseDouble(conf.get("factor"));

         int num=Integer.parseInt(conf.get("num"));

         Map<String,Double> alaph=new HashMap<String,Double>();

         Map<String,Double> link=new HashMap<String,Double>();

         String content="";

         for (Text val : values) {

             String[] line=val.toString().split(":");

             if(line[0].compareTo("content")==0){

                 content=line[1];

             }else {

                 String[] s=line[1].split(" ");

                 double d=Double.parseDouble(s[1]);

                 if(line[0].compareTo("alaph")==0){

                     alaph.put(s[0],d);

                 }else if(line[0].compareTo("link")==0){

                     link.put(s[0],d);

                 }

             }

         }

         double sum=0;

         for(Map.Entry<String,Double> entry:alaph.entrySet()){

             sum+=link.get(entry.getKey())*entry.getValue();

         }

         System.out.println("    ");

         System.out.println("sum is "+sum);

         System.out.println("    ");

         double result=factor*sum+(1-factor)/num;

         context.write(_key,new Text(String.valueOf(result)+":"+content));

     }

 }

我们可以看出，其实在MapReduce中我们将大把的精力花在了map的输出上，而之所以这样是因为我们不能直接利用他的结果，并且为了能迭代，我们又只能格式化输出，如果数据很多的，那么在map阶段将有很多的资源需要传递。总而言之，Hadoop让我们将大部分精力花在不该花的地方。

接下来看spark 。我这里用的是python，在pyspark下运行。输入文件：

A:B,C,D
B:A,D
C:A
D:B,C

先看代码

def f(x):

    links=x[1][0]

    rank=x[1][1]

    n=len(links.split(","))

    result=[]

    for s in links.split(","):

        result.append((s,rank*1.0/n))

    return result

file="hdfs://10.107.8.110:9000/spark_test/pagerank.txt"

data=sc.textFile(file)

link=data.map(lambda x:(x.split(":")[0],x.split(":")[1]))

n=data.count()

rank=link.mapValues(lambda x:1.0/n)

for i in range(10):

    rank=link.join(rank).flatMap(f).reduceByKey(lambda x,y:x+y).mapValues(lambda x:0.15/n+0.85*x)

直接分析，data=sc.textFile(file)从hdfs中获取text文件。

通过data.collect()可以发现内容为