Hadoop-PageRank算法的MapReduce实现

在本次实验中，我是参考了PageRank算法的MapReduce实现

输入文本格式：网页+\t+该网页链接到的网页的集合（相互之间用英文逗号分开）

实验要求&输出文本格式：

PageRank算法思想：

如果网页T存在一个指向网页A的连接，则表明T的所有者认为A比较重要，从而把T的一部分重要性得分赋予A。这个重要性得分值为：PR（T）/L(T)

其中PR（T）为T的PageRank值，L(T)为T的出链数。

则A的PageRank值为一系列类似于T的页面重要性得分值的累加。

即一个页面的得票数由所有链向它的页面的重要性来决定，到一个页面的超链接相当于对该页投一票。一个页面的PageRank是由所有链向它的页面（链入页面）的重要性经过递归算法得到的。一个有较多链入的页面会有较高的等级，相反如果一个页面没有任何链入页面，那么它没有等级。

解决思路：

可以得到公式

PR(A)即A的PageRank值；d为阻尼因子，实验中给出为0.85；L(B)即B网站所有的出链数量（即B网站内的所有链接的数量）。

所以公式的意义是：A的PageRank值=（1-d）+d*（链接到A的所有网站的PR值/该网站的所有出链数量之和）。这里首次计算时为每个链接附上一个初始PR值，实验中给出为1.0。

下面举个例子，这样能更清楚地知道map和reduce的输入输出对应该怎么写。 

1
2
3
4
A	B,C,D
B	A,C
C	A,B,D
D	A,C

为了计算PR(A)，就要知道所有链接到A的网站，PR值及其出链数。在例子中，这些网站就是B,C,D，

所以要得到类似 B,PR(B),2 这种值，再加上记号表明这是B到A的，可以形成<A, B,PR(B),2 >。MapReduce中会将同一个key的分到一个reduce中，因此在处理key为A时，会有如下输入对：

1
2
3
<A, B,PR(B),2>
<A, C,PR(C),3>
<A, D,PR(D),2>

就可以利用公式，计算出PR(A)。因为要进行迭代，所以还必须要有 A B,C,D 。

为了区分 ，将 B,PR(B),2 改为用 ; 分隔：B;PR(B);2。

初始值是1.0可以直接赋给网站，但是迭代中，要怎么获取网站当前的PR值呢？

在reduce中已经计算出了PR(A)，我们可以将它加到key中输出作为下一次map的输入key：A,PR(A)

TIPS：

（1）根据分析，map第一次输入为A B,C,D，迭代中的输入为<A,PR(A) , B,C,D>，是否需要写2个map函数？

答：可以只写一个map函数，在第一次中，可以利用KeyValueTextInputFormat，它将\t之前的作为key读入，为A；在迭代中读入的key为A,PR(A)。利用分隔符 , 可以得到相应的值。

（2）在实现中，所有与计算PR值有关的必须用double类型，因为要保留10位小数。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
private static final double d = 0.85;//阻尼系数
	public static class Map extends Mapper<Text, Text, Text, Text>{
		public void map(Text key, Text value, Context context) 
        		throws IOException, InterruptedException {
			//输入文件格式： a    b,c,d(first)  采用KeyValueTextInputFormat
			//           a,pr    b,c,d(next)
			String[] out = value.toString().split(",");//外链
			String[] link = key.toString().split(",");//next
			double pr = 1.0;//初始pr值
			if(link.length > 1) {
				//next
				pr = Double.parseDouble(link[1]);
			}
			int outNum = out.length;//a的出链数
			//输出格式<a的各出链:b/c/d, a;pr;outNum>
			for(String s: out) {
				context.write(new Text(s), 
						new Text(link[0]+";"+pr+";"+outNum));
			}
			//<a, out>以便迭代处理
			context.write(new Text(link[0]), value);
        }
	}
	
	public static class Reduce extends Reducer<Text, Text, Text, Text> {
		public void reduce(Text key, Iterable<Text> values, Context context) 
        		throws IOException, InterruptedException {
			double pr = (double) (1.0-d);// PageRank值  
	        String[] str;  
	        Text outLinks = new Text();// 记录该链接的所有出链信息  
	        // 集合的数据位key的所有入链链接的page,rank,count值，以及key的所有出链信息  
	        for (Text t : values) {  
	            // 入链信息以';'分割，出链信息以','分割，以此区别  
	            str = t.toString().split(";");  
	            if (str.length == 3) {  
	                // 计算key的rank值=(1-d)+d*key的入链rank值/其出链数  
	                pr += Double.parseDouble(str[1]) / Integer.parseInt(str[2])  * d;  
	            } else {  
	                outLinks.set(t.toString());  //<a,  b,c,d>
	            }  
	        }  
	        context.write(new Text(key.toString() + "," + pr), outLinks); 
        }
    }

