Hadoop學習筆記(8) ——實戰 作個倒排索引

http://www.cnblogs.com/zjfstudio/p/3913549.html

Hadoop學習筆記(8)

——實戰 作個倒排索引

倒排索引是文檔檢索系統中最經常使用數據結構。根據單詞反過來查在文檔中出現的頻率，而不是根據文檔來，因此稱倒排索引(Inverted Index)。結構以下:

這張索引表中， 每一個單詞都對應着一系列的出現該單詞的文檔，權表示該單詞在該文檔中出現的次數。如今咱們假定輸入的是如下的文件清單：

T1 ： hello world hello china T2 : hello hadoop T3 ： bye world bye hadoop bye bye

 

輸入這些文件，咱們最終將會獲得這樣的索引文件：

bye    T3:4; china    T1:1; hadoop    T2:1;T3:1; hello    T1:2;T2:1; world    T1:1;T3:1;

 

接 下來，咱們就是要想辦法利用hadoop來把這個輸入，變成輸出。從上一章中，其實也就是分析如何將hadoop中的步驟個性化，讓其工做。整個步驟中， 最主要的仍是map和reduce過程，其它的均可稱之爲配角，因此咱們先來分析下map和reduce的過程將會是怎樣？

首 先是Map的過程。Map的輸入是文本輸入，一條條的行記錄進入。輸出呢？應該包含：單詞、所在文件、單詞數。 Map的輸入是key-value。 那這三個信息誰是key，誰是value呢？ 數量是須要累計的，單詞數確定在value裏，單詞在key中，文件呢？不一樣文件內的相同單詞也不能累加的，因此這個文件應該在key中。這樣key中就 應該包含兩個值：單詞和文件，value則是默認的數量1，用於後面reduce來進行合併。

因此Map後的結果應該是這樣的：

Key value Hello;T1 1 Hello:T1 1 World:T1 1 China:T1 1 Hello:T2 1 …

即然這個key是複合的，因此常歸的類型已經不能知足咱們的要求了，因此得設置一個複合健。複合健的寫法在上一章中描述到了。因此這裏咱們就直接上代碼：

1. public static class MyType implements WritableComparable<MyType>{
2.       public MyType(){
3.       }
4.  
5.       private String word;
6.       public String Getword(){return word;}
7.       public void Setword(String value){ word = value;}
8.  
9.       private String filePath;
10.       public String GetfilePath(){return filePath;}
11.       public void SetfilePath(String value){ filePath = value;}
12.  
13.       @Override
14.       public void write(DataOutput out) throws IOException {
15.          out.writeUTF(word);
16.          out.writeUTF(filePath);
17.       }
18.  
19.       @Override
20.       public void readFields(DataInput in) throws IOException {
21.          word = in.readUTF();
22.          filePath = in.readUTF();
23.       }
24.  
25.       @Override
26.       public int compareTo(MyType arg0) {
27.             if (word != arg0.word)
28.                return word.compareTo(arg0.word);
29.          return filePath.compareTo(arg0.filePath);
30.       }
31. }

有了這個複合健的定義後，這個Map函數就好寫了：

1. public static class InvertedIndexMapper extends
2.          Mapper<Object, Text, MyType, Text> {
3.  
4.       public void map(Object key, Text value, Context context)
5.             throws InterruptedException, IOException {
6.  
7.          FileSplit split = (FileSplit) context.getInputSplit();
8.          StringTokenizer itr = new StringTokenizer(value.toString());
9.  
10.          while (itr.hasMoreTokens()) {
11.             MyType keyInfo = new MyType();
12.             keyInfo.Setword(itr.nextToken());
13.             keyInfo.SetfilePath(split.getPath().toUri().getPath().replace("/user/zjf/in/", ""));
14.             context.write(keyInfo, new Text("1"));
15.          }
16.       }
17.    }

注意：第13行，路徑是全路徑的，爲了看起來方便，咱們把目錄替換掉，直接取文件名。

 

有了Map，接下來就能夠考慮Recude了，以及在Map以後的Combine。Map的輸出的Key類型是MyType，因此Reduce以及Combine的輸入就必須是MyType了。

若是直接將Map的結果送到Reduce後，發現還須要作大量的工做來將Key中的單詞再重排一下。因此咱們考慮在Reduce前加一個Combine，先將數量進行一輪合併。

這個Combine將會輸入下面的值：

Key value bye    T3:4; china    T1:1; hadoop    T2:1; hadoop    T3:1; hello    T1:2; hello    T2:1; world    T1:1; world    T3:1;

代碼以下：

1. public static class InvertedIndexCombiner extends
2.          Reducer<MyType, Text, MyType, Text> {
3.  
4.       public void reduce(MyType key, Iterable<Text> values, Context context)
5.             throws InterruptedException, IOException {
6.          int sum = 0;
7.          for (Text value : values) {
8.             sum += Integer.parseInt(value.toString());
9.          }
10.          context.write(key, new Text(key.GetfilePath()+ ":" + sum));
11.       }
12.    }

 

有了上面Combine後的結果，再進行Reduce就容易了，只須要將value結果進行合併處理：

1. public static class InvertedIndexReducer extends
2.          Reducer<MyType, Text, Text, Text> {
3.  
4.       public void reduce(MyType key, Iterable<Text> values, Context context)
5.             throws InterruptedException, IOException {
6.          Text result = new Text();
7.  
8.          String fileList = new String();
9.          for (Text value : values) {
10.             fileList += value.toString() + ";";
11.          }
12.          result.set(fileList);
13.  
14.          context.write(new Text(key.Getword()), result);
15.       }
16.    }

    通過這個Reduce處理，就獲得了下面的結果：

bye    T3:4; china    T1:1; hadoop    T2:1;T3:1; hello    T1:2;T2:1; world    T1:1;T3:1;

 

最後，MapReduce函數都寫完後，就能夠掛在Job中運行了。

1. public static void main(String[] args) throws IOException,
2.          InterruptedException, ClassNotFoundException {
3.       Configuration conf = new Configuration();
4.       System.out.println("url:" + conf.get("fs.default.name"));
5.  
6.       Job job = new Job(conf, "InvertedIndex");
7.       job.setJarByClass(InvertedIndex.class);
8.       job.setMapperClass(InvertedIndexMapper.class);
9.       job.setMapOutputKeyClass(MyType.class);
10.       job.setMapOutputValueClass(Text.class);
11.  
12.       job.setCombinerClass(InvertedIndexCombiner.class);
13.       job.setReducerClass(InvertedIndexReducer.class);
14.  
15.       job.setOutputKeyClass(Text.class);
16.       job.setOutputValueClass(Text.class);
17.  
18.       Path path = new Path("out");
19.       FileSystem hdfs = FileSystem.get(conf);
20.       if (hdfs.exists(path))
21.          hdfs.delete(path, true);
22.  
23.       FileInputFormat.addInputPath(job, new Path("in"));
24.       FileOutputFormat.setOutputPath(job, new Path("out"));
25.       job.waitForCompletion(true);
26. }

注：這裏爲了調試方便，咱們把in和out都寫死，不用傳入執行參數了，而且，每次執行前，判斷out文件夾是否存在，若是存在則刪除。