MapReduce 组合器
组合器，也称为 半减速机， 是一个可选类，它通过接受来自 Map 类的输入然后将输出键值对传递给 Reducer 类来进行操作。
Combiner的主要作用是对同一个key的map输出记录进行汇总。组合器的输出(键值集合)将通过网络发送到实际的 Reducer 任务作为输入。
Combiner
Combiner 类用于 Map 类和 Reduce 类之间，以减少 Map 和 Reduce 之间的数据传输量。通常，map 任务的输出量很大，传递给 reduce 任务的数据量很大。
以下 MapReduce 任务图显示了 COMBINER PHASE。
合路器如何工作？
下面简单总结一下 MapReduce Combiner 的工作原理：
-
组合器没有预定义的接口，它必须实现 Reducer 接口的 reduce() 方法。
-
组合器对每个映射输出键进行操作。它必须具有与 Reducer 类相同的输出键值类型。
-
组合器可以从大型数据集中生成摘要信息，因为它替换了原始 Map 输出。
虽然，Combiner 是可选的，但它有助于将数据分成多个组以用于 Reduce 阶段，从而更容易处理。
MapReduce 组合器实现
以下示例提供了有关组合器的理论思想。让我们假设我们有以下名为的输入文本文件 输入.txt 对于 MapReduce。
What do you mean by Object What do you know about Java What is Java Virtual Machine How Java enabled High Performance
下面讨论了使用 Combiner 的 MapReduce 程序的重要阶段。
记录阅读器
这是 MapReduce 的第一阶段，记录阅读器从输入文本文件中读取每一行作为文本，并产生作为键值对的输出。
输入 ： 输入文件中的逐行文本。
输出 ：形成键值对。以下是一组预期的键值对。
<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 阶段从 Record Reader 获取输入，对其进行处理，并将输出作为另一组键值对生成。
输入 ：下面的键值对是从Record Reader中取出的输入。
<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 阶段读取每个键值对，使用 StringTokenizer 将每个单词与值相除，将每个单词视为键并将该单词的计数视为值。以下代码片段显示了 Mapper 类和 map 函数。
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);
}
}
}
输出 ： 预期输出如下：
<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 阶段从 Map 阶段获取每个键值对，对其进行处理，并将输出生成为 键值集合 pairs.
输入 ：下面的键值对是从Map阶段取的输入。
<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 的代码和操作类似于 Reducer。以下是 Mapper、Combiner 和 Reducer 类声明的代码片段。
job.setMapperClass(TokenizerMapper.class); job.setCombinerClass(IntSumReducer.class); job.setReducerClass(IntSumReducer.class);
输出 ： 预期输出如下：
<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 阶段从 Combiner 阶段获取每个键值集合对，对其进行处理，并将输出作为键值对传递。请注意，Combiner 功能与 Reducer 相同。
输入 ： 下面的键值对是从Combiner阶段取的输入。
<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 阶段读取每个键值对。以下是组合器的代码片段。
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);
}
}
输出 ：Reducer阶段的预期输出如下：
<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>
记录作家
这是 MapReduce 的最后一个阶段，其中 Record Writer 从 Reducer 阶段写入每个键值对并将输出作为文本发送。
输入 ：Reducer 阶段的每个键值对连同输出格式。
输出 ： 它为你提供文本格式的键值对。以下是预期的输出。
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
示例程序
以下代码块计算程序中的字数。
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.File输入Format;
import org.apache.hadoop.mapreduce.lib.output.File输出Format;
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.set输出KeyClass(Text.class);
job.set输出ValueClass(IntWritable.class);
File输入Format.add输入Path(job, new Path(args[0]));
File输出Format.set输出Path(job, new Path(args[1]));
System.exit(job.waitForCompletion(true) ? 0 : 1);
}
}
将上述程序另存为 WordCount.java .下面给出程序的编译和执行。
编译和执行
假设我们在 Hadoop 用户的主目录中(例如，/home/hadoop)。
按照下面给出的步骤编译和执行上述程序。
步骤 1 ： 使用下面的命令创建一个目录来存放编译好的java类。
$ mkdir units
步骤 2 ： 下载Hadoop-core-1.2.1.jar，用于编译执行MapReduce程序。你可以从以下位置下载 jar mvnrepository.com .
让我们假设下载的文件夹是/home/hadoop/。
步骤 3 ： 使用以下命令编译 WordCount.java 程序并为该程序创建一个 jar。
$ javac -classpath hadoop-core-1.2.1.jar -d units WordCount.java $ jar -cvf units.jar -C units/ .
步骤 4 ：使用以下命令在HDFS中创建一个输入目录。
$HADOOP_HOME/bin/hadoop fs -mkdir input_dir
步骤 5 ：使用以下命令复制输入文件名为 输入.txt 在 HDFS 的输入目录中。
$HADOOP_HOME/bin/hadoop fs -put /home/hadoop/input.txt input_dir
步骤 6 ： 使用以下命令验证输入目录下的文件。
$HADOOP_HOME/bin/hadoop fs -ls input_dir/
步骤 7 ： 使用以下命令通过从输入目录中获取输入文件来运行字数统计应用程序。
$HADOOP_HOME/bin/hadoop jar units.jar hadoop.ProcessUnits input_dir output_dir
等待一段时间，直到文件被执行。执行后，输出包含多个输入拆分、Map 任务和 Reducer 任务。
步骤 8 ： 使用以下命令验证输出文件夹中的结果文件。
$HADOOP_HOME/bin/hadoop fs -ls output_dir/
步骤 9 ： 使用下面的命令查看中的输出 Part-00000 文件。该文件由 HDFS 生成。
$HADOOP_HOME/bin/hadoop fs -cat output_dir/part-00000
以下是 MapReduce 程序生成的输出。
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