使用 Hadoop 進行語料處理（面試題）

時間 2019-11-20

原文原文鏈接

原創播客，如需轉載請註明出處。原文地址：http://www.cnblogs.com/crawl/p/7751741.html html

----------------------------------------------------------------------------------------------------------------------------------------------------------面試

筆記中提供了大量的代碼示例，須要說明的是，大部分代碼示例都是本人所敲代碼並進行測試，不足之處，請你們指正~數組

本博客中全部言論僅表明博主本人觀點，如有疑惑或者須要本系列分享中的資料工具，敬請聯繫 qingqing_crawl@163.com緩存

-----------------------------------------------------------------------------------------------------------------------------------------------------------服務器

前言：近日有高中同窗求助，一看題目，正好與樓主所學技術相關，便答應了下來，由於正是興趣所在，也沒有管能不能實現。app

話很少說，先上題：框架

同窗告訴我，這是浪潮實習生的面試題。ide

樓主先對題目做簡單的說明，給定了一個 corpus.txt 文件做爲語料處理的源文件，文件大小 30.3M，內容即題目要求中的圖片所示，要求對語料文件中出現的詞進行詞頻統計，並把詞頻相同的詞語用 ## 相連（如研究##落實 1008 ），並按詞頻從大到小排序。題目要求的是根據所學的 Java I/O 處理、集合框架、字符集與國際化、異常處理等基礎知識完成此題，但同窗代表可使用大數據的相關知識，讓樓主感到興趣的是，最近樓主一直在研究 Hadoop，詞頻統計的題目作了很多，便欣然接受同窗的求助。本身挖的坑總要填的，如果進行簡單的詞頻統計，很簡單，涉及到將相同詞頻的詞語排在一行並用 ## 鏈接，由於樓主水平有限，在實現的過程當中仍是遇到了很多的困難。工具

1、簡單實現詞頻統計

剛入手此題目，樓主的思路就是先實現一個簡單的詞頻統計，而後在實現簡單詞頻統計的基礎上，對代碼進行修改，實現相同詞頻的詞語放在一行使用 ## 鏈接。思路很簡單，簡單的詞頻統計實現的也很是順利，但接下的思路實現起來便沒有那麼容易了。先來看一看簡單的詞頻統計這個功能吧。oop

1. 首先先來寫 Mapper 的功能，上代碼：

 1 public class WordHandlerMapper extends Mapper<LongWritable, Text, Text, LongWritable> {
 2 
 3     @Override
 4     protected void map(LongWritable key, Text value, Context context)
 5             throws IOException, InterruptedException {
 6 
 7         String[] strs = StringUtils.split(value.toString(), "\t");
 8 
 9         for(String str : strs) {
10             int index = str.indexOf("/");
11             
12             if(index < 0) {
13                 index = 0;
14             }
15             
16             String word = str.substring(0, index);
17             context.write(new Text(word), new LongWritable(1));
18             
19         }
20         
21     }
22 
23 }

WordHandlerMapper 類的功能，首先此類實現了 Mapper 類，重寫了 mapper 方法，使用默認的 TextOutputFormat 類，將讀取到的一行數據以形參 value 的形式傳入 mapper 方法，第 7 行對這行數據也就是 value 進行處理，以 \t 進行分割，獲得了一個 String 數組，數組的形式爲：["足協/j", "盃賽/n", "常/d"] 形式，而後 10 行對數組進行遍歷，而後獲取到 /以前的內容，也就是咱們須要統計詞頻的詞語，如 10 - 16 行所示，而後將獲得的詞語傳入 context 的 write 方法，map 程序進行緩存和排序後，再傳給 reduce 程序。

2. 而後再來看 Reduce 程序：

 1 public class WordHandlerReducer extends Reducer<Text, LongWritable, Text, LongWritable> {
 2     
 3     
 4     @Override
 5     protected void reduce(Text key, Iterable<LongWritable> values, Context context)
 6             throws IOException, InterruptedException {
 7         
 8         long count = 0;
 9         
10         for(LongWritable value : values) {
11             count += value.get();
12         }
13         
14         context.write(key, new LongWritable(count));
15         
16     }
17 
18 }

WordHandlerReducer 的邏輯很簡單，此類繼承了 Reduce，重寫了 reduce 方法，傳入的 key 即須要統計詞頻的詞語，vaues 爲 {1,1,1} 形式，8 行定義了一個計數器，而後對 values 進行加強 for 循環遍歷，使計數器加 1，而後將詞語和詞頻輸出便可。

