mr实现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 = ‘\t‘;

	  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\t"+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();
			}
		}
	}
}