hadoop進行分布式并行編程第二部分

1、用 Hadoop 進行分布式并行編程, 第 2 部分程序?qū)嵗c分析曹 羽中 (caoyuz), 軟件工程師, IBM中國開發(fā)中心簡介： Hadoop 是一個實現(xiàn)了 MapReduce 計算模型的開源分布式并行編程框架，借助于 Hadoop, 程序員可以輕松地編寫分布式并行程序，將其運行于計算機集群上，完成海量數(shù)據(jù)的計算。在本文中，詳細介紹了如何針對一個具體的并行計算任務(wù)，基于 Hadoop 編寫程序，如何使用 IBM MapReduce Tools 在 Eclipse 環(huán)境中編譯并運行 Hadoop 程序。本文的標(biāo)簽：  eclipse下開發(fā)hadoop

2、, mapreduce標(biāo)記本文！發(fā)布日期： 2008 年 5 月 22 日 級別： 初級 訪問情況 ： 11069 次瀏覽 評論： 1 (查看 | 添加評論 - 登錄) 平均分 (12個評分)為本文評分前言在上一篇文章：“用 Hadoop 進行分布式并行編程 第一部分 基本概念與安裝部署”中，介紹了 MapReduce 計算模型，分布式文件系統(tǒng) HDFS，分布式并行計算等的基本原理, 并且詳細介紹了如何安裝 Hadoop，如何運行基于 Hadoop 的并行程序。在本

3、文中，將針對一個具體的計算任務(wù)，介紹如何基于 Hadoop 編寫并行程序，如何使用 IBM 開發(fā)的 Hadoop Eclipse plugin 在 Eclipse 環(huán)境中編譯并運行程序?；仨撌追治?WordCount 程序我們先來看看 Hadoop 自帶的示例程序 WordCount，這個程序用于統(tǒng)計一批文本文件中單詞出現(xiàn)的頻率，完整的代碼可在下載的 Hadoop 安裝包中得到(在 src/examples 目錄中)。1.實現(xiàn)Map類見代碼清單1。這個類實現(xiàn) Mapper 接口中的 map 方法，輸入?yún)?shù)中的 value 是文本文件中的一行，利用 StringTokenizer 將這個字符串拆

4、成單詞，然后將輸出結(jié)果 <單詞,1> 寫入到 org.apache.hadoop.mapred.OutputCollector 中。OutputCollector 由 Hadoop 框架提供, 負責(zé)收集 Mapper 和 Reducer 的輸出數(shù)據(jù)，實現(xiàn) map 函數(shù)和 reduce 函數(shù)時，只需要簡單地將其輸出的 <key,value> 對往 OutputCollector 中一丟即可，剩余的事框架自會幫你處理好。代碼中 LongWritable, IntWritable, Text 均是 Hadoop 中實現(xiàn)的用于封裝 Java 數(shù)據(jù)類型的類，這些類都能夠被串行化從

5、而便于在分布式環(huán)境中進行數(shù)據(jù)交換，你可以將它們分別視為 long, int, String 的替代品。Reporter 則可用于報告整個應(yīng)用的運行進度，本例中未使用。代碼清單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

6、 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.實現(xiàn) Reduce

7、類見代碼清單 2。這個類實現(xiàn) Reducer 接口中的 reduce 方法, 輸入?yún)?shù)中的 key, values 是由 Map 任務(wù)輸出的中間結(jié)果，values 是一個 Iterator, 遍歷這個 Iterator, 就可以得到屬于同一個 key 的所有 value. 此處，key 是一個單詞，value 是詞頻。只需要將所有的 value 相加，就可以得到這個單詞的總的出現(xiàn)次數(shù)。代碼清單 2 public static class Reduce extends MapReduceBase implements Reducer<Text, IntWritable, Text, Int