3. 而後咱們再定義個類來描述這個特定的做業：

 1 public class WordHandlerRunner {
 2     
 3     public static void main(String[] args) throws Exception {
 4         
 5         Configuration conf = new Configuration();
 6         Job wcJob = Job.getInstance(conf);
 7         
 8         wcJob.setJarByClass(WordHandlerRunner.class);
 9         
10         wcJob.setMapperClass(WordHandlerMapper.class);
11         wcJob.setReducerClass(WordHandlerReducer.class);
12         
13         wcJob.setOutputKeyClass(Text.class);
14         wcJob.setOutputValueClass(LongWritable.class);
15         
16         wcJob.setMapOutputKeyClass(Text.class);
17         wcJob.setMapOutputValueClass(LongWritable.class);
18         
19         FileInputFormat.setInputPaths(wcJob, new Path("/wc/srcdata2/"));
20         
21         FileOutputFormat.setOutputPath(wcJob, new Path("/wc/wordhandler/"));
22         
23         wcJob.waitForCompletion(true);
24         
25     }
26 
27 }

第 6 行獲得 Job 對象，以後即是對一些基本參數的設置，基本上就是見方法名而知其意了。 19 和 21 行定義存放元數據的路徑和輸入結果的路徑，23行提交做業。

在 Eclipse 中將程序打成 jar 包,名爲 wordhandler.jar 導出，上傳到 Linux 服務器，將原始的語料處理文件上傳到程序中指定了路徑下，經過

hadoop jar wordhandler.jar com.software.hadoop.mr.wordhandler.WordHandlerRunner 命令執行，很快就會執行完畢，而後到輸出路徑中查看輸出結果(圖片展現部分結果):

到此簡單的詞頻統計功能就到此結束了，觀察可知,MapReduce 默認是按輸出的 key 進行排序的。獲得的數據距離題目要求的結果還有很大的懸殊，那麼剩下須要進一步實現的還有兩處，一處是將詞頻相同的詞語放到一行並用 ## 鏈接，第二處就是對詞頻進行排序（按從大到小） 。

樓主的思路是，排序確定是要放到最後一步實現，若先進行排序，在對相同詞頻的詞語處理的話，頗有可能會打亂以前的排序。那麼，如今就是對詞頻相同的詞語進行處理了，使它們顯示在一行並用 ## 鏈接。在實現這個功能時，樓主遇到了困難，主要是對 TextOutputFarmat 默認只讀取一行數據意識不夠深刻，走了許多的彎路，好比樓主相到修改 Hadoop 的源碼，讀取多行數據等，但因爲樓主水平有限，結果以失敗了結。到此時已是夜裏將近十二點了，由於到次日還要早起，因此就沒再熬夜，暫時放下了。

次日中午，樓主想到了倒排索引，使用倒排索引實現的思路便一點一點造成。把咱們以前獲得的數據讀入，而後將詞頻當作 key，這樣 Mapper 程序便會在 Reduce 執行以前進行緩存和分類，思路來了，便立刻動手實現。

2、使用倒排索引初步實現相同詞頻寫入一行並用 ## 鏈接

1. 仍是先來 Mapper 的功能：注意，此次讀入的數據是咱們以前獲得的按默認的形式排好序，並統計出詞頻的數據。

 1 public class WordHandlerMapper2 extends Mapper<LongWritable, Text, LongWritable, Text> {
 2 
 3     @Override
 4     protected void map(LongWritable key, Text value, Context context)
 5             throws IOException, InterruptedException {
 6 
 7         String[] strs = StringUtils.split(value.toString(), "\t");
 8 
 9         String text = strs[0];
10         
11         long count = Long.parseLong(strs[1]);
12         
13         context.write(new LongWritable(count), new Text(text));
14         
15     }
16 
17 }

Map 的功能很簡單，咱們須要輸出的 key 是 LongWritable 類型，value 是 Text 類型，即 [205, {"檢驗", "加入", "生存"}] 這種類型。第 7 行一樣是對一行的數據進行拆分，而後獲得詞語(text) 和詞頻(count),而後第 13 行進行輸出便可，很簡單。

2. Reduce 程序的功能：

 1 public class WordHandlerReducer2 extends Reducer<LongWritable, Text, Text, LongWritable> {
 2     
 3     //key: 3  values: {"研究","落實"}
 4     @Override
 5     protected void reduce(LongWritable key, Iterable<Text> values, Context context)
 6             throws IOException, InterruptedException {
 7         
 8         String result = "";
 9         
10         for(Text value : values) {
11             result += value.toString() + "##";
12             
13         }
14         
15         context.write(new Text(result), key);
16         
17     }
18 
19 }

