《hadoop进阶》PeopleRank从社交关系中挖掘价值用户

package pagerankjisuan;

import java.text.DecimalFormat;
import java.util.HashMap;
import java.util.Map;

/*
 * 调度函数
 */
public class prjob {

	public static final String  HDFS = "hdfs://127.0.0.1:9000";
	
	public static void main(String[] args) {
		Map  <String,String> path= new HashMap<String,String>();	
		
		path.put("page","/home/thinkgamer/MyCode/hadoop/MyItems/pagerankjisuan/people.csv");
		path.put("pr","/home/thinkgamer/MyCode/hadoop/MyItems/pagerankjisuan/peoplerank.txt");
		
		path.put("input",HDFS + "/mr/blog_analysic_system/people");          // HDFS的目录
        path.put("input_pr",HDFS + "/mr/blog_analysic_system/pr");       // pr存储目录
        path.put("tmp1",HDFS + "/mr/blog_analysic_system/tmp1");             // 临时目录,存放邻接矩阵
        path.put("tmp2",HDFS + "/mr/blog_analysic_system/tmp2");           // 临时目录,计算到得PR,覆盖input_pr

        path.put("result",HDFS + "/mr/blog_analysic_system/result");                   // 计算结果的PR
        
        path.put("sort",HDFS +  "/mr/blog_analysic_system/sort");  //最终排序输出的结果
        
        try {
        	   dataEtl.main();
            prMatrix.main(path);
            int iter = 3;           // 迭代次数
            for (int i = 0; i < iter; i++) {
                prJisuan.main(path);
            }
           prNormal.main(path);
           prSort.main(path);

        	} catch (Exception e) {
        		e.printStackTrace();
        	}
        	System.exit(0);
    	}
	
	 	public static String scaleFloat(float f) {// 保留6位小数
	        DecimalFormat df = new DecimalFormat("##0.000000");
	        return df.format(f);
	    }
}

prSort.java

package pagerankjisuan;

import java.io.IOException;
import java.net.URISyntaxException;
import java.util.Map;

import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.FloatWritable;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.IntWritable.Comparator;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.WritableComparable;
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 prSort {  
	/**
     * @param args
     * @throws IOException 
     * @throws IllegalArgumentException 
     * @throws InterruptedException 
     * @throws ClassNotFoundException 
     */
    public static class myComparator extends Comparator {
        @SuppressWarnings("rawtypes")
        public int compare( WritableComparable a,WritableComparable b){
            return -super.compare(a,b);
        }
        public int compare(byte[] b1,int s1,int l1,byte[] b2,int s2,int l2) {
            return -super.compare(b1,s1,l1,b2,s2,l2);
        }
    }
    
    public static class sortMap extends Mapper<Object,Text,FloatWritable,IntWritable>{
        public void map(Object key,Text value,Context context) throws NumberFormatException,IOException,InterruptedException{
            String[] split = value.toString().split("t");
            context.write(new FloatWritable(Float.parseFloat(split[1])),new IntWritable(Integer.parseInt(split[0])) );
        }
    }
    public static class Reduce extends Reducer<FloatWritable,IntWritable,FloatWritable>{
        public void reduce(FloatWritable key,Iterable<IntWritable>values,Context context) throws IOException,InterruptedException{
            for (IntWritable text : values) {
                context.write( text,key);
            }
        }
    }
  
	public static void main(Map<String,String> path) throws IOException,URISyntaxException,ClassNotFoundException,InterruptedException {
        // TODO Auto-generated method stub  
  
		String input = path.get("result");
		String output = path.get("sort");
		hdfsGYT hdfs = new hdfsGYT();
		hdfs.rmr(output);
		
        Job job = new Job();  
        job.setJarByClass(prSort.class);  
        // 1  
        FileInputFormat.setInputPaths(job,new Path(input) );  
        // 2  
        job.setMapperClass(sortMap.class);  
        job.setMapOutputKeyClass(FloatWritable.class);  
        job.setMapOutputValueClass(IntWritable.class);  
        // 3  
        // 4  自定义排序
        job.setSortComparatorClass( myComparator.class); 
        // 5  
        job.setNumReduceTasks(1);  
        // 6  
        job.setReducerClass(Reduce.class);  
        job.setOutputKeyClass(IntWritable.class);  
        job.setOutputValueClass(FloatWritable.class);  
        // 7  
        FileOutputFormat.setOutputPath(job,new Path(output));  
        // 8  
        System.exit(job.waitForCompletion(true)? 0 :1 );  
    }  
}

最终排序输出的结果为：

（编辑：PHP编程网 - 湛江站长网）

【声明】本站内容均来自网络，其相关言论仅代表作者个人观点，不代表本站立场。若无意侵犯到您的权利，请及时与联系站长删除相关内容!