pagerank在hadoop上的运行

      从2月1日开始，我们云计算小组也搞了四天的研究了，虽然人没有到到齐，但我们还是在一位学长的带领下完成了一些任务，现在先来总结以下吧！

      直到现在，我们已经将pagerank的计算在自己的hadoop伪分布式机器上运行出来了！在集群上运行的代码也写出来了，只是没有在hadoop集群上进行测试了，因为昨天集群出来问题，今天主要的任务就是在集群上将成寻运行起来！

      顺便说一下昨天集群出现的问题和解决的方案：集群出现的故障是namenode启动不起来，查看它的logs日志的时候，里面什么都没有，所以我们决定将集群的datanode全部停掉，将namenode重新格式化一下（虽然这样会导致hdfs上的数据全部清空，我们也还不会怎样用secondnamenode来进行恢复数据，但这也是我们唯一能想到的办法了），但是在格式化之后还是不能启动namenode，纠结了好久...终于当我们把namenode上的datalog原数据日志文件删掉之后才搞定！之后启动datanode的时候，也要将其data1,data2日志文件删掉才能启动，因为每启动一次，都会生成日志文件，而如果之前存在的话，集群会认为你已经启动了，所以它不会再启动。

       好了，来说说我们这几天一直在研究的pagerank吧！

       之前研究过google的pagerank了，其中给出了一个公式：R(i)=(1-c)+c*（R(1)/T(1)+...+R(j)/T(j)+...）,c是阻尼系数0.85，R(i)是第i张网页的PR值，R(j)是第i张网页链入的第j张网页的PR值，T(j)是第j张网页上链出的网页数。现在我们可以根据迭代的方法将PR值算出来了，代码就不贴出来了，很简单的！只是其中怎样将迭代的方法和求矩阵的最大特征值的特征向量联系起来的还是木有搞清楚，昨晚又研究了好几个小时，结果还是木有证出来，只怪自己数学木有学好呀。。。希望有知道的牛人能帮我解答一下～～～

       我们最主要的是研究hadoop运行的机制，mapreduce是怎样执行任务的，研究过wordcount例子的人应该都知道，在提交任务的时候将配置说明一下，其中包括要执行的mapper函数，reduce函数，输入输出的key的类型，输入输出的value的类型，输入输出的文件路径等，我们在不晓得hadoop原理的时候，只能拿别人已经做出来的例子进行模仿了，修改修改配置，自己写一写mapper和reduce函数类，慢慢的就会知道一些运行的方式了！首先我们只用mapreduce实现了一次pagerank公式的迭代，因为没有什么思路，后来通过看一些资料之后，才发现我们可以将reduce函数整合了之后的输出文件格式和mapper函数的输入文件的格式弄成一样就行了！只是数据类型的问题了，虽然想到这一步了，但具体要怎样实现还是要动手实现的，因为我之前在研究输入文件的解析去了，那个迭代的过程就没有自己去实现，结果回来自己搞的时候，那真的是漏洞百出呀！因为hadoop上面好多数据类型的都是封装好了的，而我对于hadoop的数据传输是怎样的也不能肯定，最后摸索了好久，才写成下面的代码：

     注：我的输入文件是放在/home/acer/myranking/out00文件夹下面的，每一行都是这样的：网页id+/t+初始pr值+一个空格+这张网页的链出网页的id（其中网页id之间用空格分开），例如：7        1 2 3 4 为一行，7，2，3，4为网页id号，1为网页7的pr初始值

package com.apache.hadoop.io;

import java.io.IOException;
import java.text.DecimalFormat;
import java.text.NumberFormat;
import java.util.Iterator;

import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.DoubleWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.FileInputFormat;
import org.apache.hadoop.mapred.FileOutputFormat;
import org.apache.hadoop.mapred.JobClient;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.MapReduceBase;
import org.apache.hadoop.mapred.Mapper;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reducer;
import org.apache.hadoop.mapred.Reporter;
import org.apache.hadoop.mapred.TextInputFormat;
import org.apache.hadoop.mapred.TextOutputFormat;