Reduce 的邏輯比以前的稍微複雜一點，從 Mapper 中輸入的數據格式爲 [205, {"檢驗", "加入", "生存"}] 類型，咱們但願輸出的格式爲：[檢驗##加入##生存　　205], 重寫的 reduce 方法傳入的 values 即 {"檢驗", "加入", "生存"} 類型，第十行對 values 進行遍歷，11 行向 result 中追加，便可獲得咱們須要的結果，而後 15 行進行輸出。

3. 而後再來定義一個類來描述此做業，

 1 public class WordHandlerRunner2 {
 2     
 3     public static void main(String[] args) throws Exception {
 4         
 5         Configuration conf = new Configuration();
 6         Job wcJob = Job.getInstance(conf);
 7         
 8         wcJob.setJarByClass(WordHandlerRunner2.class);
 9         
10         wcJob.setMapperClass(WordHandlerMapper2.class);
11         wcJob.setReducerClass(WordHandlerReducer2.class);
12         
13         wcJob.setOutputKeyClass(Text.class);
14         wcJob.setOutputValueClass(LongWritable.class);
15         
16         wcJob.setMapOutputKeyClass(LongWritable.class);
17         wcJob.setMapOutputValueClass(Text.class);
18         
19         FileInputFormat.setInputPaths(wcJob, new Path("/wc/wordhandler/"));
20         
21         FileOutputFormat.setOutputPath(wcJob, new Path("/wc/wordhandleroutput/"));
22         
23         wcJob.waitForCompletion(true);
24         
25     }
26 
27 }

這個類與以前的那個 Job 描述類很相似，使用的 Job 的方法沒有變化，方法的參數只作稍微修改便可，樓主標紅的行即須要進行修改的行。

而後是一樣的步驟，在 Eclipse 中將程序打成 jar 包導出，也叫 wordhandler.jar，而後上傳到 Linux 服務器中，使用

hadoop jar wordhandler.jar com.software.hadoop.mr.wordhandler2.WordHandlerRunner2 進行運行，同過 Map 和 Reduce 的處理後，進入相應的目錄下，查看結果（圖片展現部分結果）：

咱們分析一下獲得的結果，是否是距離題目要求的輸出結果更接近了一步，可是還差點事，一個是每一行的最後多了一個 ##，這個好解決，在生成字符串的時候判斷該詞語是否爲最後一個便可，另外一個就是題目要求詞頻按從大到小的順序輸出，而咱們的輸出順序是從小到大。明確了問題以後，繼續開動吧。

3、實現按詞頻從大到小進行排序

排序問題是使用 Hadoop 進行詞頻處理的常見問題了，實現起來並不困難。說一說思路，由於咱們這裏是默認讀取一行，那麼咱們構造一個 Word 類，此類有屬性 text （內容），和（count）詞頻，此類須要實現 WritableComparable 接口，重寫其中的方法，使用咱們自定義的排序方式便可。既然思路明確了，那咱們一步一步的實現。

1. 先定義一個 word 類：

 1 public class Word implements WritableComparable<Word> {
 2     
 3     private String text;
 4     
 5     private long count;
 6     
 7     public Word() {}
 8     
 9     public Word(String text, long count) {
10         super();
11         this.text = text;
12         this.count = count;
13     }
14 
15     public String getText() {
16         return text;
17     }
18 
19     public void setText(String text) {
20         this.text = text;
21     }
22 
23     public long getCount() {
24         return count;
25     }
26 
27     public void setCount(long count) {
28         this.count = count;
29     }
30 
31     @Override
32     public void write(DataOutput out) throws IOException {
33         out.writeUTF(text);
34         out.writeLong(count);
35     }
36 
37     @Override
38     public void readFields(DataInput in) throws IOException {
39         text = in.readUTF();
40         count = in.readLong();
41     }
42 
43     @Override
44     public int compareTo(Word o) {
45         return count > o.getCount() ? -1 : 1;
46     }
47 
48 }

此類須要實現 WritableComparable 接口，重寫第 32 行的 write 方法，第 38 行的 readFields 方法，第 44 行 compareTo 方法，32 行和 38 行的方法是 Hadoop 中序列化相關的方法，44 行 compareTo 方法纔是咱們自定義排序方式的方法。值的注意的是，write 方法中和 readFields 方法中屬性的序列化和反序列化的順序必須一致，即 3三、34 和 3九、40 行的屬性須要對應。而後 compareTo 中第 45 行實現自定義的從大到小的排序便可。

2. Mapper 類的功能：

 1 public class WordHandlerMapper3 extends Mapper<LongWritable, Text, Word, NullWritable> {
 2 
 3     @Override
 4     protected void map(LongWritable key, Text value, Context context)
 5             throws IOException, InterruptedException {
 6 
 7         String[] strs = StringUtils.split(value.toString(), "\t");
 8         
 9         String text = strs[0];
10         
11         long count = Long.parseLong(strs[1]);
12         
13         Word word = new Word(text, count);
14         
15         context.write(word, NullWritable.get());
16         
17     }
18 
19 }

咱們定義 Mapper 的輸出的 key 爲 Word 類型是排序成功的關鍵，在 mapper 方法中常規的拆分一行數據，得到到相應的字段，而後第 13 行封裝爲一個 Word 對象，第 15 行輸出便可，樓主定義 Mapper 的輸出的 vlaue 爲 NullWritable 類型，思路爲只要輸出的 key 爲 Word 型，那麼咱們就能夠獲取到須要的信息了。

2. 再來看 Reduce 的功能：

1 public class WordHandlerReducer3 extends Reducer<Word, NullWritable, Text, LongWritable> {
2     
3     @Override
4     protected void reduce(Word key, Iterable<NullWritable> values, Context context)
5             throws IOException, InterruptedException {
6         context.write(new Text(key.getText()), new LongWritable(key.getCount()));
7     }
8 
9 }