MapReduce 教程

映射器

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

映射是将输入记录转换为中间记录的各个任务。转换后的中间记录不必与输入记录属于同一种类型。给定的输入对可以映射到零个或多个输出对。

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

总体而言，映射器实现通过 Job.setMapperClass(Class) 方法传递给作业。然后，框架为该任务的 InputSplit 中的每个键/值对调用 map(WritableComparable, Writable, Context)。然后，应用程序可以覆盖 cleanup(Context) 方法来执行任何必需的清理。

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

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

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

Mapper 输出按每个 Reducer 排序，然后分区。分区的总数与作业的归约任务数相同。用户可以通过实现自定义 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，并且可以覆盖它以初始化它们自己。然后，框架为分组输入中的每个 <key, (list of values)> 对调用 reduce(WritableComparable, Iterable<Writable>, Context) 方法。然后，应用程序可以覆盖 cleanup(Context) 方法以执行任何所需的清理。

Reducer 有 3 个主要阶段：shuffle、sort 和 reduce。

分区器

Partitioner 对键空间进行分区。

分区器控制中间映射输出的键的分区。键（或键的子集）用于派生分区，通常通过哈希函数。分区总数与作业的归约任务数相同。因此，这控制了中间键（以及记录）将被发送到哪个归约任务（m 个）进行归约。

HashPartitioner 是默认的 Partitioner。

计数器

Counter 是 MapReduce 应用程序报告其统计信息的工具。

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

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

内存管理

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

注意：mapreduce.{map|reduce}.java.opts 仅用于配置 MRAppMaster 启动的子任务。守护程序的内存选项配置记录在 配置 Hadoop 守护程序的环境 中。

还可以配置框架某些部分可用的内存。在映射和归约任务中，性能可能会受到影响，具体取决于影响操作并发性和数据命中磁盘频率的参数调整。监视作业的文件系统计数器（尤其是与映射字节计数和归约字节计数相关的计数器）对于调整这些参数非常有价值。

映射参数

从映射中发出的记录将被序列化到缓冲区中，元数据将被存储到会计缓冲区中。如以下选项中所述，当序列化缓冲区或元数据超过阈值时，缓冲区的内容将被排序并写入磁盘，同时映射继续输出记录。如果在溢出进行期间任一缓冲区完全填满，映射线程将被阻塞。当映射完成时，任何剩余记录都将被写入磁盘，所有磁盘段都将合并到一个文件中。最大程度地减少磁盘溢出次数可以减少映射时间，但更大的缓冲区也会减少映射器可用的内存。

其他说明

• 如果在溢出进行时任一溢出阈值被超过，收集将持续到溢出完成。例如，如果 mapreduce.map.sort.spill.percent 设置为 0.33，并且在溢出运行时缓冲区的剩余部分被填满，则下一个溢出将包括所有收集的记录，或缓冲区的 0.66，并且不会生成其他溢出。换句话说，阈值定义的是触发器，而不是阻塞。

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

混洗/归约参数

如前所述，每个归约通过 HTTP 将分区程序分配给它的输出提取到内存中，并定期将这些输出合并到磁盘中。如果已启用映射输出的中间压缩，则每个输出都会解压到内存中。以下选项会影响归约之前这些合并到磁盘的频率，以及在归约期间分配给映射输出的内存。

其他说明

• 如果映射输出大于分配给复制映射输出的内存的 25%，则它将直接写入磁盘，而无需首先通过内存进行暂存。

• 使用合并器运行时，关于高合并阈值和大型缓冲区的推理可能不成立。对于在获取所有映射输出之前启动的合并，合并器在溢出到磁盘时运行。在某些情况下，可以通过投入资源合并映射输出（使磁盘溢出变小并并行溢出和获取）来获得更好的 reduce 时间，而不是激进地增加缓冲区大小。

• 在将内存中的映射输出合并到磁盘以开始 reduce 时，如果由于有要溢出的段和磁盘上至少有 mapreduce.task.io.sort.factor 个段而需要中间合并，则内存中的映射输出将成为中间合并的一部分。

配置的参数

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

注意：在流式作业执行期间，“mapreduce”参数的名称会发生转换。句点 ( . ) 会变成下划线 ( _ )。例如，mapreduce.job.id 会变成 mapreduce_job_id，而 mapreduce.job.jar 会变成 mapreduce_job_jar。要在流式作业的映射器/还原器中获取值，请使用带下划线的参数名称。

任务日志

NodeManager 会读取任务的标准输出 (stdout) 和错误 (stderr) 流以及 syslog，并将其记录到 ${HADOOP_LOG_DIR}/userlogs 中。

分发库

还可使用 DistributedCache 为映射和/或还原任务分发 jar 和本机库。子 jvm 始终会将其当前工作目录添加到 java.library.path 和 LD_LIBRARY_PATH 中。因此，可以通过 System.loadLibrary 或 System.load 加载缓存的库。有关如何通过分布式缓存加载共享库的更多详细信息，请参阅 本机库 中的说明。

作业控制

用户可能需要链接 MapReduce 作业来完成无法通过单个 MapReduce 作业完成的复杂任务。这相当容易，因为作业的输出通常会转到分布式文件系统，而输出反过来又可以用作下一个作业的输入。

