MapReduce教程

Mapper

Mapper 将输入的键/值对映射为一组中间键/值对。

Map是将输入记录转换为中间记录的独立任务。转换后的中间记录不必与输入记录类型相同。一个给定的输入键值对可以映射为零个或多个输出键值对。

Hadoop MapReduce框架会为作业的InputFormat生成的每个InputSplit生成一个map任务。

总体而言，mapper实现通过Job.setMapperClass(Class)方法传递给作业。框架随后会为任务中的InputSplit里的每个键/值对调用map(WritableComparable, Writable, Context)。应用程序可以重写cleanup(Context)方法来执行任何必要的清理操作。

输出键值对不需要与输入键值对类型相同。一个给定的输入键值对可以映射到零个或多个输出键值对。通过调用context.write(WritableComparable, Writable)来收集输出键值对。

应用程序可以使用Counter来报告其统计信息。

与给定输出键关联的所有中间值随后由框架进行分组，并传递给Reducer以确定最终输出。用户可以通过Job.setGroupingComparatorClass(Class)指定Comparator来控制分组过程。

Mapper的输出会被排序，然后按Reducer进行分区。分区总数与作业的reduce任务数相同。用户可以通过实现自定义的Partitioner来控制哪些键（以及记录）发送到哪个Reducer。

用户可以选择性地通过Job.setCombinerClass(Class)指定一个combiner，用于对中间输出进行本地聚合，这有助于减少从Mapper传输到Reducer的数据量。

中间排序后的输出始终以简单的(key-len, key, value-len, value)格式存储。应用程序可以控制是否压缩中间输出以及通过Configuration指定要使用的CompressionCodec压缩编解码器。

Reducer

Reducer 将共享同一个键的一组中间值缩减为更小的一组值。

作业的reduce数量由用户通过Job.setNumReduceTasks(int)设置。

总体而言，Reducer实现会通过Job.setReducerClass(Class)方法接收作业的Job对象，并可重写该方法进行自身初始化。框架随后会对分组输入中的每个键值对调用reduce(WritableComparable, Iterable, Context)方法。应用程序还可以重写cleanup(Context)方法来执行必要的清理操作。

Reducer 包含3个主要阶段：shuffle（洗牌）、sort（排序）和reduce（归约）。

分区器

Partitioner 对键空间进行分区。

分区器(Partitioner)控制着中间map输出键的分区方式。通常通过哈希函数，使用键(或键的子集)来推导分区。分区总数与作业的reduce任务数相同。因此这决定了中间键(以及记录)会被发送到哪个m个reduce任务中进行归约处理。

HashPartitioner 是默认的 Partitioner。

计数器

Counter 是MapReduce应用程序用于报告其统计数据的工具。

Mapper 和 Reducer 实现可以使用 Counter 来报告统计信息。

Hadoop MapReduce 附带了一个库，其中包含通常有用的映射器、归约器和分区器。

内存管理

用户/管理员还可以使用mapreduce.{map|reduce}.memory.mb来指定启动的子任务及其递归启动的任何子进程的最大虚拟内存。请注意，此处设置的值是每个进程的限制。mapreduce.{map|reduce}.memory.mb的值应以兆字节(MB)为单位指定。此外，该值必须大于或等于传递给JavaVM的-Xmx参数，否则虚拟机可能无法启动。

注意：mapreduce.{map|reduce}.java.opts仅用于配置从MRAppMaster启动的子任务。守护进程的内存选项配置请参阅Configuring the Environment of the Hadoop Daemons。

框架某些部分可用的内存也是可配置的。在map和reduce任务中，通过调整影响操作并发性和数据落盘频率的参数，可能会对性能产生影响。监控作业的文件系统计数器——尤其是与map输出字节数和reduce输入字节数相关的指标——对于调优这些参数至关重要。

Map参数

从map任务发出的记录会被序列化到一个缓冲区中，元数据则存入统计缓冲区。如下文选项所述，当序列化缓冲区或元数据的容量超过阈值时，缓冲区内容将在后台进行排序并写入磁盘，同时map任务继续输出记录。若在溢出过程中任一缓冲区完全填满，map线程将被阻塞。当map任务完成时，所有剩余记录会被写入磁盘，所有磁盘上的分段将合并为单个文件。减少磁盘溢出次数可以缩短map任务时间，但更大的缓冲区也会减少mapper可用的内存。

