程序實例與分析

Hadoop 是一個實現了MapReduce 計算模型的開源分布式并行編程框架，借助于Hadoop, 程序員可以輕松地編寫分布式并行程序，將其運行于計算機集群上，完成海量數據的計算。在本文中，詳細介紹了如何針對一個具體的并行計算任務，基于Hadoop 編寫程序，如何使用 IBM MapReduce Tools 在 Eclipse 環境中編譯并運行 Hadoop 程序。

前言

在上一篇文章：“用 Hadoop 進行分布式并行編程 第一部分 基本概念與安裝部署”中，介紹了 MapReduce 計算模型，分布式文件系統 HDFS，分布式并行計算等的基本原理, 并且詳細介紹了如何安裝 Hadoop，如何運行基于 Hadoop 的并行程序。在本文中，將針對一個具體的計算任務，介紹如何基于 Hadoop 編寫并行程序，如何使用 IBM 開發的 Hadoop Eclipse plugin 在 Eclipse 環境中編譯并運行程序。

分析?WordCount?程序

我們先來看看 Hadoop 自帶的示例程序?WordCount， 這個程序用于統計一批文本文件中單詞出現的頻率，完整的代碼可在下載的 Hadoop 安裝包中得到(在 src/examples 目錄中)。

1.實現Map類

見代碼清單1。這個類實現 Mapper 接口中的 map 方法，輸入參數中的 value 是文本文件中的一行，利用 StringTokenizer 將這個字符串拆成單詞，然后將輸出結果 <單詞,1> 寫入到 org.apache.hadoop.mapred.OutputCollector 中。OutputCollector 由 Hadoop 框架提供, 負責收集 Mapper 和 Reducer 的輸出數據，實現 map 函數和 reduce 函數時，只需要簡單地將其輸出的 <key,value> 對往 OutputCollector 中一丟即可，剩余的事框架自會幫你處理好。

代碼中 LongWritable, IntWritable, Text 均是 Hadoop 中實現的用于封裝 Java 數據類型的類，這些類都能夠被串行化從而便于在分布式環境中進行數據交換，你可以將它們分別視為 long, int, String 的替代品。Reporter 則可用于報告整個應用的運行進度，本例中未使用。

代碼清單1

public static class MapClass extends MapReduceBase

????????implements Mapper<LongWritable, Text, Text, IntWritable>{

????????private final static IntWritable one = new IntWritable(1);

????????private Text word = new Text(); ????????

????????public void map(LongWritable key, Text value,

????????}

????????OutputCollector<Text, IntWritable> output, Reporter reporter) throws IOException {

????????String line = value.toString();

????????StringTokenizer itr = new StringTokenizer(line);

????????while (itr.hasMoreTokens()) {}

?????word.set(itr.nextToken()); output.collect(word, one);

}
?

2.實現Reduce類

見代碼清單2。這個類實現 Reducer 接口中的 reduce 方法, 輸入參數中的 key, values 是由 Map 任務輸出的中間結果，values 是一個 Iterator, 遍歷這個 Iterator, 就可以得到屬于同一個 key 的所有 value. 此處，key 是一個單詞，value 是詞頻。只需要將所有的 value 相加，就可以得到這個單詞的總的出現次數。

代碼清單 2

public static class Reduce extends MapReduceBase

????????implements Reducer<Text, IntWritable, Text, IntWritable> {?

????????public void reduce(Text key, Iterator<IntWritable> values,

????????}

????????OutputCollector<Text, IntWritable> output, Reporter reporter) throws IOException {

????????int sum = 0; while (values.hasNext()) { }

????????output.collect(key, new IntWritable(sum));

????????sum += values.next().get();

}
?

3.運行Job

在 Hadoop 中一次計算任務稱之為一個 job, 可以通過一個 JobConf 對象設置如何運行這個 job。此處定義了輸出的 key 的類型是 Text, value 的類型是 IntWritable, 指定使用代碼清單1中實現的 MapClass 作為 Mapper 類,　使用代碼清單2中實現的 Reduce 作為 Reducer 類和 Combiner 類, 任務的輸入路徑和輸出路徑由命令行參數指定，這樣 job 運行時會處理輸入路徑下的所有文件，并將計算結果寫到輸出路徑下。

然后將 JobConf 對象作為參數，調用 JobClient 的 runJob, 開始執行這個計算任務。至于 main 方法中使用的 ToolRunner 是一個運行 MapReduce 任務的輔助工具類，依樣畫葫蘆用之即可。

代碼清單 3