8、Writable> public void reduce(Text key, Iterator<IntWritable> values, OutputCollector<Text, IntWritable> output, Reporter reporter) throws IOException int sum = 0; while (values.hasNext() sum += values.next().get(); output.collect(key, new IntWritable(sum); 3.運行 Job在 Hadoop 中一次計算任務(wù)稱之為一

9、個 job, 可以通過一個 JobConf 對象設(shè)置如何運行這個 job。此處定義了輸出的 key 的類型是 Text, value 的類型是 IntWritable, 指定使用代碼清單1中實現(xiàn)的 MapClass 作為 Mapper 類,使用代碼清單2中實現(xiàn)的 Reduce 作為 Reducer 類和 Combiner 類, 任務(wù)的輸入路徑和輸出路徑由命令行參數(shù)指定，這樣 job 運行時會處理輸入路徑下的所有文件，并將計算結(jié)果寫到輸出路徑下。然后將 JobConf 對象作為參數(shù)，調(diào)用 JobClient 的 runJob, 開始執(zhí)行這個計算任務(wù)。至于 main 方法中使用的 ToolRunn

10、er 是一個運行 MapReduce 任務(wù)的輔助工具類，依樣畫葫蘆用之即可。代碼清單 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.clas

11、s); conf.setCombinerClass(Reduce.class); conf.setReducerClass(Reduce.class); conf.setInputPath(new Path(args0); conf.setOutputPath(new Path(args1); JobClient.runJob(conf); return 0; public static void main(String args) throws Exception if(args.length != 2) System.err.println("Usage: WordCount &

12、lt;input path> <output path>"); System.exit(-1); int res = ToolRunner.run(new Configuration(), new WordCount(), args); System.exit(res); 以上就是 WordCount 程序的全部細節(jié)，簡單到讓人吃驚，您都不敢相信就這么幾行代碼就可以分布式運行于大規(guī)模集群上，并行處理海量數(shù)據(jù)集。4. 通過 JobConf 定制計算任務(wù)通過上文所述的 JobConf 對象，程序員可以設(shè)定各種參數(shù)，定制如何完成一個計算任務(wù)。這些參數(shù)很多情況下就是一個 j

13、ava 接口，通過注入這些接口的特定實現(xiàn)，可以定義一個計算任務(wù)( job )的全部細節(jié)。了解這些參數(shù)及其缺省設(shè)置，您才能在編寫自己的并行計算程序時做到輕車熟路，游刃有余，明白哪些類是需要自己實現(xiàn)的，哪些類用 Hadoop 的缺省實現(xiàn)即可。表一是對 JobConf 對象中可以設(shè)置的一些重要參數(shù)的總結(jié)和說明，表中第一列中的參數(shù)在 JobConf 中均會有相應(yīng)的 get/set 方法，對程序員來說，只有在表中第三列中的缺省值無法滿足您的需求時，才需要調(diào)用這些 set 方法，設(shè)定合適的參數(shù)值，實現(xiàn)自己的計算目的。針對表格中第一列中的接口，除了第三列的缺省實現(xiàn)之外，Hadoop 通常還會有一些其它的實現(xiàn)

14、，我在表格第四列中列出了部分，您可以查閱 Hadoop 的 API 文檔或源代碼獲得更詳細的信息，在很多的情況下，您都不用實現(xiàn)自己的 Mapper 和 Reducer, 直接使用 Hadoop 自帶的一些實現(xiàn)即可。表一 JobConf 常用可定制參數(shù)參數(shù)作用缺省值其它實現(xiàn)InputFormat將輸入的數(shù)據(jù)集切割成小數(shù)據(jù)集 InputSplits, 每一個 InputSplit 將由一個 Mapper 負責(zé)處理。此外 InputFormat 中還提供一個 RecordReader 的實現(xiàn), 將一個 InputSplit 解析成 <key,value> 對提供給 map 函數(shù)。Text

15、InputFormat(針對文本文件，按行將文本文件切割成 InputSplits, 并用 LineRecordReader 將 InputSplit 解析成 <key,value> 對，key 是行在文件中的位置，value 是文件中的一行)SequenceFileInputFormatOutputFormat提供一個 RecordWriter 的實現(xiàn)，負責(zé)輸出最終結(jié)果TextOutputFormat(用 LineRecordWriter 將最終結(jié)果寫成純文件文件,每個 <key,value> 對一行，key 和 value 之間用 tab 分隔)SequenceFi

16、leOutputFormatOutputKeyClass輸出的最終結(jié)果中 key 的類型LongWritableOutputValueClass輸出的最終結(jié)果中 value 的類型TextMapperClassMapper 類，實現(xiàn) map 函數(shù)，完成輸入的 <key,value> 到中間結(jié)果的映射IdentityMapper(將輸入的 <key,value> 原封不動的輸出為中間結(jié)果)LongSumReducer,LogRegexMapper,InverseMapperCombinerClass實現(xiàn) combine 函數(shù)，將中間結(jié)果中的重復(fù) key 做合并null(不

17、對中間結(jié)果中的重復(fù) key 做合并)ReducerClassReducer 類，實現(xiàn) reduce 函數(shù)，對中間結(jié)果做合并，形成最終結(jié)果IdentityReducer(將中間結(jié)果直接輸出為最終結(jié)果)AccumulatingReducer, LongSumReducerInputPath設(shè)定 job 的輸入目錄, job 運行時會處理輸入目錄下的所有文件nullOutputPath設(shè)定 job 的輸出目錄，job 的最終結(jié)果會寫入輸出目錄下nullMapOutputKeyClass設(shè)定 map 函數(shù)輸出的中間結(jié)果中 key 的類型如果用戶沒有設(shè)定的話，使用 OutputKeyClassMapOu

18、tputValueClass設(shè)定 map 函數(shù)輸出的中間結(jié)果中 value 的類型如果用戶沒有設(shè)定的話，使用 OutputValuesClassOutputKeyComparator對結(jié)果中的 key 進行排序時的使用的比較器WritableComparablePartitionerClass對中間結(jié)果的 key 排序后，用此 Partition 函數(shù)將其劃分為R份,每份由一個 Reducer 負責(zé)處理。HashPartitioner(使用 Hash 函數(shù)做 partition)KeyFieldBasedPartitioner PipesPartitioner回頁首改進的 WordCount

19、程序現(xiàn)在你對 Hadoop 并行程序的細節(jié)已經(jīng)有了比較深入的了解，我們來把 WordCount 程序改進一下，目標(biāo): (1)原 WordCount 程序僅按空格切分單詞，導(dǎo)致各類標(biāo)點符號與單詞混雜在一起，改進后的程序應(yīng)該能夠正確的切出單詞,并且單詞不要區(qū)分大小寫。(2)在最終結(jié)果中，按單詞出現(xiàn)頻率的降序進行排序。1.修改 Mapper 類，實現(xiàn)目標(biāo)(1)實現(xiàn)很簡單，見代碼清單4中的注釋。代碼清單 4 public static class MapClass extends MapReduceBase implements Mapper<LongWritable, Text, Text,

20、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 IOE

21、xception String line = value.toString().toLowerCase(); /全部轉(zhuǎn)為小寫字母 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.實現(xiàn)目標(biāo)(2)用一個并行計算任務(wù)顯然是無法同時完成單詞詞頻統(tǒng)計

22、和排序的，這時我們可以利用 Hadoop 的任務(wù)管道能力，用上一個任務(wù)(詞頻統(tǒng)計)的輸出做為下一個任務(wù)(排序)的輸入，順序執(zhí)行兩個并行計算任務(wù)。主要工作是修改代碼清單3中的 run 函數(shù)，在其中定義一個排序任務(wù)并運行之。在 Hadoop 中要實現(xiàn)排序是很簡單的，因為在 MapReduce 的過程中，會把中間結(jié)果根據(jù) key 排序并按 key 切成 R 份交給 R 個 Reduce 函數(shù)，而 Reduce 函數(shù)在處理中間結(jié)果之前也會有一個按 key 進行排序的過程，故 MapReduce 輸出的最終結(jié)果實際上已經(jīng)按 key 排好序。詞頻統(tǒng)計任務(wù)輸出的 key 是單詞，value 是詞頻，為了實現(xiàn)

23、按詞頻排序，我們指定使用 InverseMapper 類作為排序任務(wù)的 Mapper 類( sortJob.setMapperClass(InverseMapper.class );)，這個類的 map 函數(shù)簡單地將輸入的 key 和 value 互換后作為中間結(jié)果輸出，在本例中即是將詞頻作為 key,單詞作為 value 輸出, 這樣自然就能得到按詞頻排好序的最終結(jié)果。我們無需指定 Reduce 類，Hadoop 會使用缺省的 IdentityReducer 類，將中間結(jié)果原樣輸出。還有一個問題需要解決: 排序任務(wù)中的 Key 的類型是 IntWritable, (sortJob.setOu

24、tputKeyClass(IntWritable.class), Hadoop 默認對 IntWritable 按升序排序，而我們需要的是按降序排列。因此我們實現(xiàn)了一個 IntWritableDecreasingComparator 類,并指定使用這個自定義的 Comparator 類對輸出結(jié)果中的 key (詞頻)進行排序：sortJob.setOutputKeyComparatorClass(IntWritableDecreasingComparator.class)詳見代碼清單 5 及其中的注釋。代碼清單 5 public int run(String args) throws Exce

25、ption 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(IntWritabl

26、e.class); conf.setMapperClass(MapClass.class); conf.setCombinerClass(Reduce.class); conf.setReducerClass(Reduce.class); conf.setInputPath(new Path(args0); conf.setOutputPath(tempDir); /先將詞頻統(tǒng)計任務(wù)的輸出結(jié)果寫到臨時目 /錄中, 下一個排序任務(wù)以臨時目錄為輸入目錄。 conf.setOutputFormat(SequenceFileOutputFormat.class); JobClient.runJob(c

27、onf); 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 的個數(shù)限定為1, 最終輸出的結(jié)果 /文件就是一個。

28、 sortJob.setOutputPath(new Path(args1); sortJob.setOutputKeyClass(IntWritable.class); sortJob.setOutputValueClass(Text.class); sortJob.setOutputKeyComparatorClass(IntWritableDecreasingComparator.class); JobClient.runJob(sortJob); finally FileSystem.get(conf).delete(tempDir);/刪除臨時目錄 return 0; private

29、 static class IntWritableDecreasingComparator extends IntWritable.Comparator public int compare(WritableComparable a, WritableComparable b) return -pare(a, b); public int compare(byte b1, int s1, int l1, byte b2, int s2, int l2) return -pare(b1, s1, l1, b2, s2, l2); 回頁首在 Eclipse 環(huán)境下進行開發(fā)和調(diào)試在 Eclipse

30、環(huán)境下可以方便地進行 Hadoop 并行程序的開發(fā)和調(diào)試。推薦使用 IBM MapReduce Tools for Eclipse, 使用這個 Eclipse plugin 可以簡化開發(fā)和部署 Hadoop 并行程序的過程。基于這個 plugin, 可以在 Eclipse 中創(chuàng)建一個 Hadoop MapReduce 應(yīng)用程序，并且提供了一些基于 MapReduce 框架的類開發(fā)的向?qū)?，可以打包?JAR 文件，部署一個 Hadoop MapReduce 應(yīng)用程序到一個 Hadoop 服務(wù)器(本地和遠程均可)，可以通過一個專門的視圖 ( perspective ) 查看 Hadoop 服務(wù)器、

31、Hadoop 分布式文件系統(tǒng)（ DFS ）和當(dāng)前運行的任務(wù)的狀態(tài)。可在 IBM alphaWorks 網(wǎng)站下載這個 MapReduce Tool, 或在本文的下載清單中下載。將下載后的壓縮包解壓到你 Eclipse 安裝目錄，重新啟動 Eclipse 即可使用了。設(shè)置 Hadoop 主目錄點擊 Eclipse 主菜單上 Windows->Preferences, 然后在左側(cè)選擇 Hadoop Home Directory,設(shè)定你的 Hadoop 主目錄,如圖一所示：圖 1 創(chuàng)立一個 MapReduce Project點擊 Eclipse 主菜單上 File->N

32、ew->Project, 在彈出的對話框中選擇 MapReduce Project, 輸入 project name 如 wordcount,然后點擊 Finish 即可。,如圖 2 所示：圖 2 此后，你就可以象一個普通的 Eclipse Java project 那樣，添加入 Java 類，比如你可以定義一個 WordCount 類，然后將本文代碼清單1,2,3中的代碼寫到此類中，添加入必要的 import 語句 ( Eclipse 快捷鍵 ctrl+shift+o 可以幫你)，即可形成一個完整的 wordcount 程序。在我們這個簡單的 wordcount 程序中，我們把全部的內(nèi)容都放在一個 WordCount 類中。實際上 IBM MapReduce tools 還提供了幾個實用的向?qū)?( wizard ) 工具，幫你創(chuàng)建單獨的 Mapper 類，Reducer 類，MapReduce Driver 類(就是代碼清單3中那部分內(nèi)容)，在編寫比較復(fù)雜的 MapReduce 程序時，將這些類獨立出來是非常有必要的，也有利于在不同的計算任務(wù)中重用你編寫的各種 Mapper 類和 Reducer 類。在 Eclip