public class PRtwo {

	// 实现Mapper内部类
	public static class PRMapper extends MapReduceBase implements
			Mapper<LongWritable, Text, Text, Text> {

		// 每一个网页的id即为key
		Text id;
		//map执行的方法
		public void map(LongWritable key, Text value, OutputCollector<Text, Text> output, Reporter reporter) throws IOException {
			// 得到一行的值
			String line = value.toString();
			// 将其分开成key和value
			String[] values = line.split("\t");
			// 将其赋值到key和value上
			// key.set(values[0]);
			Text t = new Text(values[0]);
			value.set(values[1]);
			// 再将value里面的东西分开来
			String[] ids = value.toString().split(" ");
			// 每一个网页初始pr值
			DoubleWritable pr = new DoubleWritable(Double.valueOf(ids[0]));
			//后面的链接地址
			String url="_";
			// 进行存储
			for (int i = 1; i < ids.length; i++) {
				url+=ids[i]+" ";
				id = new Text(ids[i]);
				// 向结果里面写进key-value值
				output.collect(id, new Text(String.valueOf(Double.valueOf(ids[0])/ (ids.length - 1))+""));
			}
			
			//还要将后面的链接加上来
			output.collect(t, new Text(url));
		}

	}

	// 实现Reducer内部类
	public static class PRReducer extends MapReduceBase implements Reducer<Text, Text, Text, Text>{

		// 网页迭代一次的pr值
		private Double pr;

		// reduce执行的方法
		 public void reduce(Text key, Iterator<Text> values, OutputCollector<Text, Text> out, Reporter reporter){
			double sum = 0;
			//将网页链接出去的地址找出来
			String url = "";
			// 进行循环相加
			while(values.hasNext()) {
				//要判断是不是网页
				String dw = values.next().toString();
				if(!dw.contains("_")){
					//转换成double进行相加
					sum += Double.valueOf(dw);
				}else{
					url= dw.replace("_", " ");
				}
				
			}
			pr=0.15 + 0.85 * sum;
			try {
				out.collect(key, new Text(pr+url+""));
			} catch (IOException e) {
				e.printStackTrace();
			}
		}
	}

	// 主运行函数
	public static void main(String[] args) throws Exception {

		NumberFormat nf = new DecimalFormat("00");

		// 进行迭代的过程
		for (int i = 0; i < 20; i++) {
			// 文件的输入输出路径
			String[] otherArgs = {"/home/acer/myranking/out"+nf.format(i), "/home/acer/myranking/out"+nf.format(i + 1)};
			if (otherArgs.length != 2) {
				System.err.println("Usage:wordcount <in><out>");
				System.exit(2);
			
			JobConf conf = new JobConf(PRtwo.class);
	        
	        conf.setOutputKeyClass(Text.class);
	        conf.setOutputValueClass(Text.class);
	        
	        conf.setInputFormat(TextInputFormat.class);
	        conf.setOutputFormat(TextOutputFormat.class);
	        
	        FileInputFormat.setInputPaths(conf, new Path(otherArgs[0]));
	        FileOutputFormat.setOutputPath(conf, new Path(otherArgs[1]));
	        
	        conf.setMapperClass(PRMapper.class);
	        conf.setReducerClass(PRReducer.class);
	        
	        JobClient.runJob(conf);
			
			
		}
		}

	}
}

     我的代码中是将mapper和reduce函数作为内部类了，这里我是迭代了20次，基本收敛了，这里的输入文件的格式是我们自己定义的，我们知道要计算网页的pr值，肯定要有网页，那么我们只有网页，怎么得到我们想要的数据文件格式呢？！我们这里自然就想到用网页的xml文件来解析了，昨天研究了一下这个方面的东西，今天做一下测试再来分析吧！

     希望大家有什么看法的可以讨论一下！