hadoop word count文章阅读笔记

我们先来看看 Hadoop 自带的示例程序 WordCount，这个程序用于统计一批文本文件中单词出现的频率，完整的代码可在下载的 Hadoop 安装包中得到(在 src/examples 目录中)。

见代码清单1。这个类实现 Mapper 接口中的 map 方法，输入参数中的 value 是文本文件中的一行，利用 StringTokenizer 将这个字符串拆成单词，然后将输出结果 <单词,1> 写入到 org.apache.hadoop.mapred.OutputCollector 中。OutputCollector 由 Hadoop 框架提供, 负责收集 Mapper 和 Reducer 的输出数据，实现 map 函数和 reduce 函数时，只需要简单地将其输出的 对往 OutputCollector 中一丢即可，剩余的事框架自会帮你处理好。

代码中 LongWritable, IntWritable, Text 均是 Hadoop 中实现的用于封装 Java 数据类型的类，这些类都能够被串行化从而便于在分布式环境中进行数据交换，你可以将它们分别视为 long, int, String 的替代品。Reporter 则可用于报告整个应用的运行进度，本例中未使用。

                
public static class MapClass extends MapReduceBase
    implements Mapper{
    
    private final static IntWritable one = new IntWritable(1);
    private Text word = new Text();
    
    public void map(LongWritable key, Text value, 
                    OutputCollector 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。这个类实现 Reducer 接口中的 reduce 方法, 输入参数中的 key, values 是由 Map 任务输出的中间结果，values 是一个 Iterator, 遍历这个 Iterator, 就可以得到属于同一个 key 的所有 value. 此处，key 是一个单词，value 是词频。只需要将所有的 value 相加，就可以得到这个单词的总的出现次数。

                
public static class Reduce extends MapReduceBase
    implements Reducer {
    
    public void reduce(Text key, Iterator values,
                       OutputCollector output, 
                       Reporter reporter) throws IOException {
      int sum = 0;
      while (values.hasNext()) {
        sum += values.next().get();
      }
      output.collect(key, new IntWritable(sum));
    }
  }

在 Hadoop 中一次计算任务称之为一个 job, 可以通过一个 JobConf 对象设置如何运行这个 job。此处定义了输出的 key 的类型是 Text, value 的类型是 IntWritable, 指定使用代码清单1中实现的 MapClass 作为 Mapper 类,　使用代码清单2中实现的 Reduce 作为 Reducer 类和 Combiner 类, 任务的输入路径和输出路径由命令行参数指定，这样 job 运行时会处理输入路径下的所有文件，并将计算结果写到输出路径下。

然后将 JobConf 对象作为参数，调用 JobClient 的 runJob, 开始执行这个计算任务。至于 main 方法中使用的 ToolRunner 是一个运行 MapReduce 任务的辅助工具类，依样画葫芦用之即可。

                
 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  
");
      System.exit(-1);
    }
    int res = ToolRunner.run(new Configuration(), new WordCount(), args);
    System.exit(res);
  }
}

以上就是 WordCount 程序的全部细节，简单到让人吃惊，您都不敢相信就这么几行代码就可以分布式运行于大规模集群上，并行处理海量数据集。

通过上文所述的 JobConf 对象，程序员可以设定各种参数，定制如何完成一个计算任务。这些参数很多情况下就是一个 java 接口，通过注入这些接口的特定实现，可以定义一个计算任务( job )的全部细节。了解这些参数及其缺省设置，您才能在编写自己的并行计算程序时做到轻车熟路，游刃有余，明白哪些类是需要自己实现的，哪些类用 Hadoop 的缺省实现即可。表一是对 JobConf 对象中可以设置的一些重要参数的总结和说明，表中第一列中的参数在 JobConf 中均会有相应的 get/set 方法，对程序员来说，只有在表中第三列中的缺省值无法满足您的需求时，才需要调用这些 set 方法，设定合适的参数值，实现自己的计算目的。针对表格中第一列中的接口，除了第三列的缺省实现之外，Hadoop 通常还会有一些其它的实现，我在表格第四列中列出了部分，您可以查阅 Hadoop 的 API 文档或源代码获得更详细的信息，在很多的情况下，您都不用实现自己的 Mapper 和 Reducer, 直接使用 Hadoop 自带的一些实现即可。

/*这个是很重要的表格，明白了这几个类的使用之后，要了解上面的wordcount程序就不难了*/

