1、項目需求
一本英文書籍包含成千上萬個單詞，現在我們需要在大量的單詞中，找出相同字母組成的所有單詞
2、數據集
下面是一本英文書籍截取的一部分單詞內容（書籍內容是隨意寫的，主要目的是實現這種需求）
3、分析
1、先看如下圖
在上圖中，cat、act是相同字母組成的單詞，tar、rat也是相同字母組成的單詞，bar只有一個，它不顯示，因為沒有和它是相同字母組成單詞
2、基於以上分析，我們通過以下幾步完成
1、在Map階段，對每個word（單詞）按字母進行排序生成sortedWord，然后輸出key/value鍵值對(sortedWord,word)。
2、在Reduce階段，統計出每組相同字母組成的所有單詞
3、示意流程
1、以下面幾個單詞為例，找出相同字母組成的單詞。
cat
tar
bar
act
rat
2、經過map階段處理后，如下
key value
act cat
art tar
abr bar
act act
art rat
在這一階段中，對每個單詞按字母進行排序生成sortedWord，並將其作為key,然后輸出key/value鍵值對(sortedWord,word)：
3、經過框架處理后，如下（這一步框架自動幫我們實現）
act,[act, cat]
art,[cat, rat]
abr,[bar]
在一階段，根據key（sortedWord）進行分組
4、經過reduce階段處理，如下
key value
act act,cat
art rat,tar
在一階段，拼接一組中的所有單詞，然后輸出；需要說明一點，（abr,[bar]）這個被過濾掉了，因為這一組中就它一個
4、實現
上面已經分析完畢，下面我們就着手實現它。這里需要編寫三塊代碼內容：
1、map 函數、
2、reduce函數
3、一些用來運行作業的代碼。
1、map 函數
下面我們來編寫 Mapper 類，實現 map() 函數
1 /* 2 3 * 排序、分組 4 5 */ 6 public static class AnagramMapper extends Mapper<LongWritable, Text, Text, Text>{ 7 8 public void map(LongWritable key, Text value,Context context) throws IOException , InterruptedException { 9 10 // 將Text轉換成String 11 12 String text = value.toString(); 13 14 15 16 // 將String轉換成字符數組，為排序作准備 17 18 char[] textCharArr = text.toCharArray(); 19 20 21 22 // 使用 Arrays對數組進行排序 23 24 Arrays.sort(textCharArr); 25 26 27 28 // 排序后的字符串 29 30 String sortedText = new String(textCharArr); 31 32 33 34 context.write(new Text(sortedText), value); 35 36 } 37 }
這個 Mapper 類是一個泛型類型，它有四個形參類型，分別指定 map 函數的輸入鍵、輸入值、輸出鍵和輸出值的類型。 就本示例來說，輸入鍵是一個長整數偏移量（LongWritable），輸入值是一行文本（Text）， 其次輸出鍵、輸出值也是文本類型(Text)的。
map() 方法的輸入是一個鍵（key）和一個值（value），我們首先將 Text 類型的 value 轉換成 Java 的 String 類型， 之后將其轉換成字符數組，通過Arrays.sort方法對該字符數組進行排序，然后將排序后的單詞作為key,源單詞作為value輸出。map() 方法還提供了 Context 實例用於將輸出內容進行寫入。 在這種情況下，我們將key、value封裝成Text對象，並將它們進行寫入。
2、reduce函數
下面我們來編寫 Reducer類，實現reduce函數
/* * 統計相同字母組成的單詞 */ public static class AnagramReduce extends Reducer<Text, Text, Text, Text>{ public void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException { // 值 StringBuilder value = new StringBuilder(); // 計數 int count = 0; // 拼接單詞 for(Text text : values){ // 分割符, if(value.length() > 0){ value.append(","); } value.append(text); // 計數 count++; } // 因為我們要統計相同字母組成的單詞，所以相同字母組成的單詞個數大於等於2才會輸出 if(count > 1){ context.write(key, new Text(value.toString())); } } }
同樣，reduce 函數也有四個形式參數類型用於指定輸入和輸出類型。reduce 函數的輸入類型必須匹配 map 函數的輸出類型：即 Text 類型和Text類型。 在這種情況下，reduce 函數的輸出類型也是 Text 和 Text 類型。在 map 的輸出結果中，所有相同key的鍵值對被分配到同一個reduce執行，通過迭代來拼接所有單詞，並將一組中只有一個單詞的過濾掉
3、一些用來運行作業的代碼
1 public int run(String[] arg0) throws Exception { 2 3 // 加載配置 4 5 Configuration conf = new Configuration(); 6 7 8 9 // 輸出目錄，如果存在就刪除 10 11 Path path = new Path(arg0[1]); 12 13 FileSystem fileSystem = path.getFileSystem(conf); 14 15 if(fileSystem.isDirectory(path)){ 16 17 fileSystem.delete(path, true); 18 19 } 20 21 22 23 // 創建job對象 24 25 Job job = new Job(conf,"anagram"); 26 27 job.setJarByClass(Anagram.class); 28 29 30 31 // 指定輸入、輸出目錄 32 33 FileInputFormat.addInputPath(job, new Path(arg0[0])); 34 35 FileOutputFormat.setOutputPath(job, new Path(arg0[1])); 36 37 38 39 // 指定mapper、reduce 40 41 job.setMapperClass(AnagramMapper.class); 42 43 job.setReducerClass(AnagramReduce.class); 44 45 46 47 // 指定mapper、reduce的輸出類型 48 49 job.setOutputKeyClass(Text.class); 50 51 job.setOutputValueClass(Text.class); 52 53 54 55 // 提交作業 56 57 return job.waitForCompletion(true) ? 0: 1; 58 59 }
1 public static void main(String[] args) throws Exception { 2 3 String[] arg0 = { 4 5 "hdfs://ljc:9000/buaa/anagram", 6 7 "hdfs://ljc:9000/buaa/anagram/out" 8 9 }; 10 11 12 13 // 執行mapperreduce 14 15 int status = ToolRunner.run(new Configuration(), new Anagram(), arg0); 16 17 18 19 System.exit(status); 20 }
Configuration 類讀取 Hadoop 的配置文件，如 site-core.xml、mapred-site.xml、hdfs-site.xml 等。
Job 對象指定作業執行規范，我們可以用它來控制整個作業的運行。我們在 Hadoop 集群上運行這個作業時，要把代碼打包成一個JAR文件(Hadoop在集群上發布這個文件)。 不必明確指定 JAR 文件的名稱，在 Job 對象的 setJarByClass 方法中傳遞一個類即可，Hadoop 利用這個類來查找包含它的 JAR 文件，進而找到相關的 JAR 文件；構造 Job 對象之后，需要指定輸入和輸出數據的路徑
以下特別說明：
1、調用 FileInputFormat 類的靜態方法 addInputPath() 來定義輸入數據的路徑，這個路徑可以是單個的文件、一個目錄（此時，將目錄下所有文件當作輸入）或符合特定文件模式的一系列文件。由函數名可知，可以多次調用 addInputPath() 來實現多路徑的輸入。
2、調用 FileOutputFormat 類中的靜態方法 setOutputPath() 來指定輸出路徑（只能有一個輸出路徑）。這個方法指定的是 reduce 函數輸出文件的寫入目錄。 在運行作業前該目錄是不應該存在的，否則 Hadoop 會報錯並拒絕運行作業。這種預防措施的目的是防止數據丟失（長時間運行的作業如果結果被意外覆蓋，肯定是件可怕的事情）。
3、通過 setMapperClass() 和 setReducerClass() 指定 map 類型和reduce 類型。
4、通過setOutputKeyClass() 和 setOutputValueClass() 控制 map 和 reduce 函數的輸出類型，正如本例所示，這兩個輸出類型一般都是相同的。如果不同，則通過 setMapOutputKeyClass()和setMapOutputValueClass()來設置 map 函數的輸出類型。
5、輸入的類型通過 InputFormat 類來控制，我們的例子中沒有設置，因為使用的是默認的 TextInputFormat（文本輸入格式）。
6、Job 中的 waitForCompletion() 方法提交作業並等待執行完成。該方法中的布爾參數是個詳細標識，所以作業會把進度寫到控制台。 waitForCompletion() 方法返回一個布爾值，表示執行的成（true）敗（false），這個布爾值被轉換成程序的退出代碼 0 或者 1。
5、結果
6、其他問題
3、代碼及數據
如果，您認為閱讀這篇博客讓您有些收獲，不妨點擊一下右下角的【推薦】。
如果，您希望更容易地發現我的新博客，不妨點擊一下左下角的【關注我】。
如果，您對我的博客所講述的內容有興趣，請繼續關注我的后續博客，我是【劉超★ljc】。
本文版權歸作者和博客園共有，歡迎轉載，但未經作者同意必須保留此段聲明，且在文章頁面明顯位置給出原文連接，否則保留追究法律責任的權利。