Map Reduce 简明教程

MapReduce - Combiners

一个合并器，也称为 semi-reducer, 是一个可选的类，它通过接受 Map 类的输入，然后将输出键值对传递给 Reducer 类的操作来工作。

合并器的主要功能是使用相同的键总结映射输出记录。合并器的输出（键值集合）将作为输入通过网络发送到实际 Reducer 任务。

Combiner

合并器类在 Map 类和 Reduce 类之间使用，减少 Map 和 Reduce 之间的数据传输量。通常，映射任务的输出很大，而传输到减少任务的数据也很大。

以下 MapReduce 任务图显示了 COMBINER PHASE。

How Combiner Works?

以下是关于 MapReduce 合并器工作原理的简要摘要 −

合并器没有预定义的接口，并且它必须实现 Reducer 接口的 reduce() 方法。
合并器在每个映射输出键上运行。它必须与 Reducer 类具有相同的输出键值类型。
合并器可以从大型数据集中生成摘要信息，因为它替换了原始映射输出。

虽然合并器是可选的，但它有助于将数据分为多个组以进行减少阶段的处理，从而简化了处理过程。

MapReduce Combiner Implementation

以下示例提供了有关合并器的理论思想。我们假设我们有以下名为 input.txt 的输入文本文件供 MapReduce 使用。

What do you mean by Object
What do you know about Java
What is Java Virtual Machine
How Java enabled High Performance

下面讨论了带有合并器的 MapReduce 程序的重要阶段。

Record Reader

这是 MapReduce 的第一阶段，其中记录读取器将从输入文本文件中读取每一行文本，并生成键值对作为输出。

Input − 输入文件中的逐行文本。

Output − 形成键值对。以下是一组预期的键值对。

<1, What do you mean by Object>
<2, What do you know about Java>
<3, What is Java Virtual Machine>
<4, How Java enabled High Performance>

Map Phase

映射阶段从记录读取器获取输入，对其进行处理，并将输出生成为另一组键值对。

Input − 以下键值对是从记录读取器获取的输入。

<1, What do you mean by Object>
<2, What do you know about Java>
<3, What is Java Virtual Machine>
<4, How Java enabled High Performance>

映射阶段读取每个键值对，使用 StringTokenizer 从值中划分每个单词，将每个单词视为键，并将该单词的计数视为值。以下代码片段显示了 Mapper 类和映射函数。

public static class TokenizerMapper extends Mapper<Object, Text, Text, IntWritable>
{
   private final static IntWritable one = new IntWritable(1);
   private Text word = new Text();

   public void map(Object key, Text value, Context context) throws IOException, InterruptedException
   {
      StringTokenizer itr = new StringTokenizer(value.toString());
      while (itr.hasMoreTokens())
      {
         word.set(itr.nextToken());
         context.write(word, one);
      }
   }
}

Output − 预期的输出如下 −

<What,1> <do,1> <you,1> <mean,1> <by,1> <Object,1>
<What,1> <do,1> <you,1> <know,1> <about,1> <Java,1>
<What,1> <is,1> <Java,1> <Virtual,1> <Machine,1>
<How,1> <Java,1> <enabled,1> <High,1> <Performance,1>

Combiner Phase

合并器阶段从映射阶段获取每个键值对，对其进行处理，并将输出生成 key-value collection 对。

Input − 以下键值对是从映射阶段获取的输入。

<What,1> <do,1> <you,1> <mean,1> <by,1> <Object,1>
<What,1> <do,1> <you,1> <know,1> <about,1> <Java,1>
<What,1> <is,1> <Java,1> <Virtual,1> <Machine,1>
<How,1> <Java,1> <enabled,1> <High,1> <Performance,1>

合并器阶段读取每个键值对，将公共单词组合成键，将值组合成集合。通常情况下，合并器的代码和操作与还原器的代码和操作类似。以下是对映射器、合并器和还原器类的代码段声明。

job.setMapperClass(TokenizerMapper.class);
job.setCombinerClass(IntSumReducer.class);
job.setReducerClass(IntSumReducer.class);

Output − 预期的输出如下 −

<What,1,1,1> <do,1,1> <you,1,1> <mean,1> <by,1> <Object,1>
<know,1> <about,1> <Java,1,1,1>
<is,1> <Virtual,1> <Machine,1>
<How,1> <enabled,1> <High,1> <Performance,1>

Reducer Phase