参数	作用	缺省值	其它实现
InputFormat	将 输入的数据集切割成小数据集 InputSplits, 每一个 InputSplit 将由一个 Mapper 负责处理。此外 InputFormat 中还提供一个 RecordReader 的实现, 将一个 InputSplit 解析成 对提供给 map 函数。	TextInputFormat (针对文本文件，按行将文本文件切割成 InputSplits, 并用 LineRecordReader 将 InputSplit 解析成 对，key 是行在文件中的位置，value 是文件中的一行)	SequenceFileInputFormat
OutputFormat	提供一个 RecordWriter 的实现，负责输出最终结果	TextOutputFormat (用 LineRecordWriter 将最终结果写成纯文件文件,每个 对一行，key 和 value 之间用 tab 分隔)	SequenceFileOutputFormat
OutputKeyClass	输出的最终结果中 key 的类型	LongWritable
OutputValueClass	输出的最终结果中 value 的类型	Text
MapperClass	Mapper 类，实现 map 函数，完成输入的 到中间结果的映射	IdentityMapper (将输入的 原封不动的输出为中间结果)	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

回页首

现在你对 Hadoop 并行程序的细节已经有了比较深入的了解，我们来把 WordCount 程序改进一下，目标: (1)原 WordCount 程序仅按空格切分单词，导致各类标点符号与单词混杂在一起，改进后的程序应该能够正确的切出单词,并且单词不要区分大小写。(2)在最终结果中，按单词出 现频率的降序进行排序。

实现很简单，见代码清单4中的注释。

                
public static class MapClass extends MapReduceBase
    implements Mapper {
    
    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 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);
      }
    }
  }

用 一个并行计算任务显然是无法同时完成单词词频统计和排序的，这时我们可以利用 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 及其中的注释。

                
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);

            /*InverseMapper由hadoop库提供，作用是实现map()之后的数据对的key和value交换*/
            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 环境下可以方便地进行 Hadoop 并行程序的开发和调试。推荐使用 IBM MapReduce Tools for Eclipse, 使用这个 Eclipse plugin 可以简化开发和部署 Hadoop 并行程序的过程。基于这个 plugin, 可以在 Eclipse 中创建一个 Hadoop MapReduce 应用程序，并且提供了一些基于 MapReduce 框架的类开发的向导，可以打包成 JAR 文件，部署一个 Hadoop MapReduce 应用程序到一个 Hadoop 服务器(本地和远程均可)，可以通过一个专门的视图 ( perspective ) 查看 Hadoop 服务器、Hadoop 分布式文件系统（ DFS ）和当前运行的任务的状态。

可在 IBM alphaWorks 网站下载这个 , 或在本文的下载清单中下载。将下载后的压缩包解压到你 Eclipse 安装目录，重新启动 Eclipse 即可使用了。

点击 Eclipse 主菜单上 Windows->Preferences, 然后在左侧选择 Hadoop Home Directory,设定你的 Hadoop 主目录,如图一所示：

点击 Eclipse 主菜单上 File->New->Project, 在弹出的对话框中选择 MapReduce Project, 输入 project name 如 wordcount,　然后点击 Finish 即可。,如图 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 中运行 wordcount 程序了，当然，你也可以设定断点，调试程序。

回页首

到 目前为止，我们已经介绍了 MapReduce 计算模型，分布式文件系统 HDFS，分布式并行计算等的基本原理, 如何安装和部署单机 Hadoop 环境，实际编写了一个 Hadoop 并行计算程序，并了解了一些重要的编程细节，了解了如何使用 IBM MapReduce Tools 在 Eclipse 环境中编译，运行和调试你的 Hadoop 并行计算程序。但一个 Hadoop 并行计算程序，只有部署运行在分布式集群环境中，才能发挥其真正的优势，在这篇系列文章的第 3 部分中，你将了解到如何部署你的分布式 Hadoop 环境，如何利用 IBM MapReduce Tools 将你的程序部署到分布式环境中运行等内容。

阅读总结：

mapreduce的大致工作流程：

1.InputFormat类通过InputSplit将输入数据切分成多个块(这些类需要自己设置，或使用默认泪，参见上面的表格)，并设置输入数据的key和value.

2.每个块的数据提供给一个map程序处理，生成按key值排序的中间结果。

3.如果设置了Combiner类，那么同一个key的中间数据会被合并成一个key对应多个value的形式。

4.最后reduce程序读取中间结果，进行必要的归类和总结。

在wordcount这个例子中，中间结果会根据key值进行combiner，也就是一个单词可能对应多个value(其实就是计数的1)，然后在reduce中对每个key，计算它的多个value的和，输出到保存结果的文件。