main函数中的迭代：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
Configuration conf = new Configuration();
        conf.set("fs.defaultFS", "hdfs://localhost:9000");
		// 进行迭代的过程  
        for (int i = 0; i < 10; i++) { 
        	String[] otherArgs = new String[]{"/user/pr/out"+i, "/user/pr/out"+(i+1)};
        	if (otherArgs.length != 2) {
        		System.err.println("Usage: PageRank <in> <out>");
        		System.exit(2);
        	}
        	Job job = Job.getInstance(conf, "PageRank");
        	job.setJarByClass(PageRank.class);
        	job.setInputFormatClass(KeyValueTextInputFormat.class); 
        	job.setMapperClass(Map.class); 
        	job.setReducerClass(Reduce.class);
        	//设置Map输出类型
        	job.setMapOutputKeyClass(Text.class);
        	job.setMapOutputValueClass(Text.class);
        	//设置Reduce输出类型
        	job.setOutputKeyClass(Text.class);
        	job.setOutputValueClass(Text.class);
        	FileInputFormat.addInputPath(job, new Path(otherArgs[0]));
        	FileOutputFormat.setOutputPath(job, new Path(otherArgs[1]));
        	//System.exit(job.waitForCompletion(true) ? 0 : 1);
        	job.waitForCompletion(true); 
        }

到此，输出的结果为：

可以看到输出结果为：网站,PR 网站中的链接

与标准输出还有距离，因此，我添加了一对map和reduce用来标准化输出。

为了按照PR排列，结合reduce是按照key排列的，且顺序是从小到大，因此先将PR作为key，网站作为value。

又因为，在标准输出中，是按照PR从大到小进行排列。所以可以在PR前加个负号 - 。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
//10次迭代后，按格式输出  要按pr值排序，保留10位小数
	public static class FinMap extends Mapper<Text, Text, DoubleWritable, Text>{
		public void map(Text key, Text value, Context context) 
        		throws IOException, InterruptedException {
			//           a,pr    b,c,d(next)
			//String[] k = key.toString().split(",");
			//这样是从小到大排序
			//context.write(new Text(String.format("%.10f", Double.parseDouble(k[1]))), new Text(k[0]);
		   String line = key.toString();
            String url = line.split(",")[0];
            String pr = line.split(",")[1];
            context.write(new DoubleWritable(-Double.valueOf(pr)), new Text(url));
		}
	}
	
	public static class FinReduce extends Reducer<DoubleWritable, Text, NullWritable, Text> {
		public void reduce(DoubleWritable key, Iterable<Text> values, Context context) 
        		throws IOException, InterruptedException {
			for (Text value: values) {
                Double pr = -Double.valueOf(key.toString());
                String result = "(" + value.toString() + "," + String.format("%.10f", pr) + ")";
                context.write(null, new Text(result));
            }
        }
    }

相应的，在main函数中添加：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
String[] otherArgs = new String[]{"/user/pr/out"+10, "/user/pr/out"+11};
    	if (otherArgs.length != 2) {
    		System.err.println("Usage: PageRank <in> <out>");
    		System.exit(2);
    	}
    	Job job = Job.getInstance(conf, "PageRank");
    	job.setJarByClass(PageRank.class);
    	job.setInputFormatClass(KeyValueTextInputFormat.class); 
    	job.setMapperClass(FinMap.class); 
    	job.setReducerClass(FinReduce.class);
    	//设置Map输出类型
    	job.setMapOutputKeyClass(DoubleWritable.class);
    	job.setMapOutputValueClass(Text.class);
    	//设置Reduce输出类型
    	job.setOutputKeyClass(NullWritable.class);
    	job.setOutputValueClass(Text.class);
    	FileInputFormat.addInputPath(job, new Path(otherArgs[0]));
    	FileOutputFormat.setOutputPath(job, new Path(otherArgs[1]));
    	//System.exit(job.waitForCompletion(true) ? 0 : 1);
    	job.waitForCompletion(true);

结果：

✿✿ヽ(°▽°)ノ✿

赏

谢谢你请我吃糖果

扫一扫，分享到微信

微信分享二维码

tag:

    缺失模块。
    1、请确保node版本大于6.2
    2、在博客根目录（注意不是yilia根目录）执行以下命令：
    npm i hexo-generator-json-content --save

    3、在根目录_config.yml里添加配置： 

      jsonContent:
    meta: false
    pages: false
    posts:
      title: true
      date: true
      path: true
      text: false
      raw: false
      content: false
      slug: false
      updated: false
      comments: false
      link: false
      permalink: false
      excerpt: false
      categories: false
      tags: true

机会要自己把握<br><br>再勇敢一点吧<br>女孩