public int run(String[] args) throws Exception {
?

JobConf conf = new JobConf(getConf(), WordCount.class);??

conf.setJobName("wordcount");?

conf.setOutputKeyClass(Text.class);?

conf.setOutputValueClass(IntWritable.class);?

conf.setMapperClass(MapClass.class);?

conf.setCombinerClass(Reduce.class);?

conf.setReducerClass(Reduce.class);?

conf.setInputPath(new Path(args[0]));?

conf.setOutputPath(new Path(args[1]));?

JobClient.runJob(conf); return 0;

}

public static void main(String[] args) throws Exception {

if(args.length != 2){ }

????????? int res = ToolRunner.run(new Configuration(), new WordCount(), args);?

????????? System.exit(res);

System.err.println("Usage: WordCount <input path> <output path>");??

System.exit(-1);

}

以上就是?WordCount?程序的全部細節，簡單到讓人吃驚，您都不敢相信就這么幾行代碼就可以分布式運行于大規模集群上，并行處理海量數據集。

4. 通過JobConf 定制計算任務

通過上文所述的 JobConf 對象，程序員可以設定各種參數，定制如何完成一個計算任務。這些參數很多情況下就是一個 java 接口，通過注入這些接口的特定實現，可以定義一個計算任務( job )的全部細節。了解這些參數及其缺省設置，您才能在編寫自己的并行計算程序時做到輕車熟路，游刃有余，明白哪些類是需要自己實現的，哪些類用 Hadoop 的缺省實現即可。表一是對 JobConf 對象中可以設置的一些重要參數的總結和說明，表中第一列中的參數在 JobConf 中均會有相應的 get/set 方法，對程序員來說，只有在表中第三列中的缺省值無法滿足您的需求時，才需要調用這些 set 方法，設定合適的參數值，實現自己的計算目的。針對表格中第一列中的接口，除了第三列的缺省實現之外，Hadoop 通常還會有一些其它的實現，我在表格第四列中列出了部分，您可以查閱 Hadoop 的 API 文檔或源代碼獲得更詳細的信息，在很多的情況下，您都不用實現自己的 Mapper 和 Reducer, 直接使用 Hadoop 自帶的一些實現即可。

表一 JobConf 常用可定制參數

參數	作用	缺省值	其它實現
inputFormat	將輸入的數據集切割成小數據集inputSplits, 每一個InputSplit將 由一個Mapper負責處理。此外inputFormat中還提供一個RecordReader的 實現, 將一個InputSplit解 析成《key,value》 對提供給 map 函數。	TextInputFormat(針 對文本文件，按行將文本文件切割成InputSplits, 并用LineRecordReader將InputSplit解 析成 《key,value》 對，key是行在文件中的位置，value是文件中的一行)	SequenceFileInputFormat
OutputFormat	提供一個 RecordWriter 的實現，負責輸出最終結果	TextOutputFormat(用 LineRecordWriter 將最終結果寫成純文件文件,每個《key,value》對一行，key 和 value 之間用 tab 分隔)	SequenceFileOutputFormat
OutputKeyClass	輸出的最終結果中 key 的類型	LongWritable	?
OutputValueClass	輸出的最終結果中 value 的類型	Text	?
MapperClass	Mapper 類，實現 map 函數，完成輸入的 《key,value》 到中間結果的映射	IdentityMapper(將 輸入的 《key,value》 原封不動的輸出為中間結果)	LongSumReducer, LogRegexMapper, InverseMapper
CombinerClass	實現 combine 函數，將中間結果中的重復 key 做合并	null(不對中間結果中的重復 key 做合并)	?
ReducerClass	Reducer 類，實現 reduce 函數，對中間結果做合并，形成最終結果	IdentityReducer(將 中間結果直接輸出為最終結果)	AccumulatingReducer, LongSumReducer
InputPath	設定 job 的輸入目錄, job 運行時會處理輸入目錄下的所有文件	null	?
OutputPath	設定 job 的輸出目錄，job 的最終結果會寫入輸出目錄下	null	?
MapOutputKeyClass	設定 map 函數輸出的中間結果中 key 的類型	如果用戶沒有設定的話，使用OutputKeyClass	?
MapOutputValueClass	設定 map 函數輸出的中間結果中	value 的類型 如果用戶沒有設定的話，使用 OutputValuesClass	?
OutputKeyComparator	對結果中的 key 進行排序時的使用的比較器	WritableComparable	?
PartitionerClass	對中間結果的 key 排序后，用此 Partition 函數將其劃分為R份,每份由一個 Reducer 負責處理。	HashPartitioner(使 用 Hash 函數做 partition)	KeyFieldBasedPartitioner PipesPartitioner

