mr实现pagerank

PageRank计算
什么是pagerank
PageRank是Google专有的算法，用于衡量特定网页相对于搜索引擎索引中的其他网页而言的重要程度。
是Google创始人拉里·佩奇和谢尔盖·布林于1997年创造的
PageRank实现了将链接价值概念作为排名因素。

PageRank计算
算法原理（1）
入链 ====投票
PageRank让链接来“投票“，到一个页面的超链接相当于对该页投一票。
入链数量
如果一个页面节点接收到的其他网页指向的入链数量越多，那么这个页面越重要。
入链质量
指向页面A的入链质量不同，质量高的页面会通过链接向其他页面传递更多的权重。所以越是质量高的页面指向页面A，则页面A越重要。
PageRank计算
网络上各个页面的链接图

算法原理（2）
初始值
每个页面设置相同的PR值
Google的pagerank算法给每个页面的PR初始值为1。
迭代递归计算（收敛）
Google不断的重复计算每个页面的PageRank。那么经过不断的重复计算，这些页面的PR值会趋向于稳定，也就是收敛的状态。
在具体企业应用中怎么样确定收敛标准？
1、每个页面的PR值和上一次计算的PR相等
2、设定一个差值指标（0.0001）。当所有页面和上一次计算的PR差值平均小于该标准时，则收敛。
3、设定一个百分比（99%），当99%的页面和上一次计算的PR相等

算法原理（3）
修正PageRank计算公式
由于存在一些出链为0，也就是那些不链接任何其他网页的网， 也称为孤立网页，使得很多网页能被访问到。因此需要对 PageRank公式进行修正，即在简单公式的基础上增加了阻尼系数（damping factor）q， q一般取值q=0.85。

完整PageRank计算公式

 

package com.laoxiao.mr.pagerank;

import java.io.IOException;
import java.util.Arrays;

import org.apache.commons.lang.StringUtils;

public class Node {

	 private double pageRank=1.0;
	  private String[] adjacentNodeNames;

	  public static final char fieldSeparator = '	';

	  public double getPageRank() {
	    return pageRank;
	  }

	  public Node setPageRank(double pageRank) {
	    this.pageRank = pageRank;
	    return this;
	  }

	  public String[] getAdjacentNodeNames() {
	    return adjacentNodeNames;
	  }

	  public Node setAdjacentNodeNames(String[] adjacentNodeNames) {
	    this.adjacentNodeNames = adjacentNodeNames;
	    return this;
	  }

	  public boolean containsAdjacentNodes() {
	    return adjacentNodeNames != null && adjacentNodeNames.length>0;
	  }

	  @Override
	  public String toString() {
	    StringBuilder sb = new StringBuilder();
	    sb.append(pageRank);

	    if (getAdjacentNodeNames() != null) {
	      sb.append(fieldSeparator).append(StringUtils.join(getAdjacentNodeNames(), fieldSeparator));
	    }
	    return sb.toString();
	  }

	  //value =1.0	B	D
	  public static Node fromMR(String value) throws IOException {
	    String[] parts = StringUtils.splitPreserveAllTokens(value, fieldSeparator);
	    if (parts.length < 1) {
	      throw new IOException(
	          "Expected 1 or more parts but received " + parts.length);
	    }
	    Node node = new Node()
	        .setPageRank(Double.valueOf(parts[0]));
	    if (parts.length > 1) {
	      node.setAdjacentNodeNames(Arrays.copyOfRange(parts, 1,
	          parts.length));
	    }
	    return node;
	  }
}



package com.laoxiao.mr.pagerank;


public enum Mycounter {
	my
}


package com.laoxiao.mr.pagerank;

import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

public class pagerankMapper extends Mapper<Text, Text, Text, Text>{

	protected void map(Text key, Text value, Context context) 
			throws java.io.IOException ,InterruptedException {
		
		int runCount= context.getConfiguration().getInt("runCount", 1);
		String page=key.toString();
		Node node=null;
		if(runCount==1){
			node=Node.fromMR("1.0	"+value.toString());
		}else{
			node=Node.fromMR(value.toString());
		}
		
		Double pagerank=node.getPageRank()/node.getAdjacentNodeNames().length;
		context.write(new Text(page), new Text(node.toString()));//A:1.0	B	D
		if(node.containsAdjacentNodes()){
			double outValue =node.getPageRank()/node.getAdjacentNodeNames().length;
			for (int i = 0; i < node.getAdjacentNodeNames().length; i++) {
				String  outPage = node.getAdjacentNodeNames()[i];
				context.write(new Text(outPage), new Text(outValue+""));//B:0.5  D:0.5
			}
		}
	};
}


package com.laoxiao.mr.pagerank;

import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

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

	protected void reduce(Text arg0, java.lang.Iterable<Text> arg1, Context arg2) 
			throws java.io.IOException ,InterruptedException {
		double sum =0.0;
		Node sourceNode =null;
		for(Text i:arg1){
			Node node =Node.fromMR(i.toString());
			if(node.containsAdjacentNodes()){
				sourceNode =node;
			}else{
				sum=sum+node.getPageRank();
			}
		}
		
		double newPR=(0.15/4)+(0.85*sum);
		System.out.println("*********** new pageRank value is "+newPR);
		//把新的pr值和计算之前的pr比较
		double d= newPR -sourceNode.getPageRank();
		
		int j=(int)( d*1000.0);
		j=Math.abs(j);
		//累加器的使用
		arg2.getCounter(Mycounter.my).increment(j);
		
		
		sourceNode.setPageRank(newPR);
		arg2.write(arg0, new Text(sourceNode.toString()));
		
	};
}


package com.laoxiao.mr.pagerank;

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.input.KeyValueTextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

public class RunJob {

	public static void main(String[] args) {
		int i=0;
		Configuration config=new Configuration();
		config.set("fs.defaultFS", "hdfs://node1:8020");
		config.set("yarn.resourcemanager.hostname", "node1");
		
		double d =0.001;
		while(true){
			i++;
			try {
				config.setInt("runCount", i);//传值到mr中
				FileSystem fs =FileSystem.get(config);
				Job job =Job.getInstance(config);
				job.setJarByClass(RunJob.class);
				job.setJobName("pr"+i);
				job.setMapperClass(pagerankMapper.class);
				job.setReducerClass(pagerankReducer.class);
				job.setMapOutputKeyClass(Text.class);
				job.setMapOutputValueClass(Text.class);
				job.setInputFormatClass(KeyValueTextInputFormat.class);
				Path inputPath =new Path("/root/input/data/pagerank.txt");
				if(i>1){
					inputPath =new Path("/usr/output/pr"+(i-1));
				}
				FileInputFormat.addInputPath(job, inputPath);
				
				Path outpath =new Path("/usr/output/pr"+i);
				if(fs.exists(outpath)){
					fs.delete(outpath, true);
				}
				FileOutputFormat.setOutputPath(job, outpath);
				
				boolean f= job.waitForCompletion(true);
				if(f){
					System.out.println("success.");
					//累加器的使用
					long sum= job.getCounters().findCounter(Mycounter.my).getValue();
					System.out.println(sum);
					double avgd= sum/4000.0;
					if(avgd<d){
						break;
					}
				}
			} catch (Exception e) {
				e.printStackTrace();
			}
		}
	}
}