其他注意事项

• 如果在溢出过程中任一溢出阈值被超过，收集操作将持续到溢出完成。例如，如果mapreduce.map.sort.spill.percent设置为0.33，且在溢出运行时缓冲区的剩余部分被填满，下一次溢出将包含所有已收集的记录（即缓冲区的0.66部分），而不会产生额外的溢出。换句话说，这些阈值定义的是触发条件，而非阻塞条件。

• 大于序列化缓冲区的记录将首先触发溢出，然后被溢出到一个单独的文件中。该记录是否会先经过合并器是未定义的。

Shuffle/Reduce 参数

如前所述，每个reduce任务通过HTTP获取由Partitioner分配给它的输出到内存中，并定期将这些输出合并到磁盘。如果启用了map输出的中间压缩，每个输出会被解压到内存中。以下选项会影响reduce之前合并到磁盘的频率以及在reduce期间分配给map输出的内存。

其他注意事项

• 如果map输出的大小超过分配给复制map输出的内存的25%，它将直接写入磁盘，而无需先通过内存暂存。

• 当运行带有combiner时，关于高合并阈值和大缓冲区的推理可能不成立。对于在所有map输出获取完成前启动的合并操作，combiner会在数据溢出到磁盘时运行。在某些情况下，通过投入资源来合并map输出（使磁盘溢出量变小并实现溢出与获取的并行化），而不是激进地增加缓冲区大小，可以获得更好的reduce时间。

• 当将内存中的map输出合并到磁盘以开始reduce操作时，如果由于存在待溢出的分段且磁盘上已有至少mapreduce.task.io.sort.factor个分段而需要进行中间合并，则内存中的map输出将参与该中间合并过程。

已配置参数

以下属性在每个任务执行的作业配置中被本地化：

注意：在流式作业执行期间，"mapreduce"参数的名称会被转换。点号(.)会变成下划线(_)。例如，mapreduce.job.id会变成mapreduce_job_id，mapreduce.job.jar会变成mapreduce_job_jar。要在流式作业的mapper/reducer中获取这些值，请使用带下划线的参数名称。

任务日志

任务的标准输出(stdout)、错误输出(stderr)流以及系统日志由NodeManager读取，并记录到${HADOOP_LOG_DIR}/userlogs目录中。

分发库

DistributedCache 也可用于分发jar包和本地库，供map和/或reduce任务使用。子JVM始终会将其当前工作目录添加到java.library.path和LD_LIBRARY_PATH中。因此可以通过System.loadLibrary或System.load加载缓存的库。关于如何通过分布式缓存加载共享库的更多细节，请参阅Native Libraries文档。

作业控制

用户可能需要串联多个MapReduce作业来完成单个MapReduce作业无法处理的复杂任务。这相当容易实现，因为作业的输出通常会写入分布式文件系统，而这些输出又可以被用作下一个作业的输入。

然而，这也意味着确保作业完成（成功/失败）的责任完全落在客户端身上。在这种情况下，各种作业控制选项包括：

• Job.submit() : 将作业提交到集群并立即返回。

• Job.waitForCompletion(boolean) : 将作业提交到集群并等待其完成。

InputSplit

InputSplit 表示由单个 Mapper 处理的数据。

通常InputSplit提供面向字节的输入视图，而RecordReader负责处理并呈现面向记录的视图。

FileSplit 是默认的 InputSplit。它会将 mapreduce.map.input.file 设置为逻辑分片对应的输入文件路径。

RecordReader

RecordReader 从 InputSplit 中读取 键值对。

通常，RecordReader会将由InputSplit提供的面向字节的输入视图转换为面向记录的视图，供Mapper实现处理。因此，RecordReader承担了处理记录边界的职责，并为任务提供键和值。

OutputCommitter

OutputCommitter 描述了MapReduce作业中任务输出的提交过程。

MapReduce框架依赖于作业的OutputCommitter来实现以下功能：

1. 在初始化期间设置作业。例如，在作业初始化期间为作业创建临时输出目录。当作业处于PREP状态并在初始化任务之后，作业设置由一个单独的任务完成。一旦设置任务完成，作业将移动到RUNNING状态。

2. 在作业完成后进行清理工作。例如，在作业完成后移除临时输出目录。作业清理由作业结束时的一个独立任务完成。作业在清理任务完成后被声明为成功/失败/已终止。