改進的?WordCount?程序

現在你對 Hadoop 并行程序的細節已經有了比較深入的了解，我們來把?WordCount?程序改進一下，目標: (1)原?WordCount?程序僅按空格切分單詞，導致各類標點符號與單詞混雜在一起，改進后的程序應該能夠正確的切出單詞,并且單詞不要區分大小寫。(2)在最終結果中，按單詞出 現頻率的降序進行排序。

1.修改 Mapper 類，實現目標(1)

實現很簡單，見代碼清單4中 的注釋。

代碼清單 4

public static class MapClass extends MapReduceBase?
?

????????implements Mapper<LongWritable, Text, Text, IntWritable> {?

????????????????private final static IntWritable one = new IntWritable(1);?

????????????????private Text word = new Text();?

????????????????private String pattern="[^\\w]"; //正則表達式，代表不是0-9, a-z, A-Z的所有其它字符

????????????????public void map(LongWritable key, Text value,OutputCollector<Text, IntWritable> output, Reporter reporter)

????????throws IOException {

????????????????String line = value.toString().toLowerCase(); //全部轉為小寫字母?

????????????????line = line.replaceAll(pattern, " "); //將非0-9, a-z, A-Z的字符替換為空格

????????????????StringTokenizer itr = new StringTokenizer(line);?

????????while (itr.hasMoreTokens()) {

????????????????????????word.set(itr.nextToken()); output.collect(word, one);

????????????????}

????????}

}
?

2.實現目標(2)

用一個并行計算任務顯然是無法同時完成單詞詞頻統計和排序的，這時我們可以利用 Hadoop 的任務管道能力，用上一個任務(詞頻統計)的輸出做為下一個任務(排序)的輸入，順序執行兩個并行計算任務。主要工作是修改代碼清單3中的 run 函數，在其中定義一個排序任務并運行之。

在 Hadoop 中要實現排序是很簡單的，因為在 MapReduce 的過程中，會把中間結果根據 key 排序并按 key 切成 R 份交給 R 個 Reduce 函數，而 Reduce 函數在處理中間結果之前也會有一個按 key 進行排序的過程，故 MapReduce 輸出的最終結果實際上已經按 key 排好序。詞頻統計任務輸出的 key 是單詞，value 是詞頻，為了實現按詞頻排序，我們指定使用 InverseMapper 類作為排序任務的 Mapper 類( sortJob.setMapperClass(InverseMapper.class );)，這個類的 map 函數簡單地將輸入的 key 和 value 互換后作為中間結果輸出，在本例中即是將詞頻作為 key,單詞作為 value 輸出, 這樣自然就能得到按詞頻排好序的最終結果。我們無需指定 Reduce 類，Hadoop 會使用缺省的 IdentityReducer 類，將中間結果原樣輸出。

還有一個問題需要解決: 排序任務中的 Key 的類型是 IntWritable, (sortJob.setOutputKeyClass(IntWritable.class)), Hadoop 默認對 IntWritable 按升序排序，而我們需要的是按降序排列。因此我們實現了一個 IntWritableDecreasingComparator 類,　并指定使用這個自定義的 Comparator 類對輸出結果中的 key (詞頻)進行排序：sortJob.setOutputKeyComparatorClass(IntWritableDecreasingComparator.class)

詳見代碼清單 5 及其中的注釋。 代碼清單 5

public int run(String[] args) throws Exception {

????????Path tempDir = new Path("wordcount-temp-" + Integer.toString(
????????new Random().nextInt(Integer.MAX_VALUE))); //定義一個臨時目錄?
????????JobConf conf = new JobConf(getConf(), WordCount.class);

try {
????????conf.setJobName("wordcount");

????????conf.setOutputKeyClass(Text.class);

conf.setOutputValueClass(IntWritable.class);

conf.setMapperClass(MapClass.class);

conf.setCombinerClass(Reduce.class);

conf.setReducerClass(Reduce.class);

conf.setInputPath(new Path(args[0]));

conf.setOutputPath(tempDir); //先將詞頻統計任務的輸出結果寫到臨時目
//錄中, 下一個排序任務以臨時目錄為輸入目錄。

conf.setOutputFormat(SequenceFileOutputFormat.class);

JobClient.runJob(conf);

JobConf sortJob = new JobConf(getConf(), WordCount.class);

sortJob.setJobName("sort"); sortJob.setInputPath(tempDir);

sortJob.setInputFormat(SequenceFileInputFormat.class);

sortJob.setMapperClass(InverseMapper.class);

sortJob.setNumReduceTasks(1); //將 Reducer 的個數限定為1, 最終輸出的結果
//文件就是一個。?
sortJob.setOutputPath(new Path(args[1]));
sortJob.setOutputKeyClass(IntWritable.class);

sortJob.setOutputValueClass(Text.class);

sortJob.setOutputKeyComparatorClass(IntWritableDecreasingComparator.class);

JobClient.runJob(sortJob);?} finally {
????????????????FileSystem.get(conf).delete(tempDir); 　//刪除臨時目錄

????????}
????????????????return 0;
}?
?

