前言

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

回頁首

分析 WordCount 程序

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

1.實(shí)現(xiàn)Map類

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

代碼中 LongWritable, IntWritable, Text 均是 Hadoop 中實(shí)現(xiàn)的用于封裝 Java 數(shù)據(jù)類型的類，這些類都能夠被串行化從而便于在分布式環(huán)境中進(jìn)行數(shù)據(jù)交換，你可以將它們分別視為 long, int, String 的替代品。Reporter 則可用于報(bào)告整個(gè)應(yīng)用的運(yùn)行進(jìn)度，本例中未使用。

                            
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.實(shí)現(xiàn) Reduce 類

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

                            
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()) {
        sum += values.next().get();
      }
      output.collect(key, new IntWritable(sum));
    }
  }

          

3.運(yùn)行 Job

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

然后將 JobConf 對象作為參數(shù)，調(diào)用 JobClient 的 runJob, 開始執(zhí)行這個(gè)計(jì)算任務(wù)。至于 main 方法中使用的 ToolRunner 是一個(gè)運(yùn)行 MapReduce 任務(wù)的輔助工具類，依樣畫葫蘆用之即可。

                            
 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){
      System.err.println("Usage: WordCount <input path> <output path>");
      System.exit(-1);
    }
    int res = ToolRunner.run(new Configuration(), new WordCount(), args);
    System.exit(res);
  }
}

          

以上就是 WordCount 程序的全部細(xì)節(jié)，簡單到讓人吃驚，您都不敢相信就這么幾行代碼就可以分布式運(yùn)行于大規(guī)模集群上，并行處理海量數(shù)據(jù)集。

4. 通過 JobConf 定制計(jì)算任務(wù)

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

回頁首

改進(jìn)的 WordCount 程序

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

1.修改 Mapper 類，實(shí)現(xiàn)目標(biāo)(1)

實(shí)現(xiàn)很簡單，見代碼清單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]"; //正則表達(dá)式，代表不是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(); //全部轉(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.實(shí)現(xiàn)目標(biāo)(2)

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

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

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

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

                            
public int run(String[] args) throws Exception {
        Path tempDir = new Path("wordcount-temp-" + Integer.toString(
            new Random().nextInt(Integer.MAX_VALUE))); //定義一個(gè)臨時(shí)目錄

        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); //先將詞頻統(tǒng)計(jì)任務(wù)的輸出結(jié)果寫到臨時(shí)目
                                         //錄中, 下一個(gè)排序任務(wù)以臨時(shí)目錄為輸入目錄。
            
            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 的個(gè)數(shù)限定為1, 最終輸出的結(jié)果
　　　　　　　　　　                      //文件就是一個(gè)。
            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);　//刪除臨時(shí)目錄
        }
    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 環(huán)境下進(jìn)行開發(fā)和調(diào)試

在 Eclipse 環(huán)境下可以方便地進(jìn)行 Hadoop 并行程序的開發(fā)和調(diào)試。推薦使用 IBM MapReduce Tools for Eclipse, 使用這個(gè) Eclipse plugin 可以簡化開發(fā)和部署 Hadoop 并行程序的過程。基于這個(gè) plugin, 可以在 Eclipse 中創(chuàng)建一個(gè) Hadoop MapReduce 應(yīng)用程序，并且提供了一些基于 MapReduce 框架的類開發(fā)的向?qū)В梢源虬?JAR 文件，部署一個(gè) Hadoop MapReduce 應(yīng)用程序到一個(gè) Hadoop 服務(wù)器(本地和遠(yuǎn)程均可)，可以通過一個(gè)專門的視圖 ( perspective ) 查看 Hadoop 服務(wù)器、Hadoop 分布式文件系統(tǒng)（ DFS ）和當(dāng)前運(yùn)行的任務(wù)的狀態(tài)。

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

設(shè)置 Hadoop 主目錄

點(diǎn)擊 Eclipse 主菜單上 Windows->Preferences, 然后在左側(cè)選擇 Hadoop Home Directory,設(shè)定你的 Hadoop 主目錄,如圖一所示：

創(chuàng)立一個(gè) MapReduce Project

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

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

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

在 Eclipse 中運(yùn)行

如圖三所示，設(shè)定程序的運(yùn)行參數(shù):輸入目錄和輸出目錄之后，你就可以在 Eclipse 中運(yùn)行 wordcount 程序了，當(dāng)然，你也可以設(shè)定斷點(diǎn)，調(diào)試程序。

用 Hadoop 進(jìn)行分布式并行編程, 第 2 部分