3. 设置任务的临时输出。任务设置作为同一任务的一部分，在任务初始化期间完成。

4. 检查任务是否需要提交。这是为了避免在任务不需要提交时执行提交流程。

5. 提交任务输出。任务完成后，如有需要，该任务将提交其输出。

6. 放弃任务提交。如果任务已经失败/被终止，其输出将被清理。如果任务无法清理（在异常块中），将启动一个具有相同尝试ID的单独任务来执行清理。

FileOutputCommitter是默认的OutputCommitter。作业设置/清理任务会占用NodeManager上可用的map或reduce容器。其中JobCleanup任务、TaskCleanup任务和JobSetup任务具有最高优先级，且优先级按此顺序排列。

任务副作用文件

在某些应用中，组件任务需要创建和/或写入辅助文件，这些文件与实际的工作输出文件不同。

在这种情况下，可能会出现同一个Mapper或Reducer的两个实例同时运行（例如推测性任务）并尝试在FileSystem上打开和/或写入同一文件（路径）的问题。因此，应用程序编写者必须为每个任务尝试（例如使用attemptid，如attempt_200709221812_0001_m_000000_0）选择唯一名称，而不仅仅是为每个任务选择。

为避免这些问题，当OutputCommitter为FileOutputCommitter时，MapReduce框架会在FileSystem上为每个任务尝试维护一个特殊的${mapreduce.output.fileoutputformat.outputdir}/_temporary/_${taskid}子目录（可通过${mapreduce.task.output.dir}访问），用于存储该任务尝试的输出。当任务尝试成功完成时，${mapreduce.output.fileoutputformat.outputdir}/_temporary/_${taskid}中的文件（仅限该目录）会被提升至${mapreduce.output.fileoutputformat.outputdir}。当然，框架会丢弃失败任务尝试的子目录。这个过程对应用程序完全透明。

应用程序编写者可以利用这一特性，通过在任务执行期间使用FileOutputFormat.getWorkOutputPath(Conext)在${mapreduce.task.output.dir}中创建所需的任何辅助文件，框架会为成功的任务尝试同样地提升这些文件，从而消除了为每个任务尝试选择唯一路径的需要。

注意：在执行特定任务尝试期间，${mapreduce.task.output.dir}的实际值为${mapreduce.output.fileoutputformat.outputdir}/_temporary/_{$taskid}，该值由MapReduce框架设置。因此，只需在MapReduce任务中通过FileOutputFormat.getWorkOutputPath(Conext)返回的路径中创建任何辅助文件，即可利用此功能。

整个讨论同样适用于reducer=NONE（即0个reducer）的作业映射，因为在这种情况下，映射的输出会直接写入HDFS。

RecordWriter

RecordWriter 将输出的 键值对写入输出文件。

RecordWriter实现将作业输出写入FileSystem。

向队列提交作业

用户将作业提交到队列。队列作为作业的集合，使系统能够提供特定功能。例如，队列使用ACL来控制哪些用户可以提交作业。队列主要供Hadoop调度器使用。

Hadoop默认配置了一个名为'default'的强制性队列。队列名称在Hadoop站点配置的mapreduce.job.queuename属性中定义。某些作业调度器（如Capacity Scheduler）支持多队列配置。

作业通过mapreduce.job.queuename属性或Configuration.set(MRJobConfig.QUEUE_NAME, String) API定义需要提交到的队列。设置队列名称是可选的。如果作业提交时未关联队列名称，则会被提交到'default'默认队列。

计数器

Counters代表全局计数器，由MapReduce框架或应用程序定义。每个Counter可以是任意Enum类型。特定Enum的计数器会被分组到Counters.Group类型的组中。

应用程序可以定义任意的Counters（类型为Enum），并通过在map和/或reduce方法中调用Counters.incrCounter(Enum, long)或Counters.incrCounter(String, String, long)来更新它们。这些计数器随后会被框架进行全局聚合。

DistributedCache

DistributedCache 高效地分发应用程序特定的大型只读文件。

DistributedCache 是MapReduce框架提供的一种机制，用于缓存应用程序所需的文件（文本、归档文件、jar包等）。

应用程序通过Job中的URL（hdfs://）指定需要缓存的文件。DistributedCache假定通过hdfs:// URL指定的文件已经存在于FileSystem上。