private static class IntWritableDecreasingComparator extends IntWritable.Comparator {

????????public int compare(WritableComparable a, WritableComparable b) {

return -super.compare(a, b);

???????? }public int compare(byte[] b1, int s1, int l1, byte[] b2, int s2, int l2) {

??????????????? ?? return -super.compare(b1, s1, l1, b2, s2, l2);

????????? }

}

在Eclipse 環境下進行開發和調試

在 Eclipse 環境下可以方便地進行 Hadoop 并行程序的開發和調試。推薦使用 IBM MapReduce Tools for Eclipse, 使用這個 Eclipse plugin 可以簡化開發和部署 Hadoop 并行程序的過程。基于這個 plugin, 可以在 Eclipse 中創建一個 Hadoop MapReduce 應用程序，并且提供了一些基于 MapReduce 框架的類開發的向導，可以打包成 JAR 文件，部署一個 Hadoop MapReduce 應用程序到一個 Hadoop 服務器(本地和遠程均可)，可以通過一個專門的視圖 ( perspective ) 查看 Hadoop 服務器、Hadoop 分布式文件系統（ DFS ）和當前運行的任務的狀態。

可在 IBM alphaWorks 網站下載這個 MapReduce Tool, 或在本文的下載清單中下載。將下載后的壓縮包解壓到你 Eclipse 安裝目錄，重新啟動 Eclipse 即可使用了。

設置 Hadoop 主目錄

點擊 Eclipse 主菜單上 Windows->Preferences, 然后在左側選擇 HadoopHome Directory,設定你的 Hadoop 主目錄,如圖一所示：

圖 1

創立一個 MapReduce Project

點擊 Eclipse 主菜單上 File->New->Project, 在彈出的對話框中選擇 MapReduce Project, 輸入 project name 如 wordcount,　然后點擊 Finish 即可。,如圖 2 所示：

圖 2

此后，你就可以象一個普通 的 Eclipse Java project 那樣，添加入 Java 類，比如你可以定義一個?WordCount?類，然后將本文代碼清單1,2,3中的代碼寫到此類中，添加入必要的 import 語句 ( Eclipse 快捷鍵 ctrl+shift+o 可以幫你)，即可形成一個完整的 wordcount 程序。

在我們這個簡單的 wordcount 程序中，我們把全部的內容都放在一個?WordCount?類中。實際上 IBM MapReduce tools 還提供了幾個實用的向導 ( wizard ) 工具，幫你創建單獨的 Mapper 類，Reducer 類，MapReduce Driver 類(就是代碼清單3中那部分內容)，在編寫比較復雜的 MapReduce 程序時，將這些類獨立出來是非常有必要的，也有利于在不同的計算任務中重用你編寫的各種 Mapper 類和 Reducer 類。

在 Eclipse 中運行

如圖三所示，設定程序的運行參數:輸入目錄和輸出目錄之后，你就可以在 Eclipse 中運行 wordcount 程序了，當然，你也可以設定斷點，調試程序。

圖 3

結束語

到目前為止，我們已經介紹了 MapReduce 計算模型，分布式文件系統 HDFS，分布式并行計算等的基本原理, 如何安裝和部署單機 Hadoop 環境，實際編寫了一個 Hadoop 并行計算程序，并了解了一些重要的編程細節，了解了如何使用 IBM MapReduce Tools 在 Eclipse 環境中編譯，運行和調試你的 Hadoop 并行計算程序。但一個 Hadoop 并行計算程序，只有部署運行在分布式集群環境中，才能發揮其真正的優勢，在這篇系列文章的第 3 部分中，你將了解到如何部署你的分布式 Hadoop 環境，如何利用 IBM MapReduce Tools 將你的程序部署到分布式環境中運行等內容。

總結

以上是生活随笔為你收集整理的用Hadoop进行分布式并行编程的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。