Hadoop源代码分析（MapReduce概论）

大家都熟悉文件系统，在对HDFS进行分析前，我们并没有花很多的时间去介绍HDFS的背景，毕竟大家对文件系统的还是有一定的理解的，而且也有很好的文档。在分析Hadoop的MapReduce部分前，我们还是先了解系统是如何工作的，然后再进入我们的分析部分。下面的图来自http://horicky.blogspot.com/2008/11/hadoop-mapreduce-implementation.html，是我看到的讲MapReduce最好的图。

 

以Hadoop带的wordcount为例子（下面是启动行）：

hadoop jar hadoop-0.19.0-examples.jar wordcount /usr/input /usr/output

用户提交一个任务以后，该任务由JobTracker协调，先执行Map阶段（图中M1，M2和M3），然后执行Reduce阶段（图中R1和R2）。Map阶段和Reduce阶段动作都受监控，并运行在独立于TaskTracker的Java虚拟机中。

我们的输入和输出都是HDFS 上的目录（如上图所示）。输入由InputFormat接口描述，它的实现如ASCII文件，JDBC数据库等，分别处理对于的数据源，并提供了数据的一 些特征。通过InputFormat实现，可以获取InputSplit接口的实现，这个实现用于对数据进行划分（图中的splite1到 splite5，就是划分以后的结果），同时从InputFormat也可以获取RecordReader接口的实现，并从输入中生 成对。有了，就可以开始做map操作了。

map操作通过context.collect（最终通过OutputCollector. collect）将结果写到context中。当Mapper的输出被收集后，它们会被Partitioner类以指定的方式区分地写出到输出文件里。我们可以为Mapper提供Combiner，在Mapper输出它的时，键值对不会被马上写到输出里，他们会被收集在list里（一个key值一个list），当写入一定数量的键值对时，这部分缓冲会被Combiner中进行合并，然后再输出到Partitioner中（图中M1的黄颜色部分对应着Combiner和Partitioner）。

Map的动作做完以后，进入Reduce阶段。这个阶段分3个步骤：混洗（Shuffle），排序（sort）和reduce。

混洗阶段，Hadoop 的MapReduce框架会根据Map结果中的key，将相关的结果传输到某一个Reducer上（多个Mapper产生的同一个key的中间结果分布在 不同的机器上，这一步结束后，他们传输都到了处理这个key的Reducer的机器上）。这个步骤中的文件传输使用了HTTP协议。

排序和混洗是一块进行的，这个阶段将来自不同Mapper具有相同key值的对合并到一起。

Reduce阶段，上面通过Shuffle和sort后得到的会送到Reducer. reduce方法中处理，输出的结果通过OutputFormat，输出到DFS中。

我的理解：

1.文件被分片后传给M模块。

2.M模块用InputFormat类对输入文件进行Splite操作，生成对放到内存中，如果指明了Combiner类，那么对进行缓存，对相同的K的键值对进行合并，形成的形式。如果没有指明Combiner类，默认是空操作。

3.每个M模块调用Partition类(默认是HashPartiton)，按Key值进行划分，放到不他同的Region中。(我的理解是所有M模块调用的Partition类都是相同的，这样可以保证相同的Key会放到编号相同的Region中)。

4.R模块从不同的M模块读取相同Region中的中间数据，然后对它们进行排序，这中间也有可能存在一个Combiner的过程，因为不同M模块肯定包含有相同的Key值，这里再把它们进行一次排序和合并。写出到输出文件。

问题：

1.既然是Hash进行Partition，那么分到各个R模块的数据肯定是无序的，怎样保证输出数据的全局有序呢？如果是每个R模块输出一个文件，那么他们能全局有序吗？还是得在做一次外排？

2.从wordcount这个简单例子的输出结果来看，输出结果是一个文件，并且按Key值排序。

3.上面例子中给出的这个要求全局数据有序的排序方法是在那个阶段执行的？

sortJob.setOutputKeyComparatorClass(IntWritableDecreasingComparator.class);

可以这样解释，Region的数目和R模块的数目是相等的，在上面的代码中设置了Reducer的数目是1，那么Region只有1个，所有的数据都会被这个Reducer读入，进行合并排序，输出到一个文件中。所以能保证最后的结果是有序的。

下面这段话说明OutputKeyComparation的操作是在reduce之前进行的。

All intermediate values associated with a given output key are subsequently grouped by the framework, and passed to the Reducer(s) to determine the final output. Users can control the grouping by specifying a Comparator via JobConf.setOutputKeyComparatorClass(Class).