该框架会在任何任务在该工作节点上执行之前，将必要文件复制到工作节点。其高效性源于每个作业仅需复制一次文件的能力，以及可以缓存这些在worker节点上解压的归档文件。

DistributedCache会跟踪缓存文件的修改时间戳。显然，在作业执行期间，应用程序或外部程序不应修改这些缓存文件。

DistributedCache 可用于分发简单的只读数据/文本文件以及更复杂的类型，如压缩包和jar文件。压缩包（zip、tar、tgz和tar.gz文件）在工作节点上会被解压。文件会被设置执行权限。

可以通过设置属性mapreduce.job.cache.{files |archives}来分发文件/归档文件。如果需要分发多个文件/归档文件，可以用逗号分隔的路径来添加。这些属性也可以通过APIJob.addCacheFile(URI)/Job.addCacheArchive(URI)和Job.setCacheFiles(URI[])/Job.setCacheArchives(URI[])来设置，其中URI的格式为hdfs://host:port/absolute-path#link-name。在Streaming中，可以通过命令行选项-cacheFile/-cacheArchive来分发文件。

DistributedCache也可以作为基础软件分发机制，用于map和/或reduce任务中。它可用于分发jar包和本地库。Job.addArchiveToClassPath(Path)或Job.addFileToClassPath(Path)接口可用于缓存文件/jar包，并将它们添加到子JVM的classpath中。同样可以通过设置配置属性mapreduce.job.classpath.{files |archives}来实现。类似地，被符号链接到任务工作目录的缓存文件可用于分发本地库并加载它们。

性能分析

性能分析是一种实用工具，用于获取内置Java分析器在部分map和reduce任务中的代表性样本（2到3个）。

用户可以通过设置配置属性mapreduce.task.profile来指定系统是否应收集作业中某些任务的性能分析信息。该值可通过API Configuration.set(MRJobConfig.TASK_PROFILE, boolean)进行设置。若该值设为true，则启用任务性能分析。分析信息将存储在用户日志目录中。默认情况下，作业不启用性能分析功能。

一旦用户配置需要进行分析，她/他可以使用配置属性mapreduce.task.profile.{maps|reduces}来设置要分析的MapReduce任务范围。该值可以通过API Configuration.set(MRJobConfig.NUM_{MAP|REDUCE}_PROFILES, String)进行设置。默认情况下，指定的范围是0-2。

用户还可以通过设置配置属性mapreduce.task.profile.params来指定性能分析器的配置参数。该值可以使用API Configuration.set(MRJobConfig.TASK_PROFILE_PARAMS, String)进行设置。如果字符串中包含%s，在任务运行时它将被替换为性能分析输出文件的名称。这些参数会通过命令行传递给任务子JVM。性能分析参数的默认值为-agentlib:hprof=cpu=samples,heap=sites,force=n,thread=y,verbose=n,file=%s。

调试

MapReduce框架提供了一种运行用户自定义脚本进行调试的功能。当MapReduce任务失败时，用户可以运行调试脚本，例如处理任务日志。该脚本可以访问任务的stdout和stderr输出、系统日志以及jobconf。调试脚本的stdout和stderr输出会显示在控制台诊断信息中，同时也作为作业用户界面的一部分展示。

在以下章节中，我们将讨论如何为作业提交调试脚本。该脚本文件需要分发并提交至框架。

数据压缩

Hadoop MapReduce为应用程序开发者提供了对中间map输出和作业输出(即reduce的输出)进行压缩的功能。它还内置了对zlib压缩算法的CompressionCodec实现。同时支持gzip、bzip2、snappy和lz4文件格式。

出于性能（zlib）和Java库不可用的原因，Hadoop还提供了上述压缩编解码器的本地实现。关于它们的使用和可用性的更多详细信息请参见此处。

跳过错误记录

Hadoop提供了一个选项，可以在处理map输入时跳过一组特定的错误输入记录。应用程序可以通过SkipBadRecords类来控制此功能。

当map任务在某些输入上确定性地崩溃时，可以使用此功能。这通常是由于map函数中的错误导致的。通常情况下，用户需要修复这些错误。然而，有时这是不可能的。例如，错误可能存在于无法获取源代码的第三方库中。在这种情况下，即使经过多次尝试，任务也无法成功完成，作业将失败。通过此功能，只有坏记录周围的一小部分数据会丢失，这对于某些应用程序（例如对非常大的数据进行统计分析）可能是可以接受的。

