Hadoop-文档倒排索引算法实现

2018-05-13

关于Hadoop的安装以及实验一可以参考：

Tiny_16 Hadoop学习
下面来看实验二：文档倒排索引算法实现
（debug的过程还是挺艰辛的）

实验内容包括：（1）去除stop-words　（2）统计单词在每篇文档中出现的频率

一开始的思路我是参考的Hadoop之倒排索引，感觉挺好的，但是做到后面发现有问题。

还是从头开始吧。

（1）首先是处理停用词表

利用MapReduce的分布式缓存

1 2	//添加停用词表 job.addCacheFile(new Path("hdfs://10.102.0.197:9000/stop_words/stop_words_eng.txt").toUri());

在一开始进行处理，所以在map函数中通过context来访问到缓存的文件，重写setup方法来进行初始化。setup方法用于初始化，只会执行一次。将停用词表的每一个词读出存入set中。（感觉在reduce的setup方法中处理也可以）

private Set<String> stopwords = new TreeSet<String>();
	private URI[] file;
	public void setup(Context context) 
			throws IOException, InterruptedException{
		//初始化 只执行一次
		Configuration conf = context.getConfiguration();
		//获得停用词表
		file = Job.getInstance(conf).getCacheFiles();
		for (int i = 0; i < file.length; i++){
			String line;
			Path pa = new Path(file[i].getPath());
			BufferedReader br = new BufferedReader(
                new FileReader(pa.getName().toString()));
			while((line = br.readLine()) != null){
				StringTokenizer itr = new StringTokenizer(line);
				while (itr.hasMoreTokens()){
					stopwords.add(itr.nextToken());
				}
			}
		}
	}

（2）分析MapReduce的输入、输出对。

按照之前链接上的思路，与给的标准答案对比，会出现txt排序不正确的问题，因为MapReduce的默认排序是对key。因此必须将txt也作为reduce输入key的一部分。

map类：输入对<offset, line>，输出对 < key txt, 1 >

tips：中间分隔符应该用ASCII表排在数字之前的符号，否则单个数字就会被排到后面去

//<offset,line>--><word:txt,1>
private static final IntWritable one = new IntWritable(1);
private Text key1 = new Text();
//private Text value1 = new Text();
private FileSplit split;
@Override
public void map(Object key, Text value, Context context) 
    throws IOException, InterruptedException {
    split = (FileSplit) context.getInputSplit();
    //StringTokenizer itr = new StringTokenizer(value.toString());
    String[] itr = value.toString().split("\\W");
    //[^a-zA-Z0-9] only letters
    for (String s : itr) {
        s = s.toLowerCase();
        if(!stopwords.contains(s) && !s.equals("")) {
        	key1.set(s + " "+ split.getPath().getName());
        	//必须用空格作为分隔符吗？？ASCII
        	//用：就会不对
        	//value1.set("1");
        	context.write(key1, one);
        }
    }
}

为了方便后面的计算频数，在map和reduce之间添加一个combine类，先将一个词在同一个txt的出现次数加起来。

public static class Combine extends Reducer<Text, IntWritable, Text, IntWritable>{
    	//private IntWritable value2 = new IntWritable(1);
    	//private Text key2 = new Text();
    	//private Text value2 = new Text();
    	private IntWritable value2 = new IntWritable();
    	public void reduce(Text key, Iterable<IntWritable> values, Context context) 
    			throws IOException, InterruptedException{
    		int sum = 0;//frequence
    		for(IntWritable val : values) {
       			sum += val.get();
    		}
    		/*
    		int splitIndex = key.toString().indexOf(":");
    		value2.set(key.toString().substring(splitIndex+1)+","+sum);//txt,sum
    		key.set(key.toString().substring(0, splitIndex));//word
    		//key2.set(key.toString().substring(0, splitIndex));//word
    		//String[] itr = key.toString().split(":");
    		//key2.set(itr[0]);
    		//value2.set(itr[1]+","+sum);
    		context.write(key, value2);//<word:txt,"1"> -> <word, txt,sum>
    		//value2.set(sum);
    		//context.write(key, value2);//<word:txt, 1> -> <word:txt, sum>
    		 * */
    		//这样文件的位置不对 所以key必须是<word:txt>
    		value2.set(sum);
    		context.write(key, value2);
    	}
    }

因为当前key为word txt，为了将同一个word分到一个reduce上，需要自定义partioner类。

   //所以为了将按word分到一起...
   public static class Partitioner extends HashPartitioner<Text, IntWritable>
{
	public int getPartition(Text key, IntWritable value, int numReduceTasks){
		String word = new String();
		word = key.toString().split(" ")[0];	//=> word
		return super.getPartition(new Text(word), value, numReduceTasks);
	}
}

最后的reduce类：输入，输出<word, ..(格式).>

public static class Reduce extends Reducer<Text, IntWritable, Text, Text> {
    	//<word   ,txt,sum> -> <word, <>;<>>  ×
    	//change to <word:txt, sum> -> <word, <>;<>.>
    	private Text key3 = new Text();
        private Text value3 = new Text();
        private String cur = " ";//记录word
        private String list = "";//记录文件列表
        private long total = 0;
        public void reduce(Text key, Iterable<IntWritable> values, Context context) 
        		throws IOException, InterruptedException {
        	 int temp = 0;
        	 for(IntWritable val : values) {
        		 temp += val.get();
        	 }
        	 if(!cur.equals((key.toString().split(" "))[0]) && !cur.equals(" ")) {
        		 //如果cur不等于当前的word,说明上一个已经结束，下一个开始
        		 //输出上一个total
        		 list += "<total,"+total+">.";
        		 context.write(new Text(cur), new Text(list));
        		 //下一个初始化
        		 list = "";
        		 total = 0;
        	 }
        	 cur = (key.toString().split(" "))[0];
        	 list += "<"+(key.toString().split(" "))[1]+","+temp+">;";
        	 total += temp;
        }
        
        public void cleanup(Context context) throws IOException, InterruptedException{
        	//释放 只执行一次
			//最后一个word
        	list += "<total,"+total+">.";
   		 	context.write(new Text(cur), new Text(list));
		}
    }

在main方法中，要注意设置的map和reduce的输出和前面写的一样，否则会报错，(╥╯^╰╥)

还要注意的是上一个输出对和下一个输入对类型要一致，否则会报错。

由此，实验二结束，✿✿ヽ(°▽°)ノ✿

实验结果