【大创_社区划分】——PageRank算法MapReduce实现

PageRank算法的分析和Python实现参考：http://blog.csdn.net/gamer_gyt/article/details/47443877

举例来讲：

假设每个网页都有一个自己的默认PR值，相当于人为添加给它是一种属性，用来标识网页的等级或者重要性，从而依据此标识达到排名目的。假设有ID号是1的一个网页，PR值是10，假如它产生了到ID=3，ID=6，ID=8 ，ID=9这4个网页的链接。那么可以理解为ID=1的网页向ID=3，6，8，9的4个网页各贡献了2.5的PR值。如果想求任意一个网页假设其ID=3的PR值，需要得到所有的其他网页对ID=3这个网页的贡献的总和，再按照函数“所求PR”=“总和”*0.85+0.15得到。经过循环多次上述过程求得的网页PR值，可以作为网页排名的标识。

1：准备数据 

理论数据是通过网页爬虫得到了某个特定封闭系统的所有网页的信息，为了测试程序，可以自己模拟生成自己定义特定格式的数据。下面是我用来测试的数据，存储方式如图

（注：对于自定义模拟数据，在PR初始值的选取时，所有的网页是“平等”的，不会说自己写的网页和Google的热门网页有多少差别，但是按照某种法则经过一定运算后PR是不一样的，比如很多其他的网页可能会链接到google，它的PR自然会比你的高。所以初始值的选取按照这种逻辑来讲符合现实些，即所有网页默认PR值相等。但是即使初始值定的不一样，整个系统的PR总和可能会变化，最后的每个网页PR也会发生变化，但是这种量之间的变化，不会影响到网页自身的通过比较大小方式上的逻辑排名。

2：MapReduce过程

map接受的数据格式默认是<偏移量，文本行>这样的<key,value>对，形如<0,1    5  2 3 4 5><20,2    10 3 5 8 9>.

目标 ：将默认数据格式，转换成自定义格式<key,value>对。

已知 ：hadoop框架在Map阶段的时候会自动实现sort过程，就是将相同的key的所有value保存到list，形如<key,list(1,1,1,2)>这种形式，例如上述对ID=2的网页有ID=1，6，7，8.这4个网页贡献（1.25，1，5/3，5），那么如果你采用的key是网页ID，那么hadoop框架会以此种形式<2,list(1.25，1，5/3，5)>输出，传递给reduce。

Reduce阶段：

分析：这个阶段操作就相对简单很多，读取map的输出<key,value>，并解析出来。

具体操作：把values中的数字相加即为对应id的PageRank值。

结果如下图：

代码如下

package pageRank; import java.io.IOException; import java.util.StringTokenizer; import org.apache.hadoop.fs.Path; import org.apache.hadoop.io.*; 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; public class pageRank { public static class Map extends Mapper<Object,Text,IntWritable,FloatWritable>{ private final IntWritable word = new IntWritable(); private String pr; public void map(Object key,Text value,Context context) throws IOException, InterruptedException{ StringTokenizer itr = new StringTokenizer(value.toString()); if(itr.hasMoreTokens()) {String id = itr.nextToken();} else return; pr = itr.nextToken(); //网页的pr值 int count = itr.countTokens(); //链接ID的数目 float average_pr = Float.parseFloat(pr)/count; while(itr.hasMoreTokens()){ word.set(Integer.parseInt(itr.nextToken())); context.write(word, new FloatWritable(average_pr)); } } } public static class Reduce extends Reducer<IntWritable,FloatWritable,IntWritable,FloatWritable>{ float sum; public void reduce(IntWritable key,Iterable<FloatWritable>values,Context context) throws IOException, InterruptedException{ for(FloatWritable val:values){ sum += val.get(); } context.write(key,new FloatWritable(sum)); } } public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException { // TODO Auto-generated method stub Job job = new Job(); job.setJarByClass(pageRank.class); job.setNumReduceTasks(1); job.setMapperClass(Map.class); job.setReducerClass(Reduce.class); job.setOutputKeyClass(IntWritable.class); job.setOutputValueClass(FloatWritable.class); FileInputFormat.addInputPath(job, new Path("/thinkgamer/input")); FileOutputFormat.setOutputPath(job, new Path("/thinkgamer/output")); System.exit(job.waitForCompletion(true)? 0 : 1); } }