还原器阶段从合并器阶段接收每个键值集合对，处理它，并将输出作为键值对传递。请注意，合并器功能与还原器相同。

Input - 以下键值对是从合并器阶段获取的输入。

<What,1,1,1> <do,1,1> <you,1,1> <mean,1> <by,1> <Object,1>
<know,1> <about,1> <Java,1,1,1>
<is,1> <Virtual,1> <Machine,1>
<How,1> <enabled,1> <High,1> <Performance,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);
   }
}

Output - 还原器阶段的预期输出如下：

<What,3> <do,2> <you,2> <mean,1> <by,1> <Object,1>
<know,1> <about,1> <Java,3>
<is,1> <Virtual,1> <Machine,1>
<How,1> <enabled,1> <High,1> <Performance,1>

Record Writer

这是 MapReduce 的最后一个阶段，其中记录编写器从还原器阶段编写每个键值对，并将输出作为文本发送。

Input - 还原器阶段的每个键值对以及输出格式。

Output - 它以文本格式提供给您键值对。以下为预期输出。

What           3
do             2
you            2
mean           1
by             1
Object         1
know           1
about          1
Java           3
is             1
Virtual        1
Machine        1
How            1
enabled        1
High           1
Performance    1

Example Program

以下代码块计算程序中的单词数。

import java.io.IOException;
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;

public class WordCount {
   public static class TokenizerMapper extends Mapper<Object, Text, Text, IntWritable>
   {
      private final static IntWritable one = new IntWritable(1);
      private Text word = new Text();

      public void map(Object key, Text value, Context context) throws IOException, InterruptedException
      {
         StringTokenizer itr = new StringTokenizer(value.toString());
         while (itr.hasMoreTokens())
         {
            word.set(itr.nextToken());
            context.write(word, one);
         }
      }
   }

   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();
      Job job = Job.getInstance(conf, "word count");

      job.setJarByClass(WordCount.class);
      job.setMapperClass(TokenizerMapper.class);
      job.setCombinerClass(IntSumReducer.class);
      job.setReducerClass(IntSumReducer.class);

      job.setOutputKeyClass(Text.class);
      job.setOutputValueClass(IntWritable.class);

      FileInputFormat.addInputPath(job, new Path(args[0]));
      FileOutputFormat.setOutputPath(job, new Path(args[1]));

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

将上述程序另存为 WordCount.java 。下面给出了程序的编译和执行。

Compilation and Execution

我们假设我们位于 Hadoop 用户的主目录中（例如，/home/hadoop）。

按照以下步骤编译和执行上述程序。

Step 1 - 使用以下命令创建目录来存储已编译的 Java 类。

$ mkdir units

Step 2 - 下载用于编译和执行 MapReduce 程序的 Hadoop-core-1.2.1.jar。您可以从 mvnrepository.com 下载 jar。

让我们假设下载的文件夹为 /home/hadoop/。

Step 3 - 使用以下命令编译 WordCount.java 程序并为程序创建 jar。

$ javac -classpath hadoop-core-1.2.1.jar -d units WordCount.java
$ jar -cvf units.jar -C units/ .

Step 4 - 使用以下命令在 HDFS 中创建输入目录。

$HADOOP_HOME/bin/hadoop fs -mkdir input_dir

Step 5 - 使用以下命令在 HDFS 的输入目录中复制名为 input.txt 的输入文件。

$HADOOP_HOME/bin/hadoop fs -put /home/hadoop/input.txt input_dir

Step 6 - 使用以下命令验证输入目录中的文件。

$HADOOP_HOME/bin/hadoop fs -ls input_dir/

Step 7 - 使用以下命令通过从输入目录获取输入文件来运行单词计数应用程序。

$HADOOP_HOME/bin/hadoop jar units.jar hadoop.ProcessUnits input_dir output_dir

Wait for a while till the file gets executed. After execution, the output contains a number of input splits, Map tasks, and Reducer tasks.

Step 8 − Use the following command to verify the resultant files in the output folder.

$HADOOP_HOME/bin/hadoop fs -ls output_dir/

Step 9 − Use the following command to see the output in Part-00000 file. This file is generated by HDFS.

$HADOOP_HOME/bin/hadoop fs -cat output_dir/part-00000

Following is the output generated by the MapReduce program.

What           3
do             2
you            2
mean           1
by             1
Object         1
know           1
about          1
Java           3
is             1
Virtual        1
Machine        1
How            1
enabled        1
High           1
Performance    1