但是，这也意味着确保作业完成（成功/失败）的责任完全在于客户端。在这种情况下，各种作业控制选项是

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

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

InputSplit

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

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

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

RecordReader

RecordReader 从 InputSplit 读取 <key, value> 对。

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

OutputCommitter

OutputCommitter 描述 MapReduce 作业的任务输出提交。

MapReduce 框架依赖于作业的 OutputCommitter 来

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

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

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

4. 检查任务是否需要提交。如果任务不需要提交，则避免提交过程。

5. 提交任务输出。任务完成后，任务将根据需要提交其输出。

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

FileOutputCommitter 是默认的 OutputCommitter。作业设置/清理任务占用映射或归约容器，无论哪个在 NodeManager 上可用。作业清理任务、任务清理任务和作业设置任务具有最高优先级，并按此顺序排列。

任务副作用文件

在某些应用程序中，组件任务需要创建和/或写入副作用文件，这些文件不同于实际作业输出文件。

在这种情况下，同一 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}。当然，框架会丢弃失败的任务尝试的子目录。此过程对应用程序完全透明。

应用程序编写人员可以通过在执行任务期间在 ${mapreduce.task.output.dir} 中创建任何所需的辅助文件来利用此功能，通过 FileOutputFormat.getWorkOutputPath(Conext)，框架会以类似的方式为成功的任务尝试提升这些文件，从而无需为每个任务尝试选择唯一路径。

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

对于 reducer=NONE 的作业映射（即 0 个 reduce），整个讨论都成立，因为在这种情况下，映射的输出直接进入 HDFS。

RecordWriter

RecordWriter 将输出 <key, value> 对写入输出文件。

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

将作业提交到队列

用户将作业提交到队列。队列作为作业集合，允许系统提供特定功能。例如，队列使用 ACL 来控制可以向其提交作业的用户。队列预计主要由 Hadoop 调度程序使用。

Hadoop 配置为使用一个名为“default”的单一强制队列。队列名称在 Hadoop 站点配置的 mapreduce.job.queuename 属性中定义。一些作业调度程序（例如 容量调度程序）支持多个队列。

作业通过 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 中。

在该节点上执行任何作业任务之前，框架会将必要的文件复制到工作节点。其效率源于以下事实：每个作业只复制一次文件，并且能够缓存已解压缩的存档。

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

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

可以通过设置属性 mapreduce.job.cache.{files |archives} 来分发文件/存档。如果必须分发多个文件/存档，则可以将它们作为逗号分隔的路径添加。还可以通过 API Job.addCacheFile(URI)/ Job.addCacheArchive(URI) 和 Job.setCacheFiles(URI[])/ Job.setCacheArchives(URI[]) 设置属性，其中 URI 的形式为 hdfs://host:port/absolute-path#link-name。在流处理中，可以通过命令行选项 -cacheFile/-cacheArchive 分发文件。

DistributedCache 还可以用作地图和/或缩减任务中使用的基本软件分发机制。它可用于分发 jar 和本机库。 Job.addArchiveToClassPath(Path) 或 Job.addFileToClassPath(Path) api 可用于缓存文件/jar，并将它们添加到子 jvm 的类路径中。通过设置配置属性 mapreduce.job.classpath.{files |archives} 也可以执行相同的操作。类似地，符号链接到任务工作目录的缓存文件可用于分发本机库并加载它们。

分析

分析是一个实用工具，可为部分映射和缩减获取内置 java 分析器的代表性（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 输出、syslog 和 jobconf。调试脚本的 stdout 和 stderr 的输出显示在控制台诊断中，也作为作业 UI 的一部分显示。

在以下部分中，我们将讨论如何使用作业提交调试脚本。脚本文件需要分发并提交到框架。

数据压缩

Hadoop MapReduce 为应用程序编写者提供了指定中间映射输出和作业输出（即归约输出）的压缩功能。它还捆绑了 CompressionCodec 实现，用于 zlib 压缩算法。还支持 gzip、bzip2、snappy 和 lz4 文件格式。

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

跳过错误记录

Hadoop 提供了一个选项，在处理映射输入时可以跳过某些错误输入记录。应用程序可以通过 SkipBadRecords 类控制此功能。

当映射任务在某些输入上确定性崩溃时，可以使用此功能。这通常是由于映射函数中的错误造成的。通常，用户必须修复这些错误。但是，有时这是不可能的。错误可能出现在第三方库中，例如无法获得其源代码的库。在这种情况下，即使经过多次尝试，任务也永远无法成功完成，并且作业将失败。使用此功能，只会丢失围绕错误记录的小部分数据，这对于某些应用程序（例如对非常大的数据执行统计分析的应用程序）可能是可以接受的。

默认情况下，此功能处于禁用状态。要启用它，请参阅 SkipBadRecords.setMapperMaxSkipRecords(Configuration, long) 和 SkipBadRecords.setReducerMaxSkipGroups(Configuration, long)。

启用此功能后，在一定数量的映射失败后，框架将进入“跳过模式”。有关更多详细信息，请参阅 SkipBadRecords.setAttemptsToStartSkipping(Configuration, int)。在“跳过模式”中，映射任务将维护正在处理的记录范围。为此，框架依赖于已处理记录计数器。请参阅 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. 在美国和其他国家/地区的商标或注册商标。