默认情况下此功能是禁用的。如需启用，请参考SkipBadRecords.setMapperMaxSkipRecords(Configuration, long)和SkipBadRecords.setReducerMaxSkipGroups(Configuration, long)。

启用此功能后，框架在经历一定数量的map失败后会进入"跳过模式"。更多详情请参阅SkipBadRecords.setAttemptsToStartSkipping(Configuration, int)。在"跳过模式"下，map任务会维护正在处理的记录范围。为此，框架依赖于已处理记录计数器。参见SkipBadRecords.COUNTER_MAP_PROCESSED_RECORDS和SkipBadRecords.COUNTER_REDUCE_PROCESSED_GROUPS。该计数器使框架能够知道有多少记录已成功处理，从而确定导致任务崩溃的记录范围。在后续尝试中，将跳过这些记录范围。

跳过的记录数量取决于应用程序增加处理记录计数器的频率。建议在每条记录处理完成后增加该计数器。对于通常采用批量处理方式的应用，这可能无法实现。在这种情况下，框架可能会跳过坏记录周围的额外记录。用户可以通过SkipBadRecords.setMapperMaxSkipRecords(Configuration, long)和SkipBadRecords.setReducerMaxSkipGroups(Configuration, long)控制跳过的记录数量。框架会尝试使用类似二分查找的方法缩小跳过记录的范围。跳过的范围会被分成两半，只执行其中一半。在后续失败时，框架会确定哪一半包含坏记录。任务将被重新执行，直到达到可接受的跳过值或用尽所有任务尝试次数。要增加任务尝试次数，可使用Job.setMaxMapAttempts(int)和Job.setMaxReduceAttempts(int)

跳过的记录以序列文件格式写入HDFS，供后续分析。可以通过SkipBadRecords.setSkipOutputPath(JobConf, Path)更改存储位置。

源代码

import java.io.BufferedReader;
import java.io.FileReader;
import java.io.IOException;
import java.net.URI;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.StringTokenizer;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.Counter;
import org.apache.hadoop.util.GenericOptionsParser;
import org.apache.hadoop.util.StringUtils;

public class WordCount2 {

  public static class TokenizerMapper
       extends Mapper<Object, Text, Text, IntWritable>{

    static enum CountersEnum { INPUT_WORDS }

    private final static IntWritable one = new IntWritable(1);
    private Text word = new Text();

    private boolean caseSensitive;
    private Set<String> patternsToSkip = new HashSet<String>();

    private Configuration conf;
    private BufferedReader fis;

    @Override
    public void setup(Context context) throws IOException,
        InterruptedException {
      conf = context.getConfiguration();
      caseSensitive = conf.getBoolean("wordcount.case.sensitive", true);
      if (conf.getBoolean("wordcount.skip.patterns", false)) {
        URI[] patternsURIs = Job.getInstance(conf).getCacheFiles();
        for (URI patternsURI : patternsURIs) {
          Path patternsPath = new Path(patternsURI.getPath());
          String patternsFileName = patternsPath.getName().toString();
          parseSkipFile(patternsFileName);
        }
      }
    }

    private void parseSkipFile(String fileName) {
      try {
        fis = new BufferedReader(new FileReader(fileName));
        String pattern = null;
        while ((pattern = fis.readLine()) != null) {
          patternsToSkip.add(pattern);
        }
      } catch (IOException ioe) {
        System.err.println("Caught exception while parsing the cached file '"
            + StringUtils.stringifyException(ioe));
      }
    }

    @Override
    public void map(Object key, Text value, Context context
                    ) throws IOException, InterruptedException {
      String line = (caseSensitive) ?
          value.toString() : value.toString().toLowerCase();
      for (String pattern : patternsToSkip) {
        line = line.replaceAll(pattern, "");
      }
      StringTokenizer itr = new StringTokenizer(line);
      while (itr.hasMoreTokens()) {
        word.set(itr.nextToken());
        context.write(word, one);
        Counter counter = context.getCounter(CountersEnum.class.getName(),
            CountersEnum.INPUT_WORDS.toString());
        counter.increment(1);
      }
    }
  }

  public static class IntSumReducer
       extends Reducer<Text,IntWritable,Text,IntWritable> {
    private IntWritable result = new IntWritable();

    public void reduce(Text key, Iterable<IntWritable> values,
                       Context context
                       ) throws IOException, InterruptedException {
      int sum = 0;
      for (IntWritable val : values) {
        sum += val.get();
      }
      result.set(sum);
      context.write(key, result);
    }
  }

  public static void main(String[] args) throws Exception {
    Configuration conf = new Configuration();
    GenericOptionsParser optionParser = new GenericOptionsParser(conf, args);
    String[] remainingArgs = optionParser.getRemainingArgs();
    if ((remainingArgs.length != 2) && (remainingArgs.length != 4)) {
      System.err.println("Usage: wordcount <in> <out> [-skip skipPatternFile]");
      System.exit(2);
    }
    Job job = Job.getInstance(conf, "word count");
    job.setJarByClass(WordCount2.class);
    job.setMapperClass(TokenizerMapper.class);
    job.setCombinerClass(IntSumReducer.class);
    job.setReducerClass(IntSumReducer.class);
    job.setOutputKeyClass(Text.class);
    job.setOutputValueClass(IntWritable.class);

    List<String> otherArgs = new ArrayList<String>();
    for (int i=0; i < remainingArgs.length; ++i) {
      if ("-skip".equals(remainingArgs[i])) {
        job.addCacheFile(new Path(remainingArgs[++i]).toUri());
        job.getConfiguration().setBoolean("wordcount.skip.patterns", true);
      } else {
        otherArgs.add(remainingArgs[i]);
      }
    }
    FileInputFormat.addInputPath(job, new Path(otherArgs.get(0)));
    FileOutputFormat.setOutputPath(job, new Path(otherArgs.get(1)));

    System.exit(job.waitForCompletion(true) ? 0 : 1);
  }
}

示例运行

作为输入的示例文本文件：

$ bin/hadoop fs -ls /user/joe/wordcount/input/
/user/joe/wordcount/input/file01
/user/joe/wordcount/input/file02

$ bin/hadoop fs -cat /user/joe/wordcount/input/file01
Hello World, Bye World!

$ bin/hadoop fs -cat /user/joe/wordcount/input/file02
Hello Hadoop, Goodbye to hadoop.

运行应用程序：

$ bin/hadoop jar wc.jar WordCount2 /user/joe/wordcount/input /user/joe/wordcount/output

输出：

$ bin/hadoop fs -cat /user/joe/wordcount/output/part-r-00000
Bye 1
Goodbye 1
Hadoop, 1
Hello 2
World! 1
World, 1
hadoop. 1
to 1

请注意，这里的输入与我们最初看到的版本有所不同，并且这些差异如何影响输出结果。

现在，让我们通过DistributedCache插入一个模式文件，该文件列出了需要忽略的单词模式。

$ bin/hadoop fs -cat /user/joe/wordcount/patterns.txt
\.
\,
\!
to

再次运行它，这次使用更多选项：

$ bin/hadoop jar wc.jar WordCount2 -Dwordcount.case.sensitive=true /user/joe/wordcount/input /user/joe/wordcount/output -skip /user/joe/wordcount/patterns.txt

如预期所示，输出：

$ bin/hadoop fs -cat /user/joe/wordcount/output/part-r-00000
Bye 1
Goodbye 1
Hadoop 1
Hello 2
World 2
hadoop 1

再运行一次，这次关闭大小写敏感：

$ bin/hadoop jar wc.jar WordCount2 -Dwordcount.case.sensitive=false /user/joe/wordcount/input /user/joe/wordcount/output -skip /user/joe/wordcount/patterns.txt

果然，输出如下：

$ bin/hadoop fs -cat /user/joe/wordcount/output/part-r-00000
bye 1
goodbye 1
hadoop 2
hello 2
world 2

亮点

第二个版本的WordCount通过利用MapReduce框架提供的某些特性改进了前一个版本：

• 演示应用程序如何在Mapper（和Reducer）实现的setup方法中访问配置参数。

• 演示如何使用DistributedCache来分发作业所需的只读数据。在此示例中，它允许用户在计数时指定要跳过的单词模式。

• 展示GenericOptionsParser在处理Hadoop通用命令行选项方面的实用性。

• 演示应用程序如何使用Counters计数器，以及如何设置传递给map映射(和reduce归约)方法的应用程序特定状态信息。

Java和JNI是Oracle America, Inc.在美国及其他国家的商标或注册商标。