《Hadoop实战第2版》——3.2节MapReduce计算模型

3.2　MapReduce计算模型
要了解MapReduce，首先需要了解MapReduce的载体是什么。在Hadoop中，用于执行MapReduce任务的机器有两个角色：一个是JobTracker，另一个是TaskTracker。JobTracker是用于管理和调度工作的，TaskTracker是用于执行工作的。一个Hadoop集群中只有一台JobTracker。

3.2.1　MapReduce Job
在Hadoop中，每个MapReduce任务都被初始化为一个Job。每个Job又可以分为两个阶段：Map阶段和Reduce阶段。这两个阶段分别用两个函数来表示，即Map函数和Reduce函数。Map函数接收一个形式的输入，然后产生同样为形式的中间输出，Hadoop会负责将所有具有相同中间key值的value集合到一起传递给Reduce函数，Reduce函数接收一个如形式的输入，然后对这个value集合进行处理并输出结果，Reduce的输出也是形式的。
为了方便理解，分别将三个对标记为、、，那么上面所述的过程就可以用图3-1来表示了。

3.2.2　Hadoop中的Hello World程序
上面所述的过程是MapReduce的核心，所有的MapReduce程序都具有图3-1所示的结构。下面我再举一个例子详述MapReduce的执行过程。
大家初次接触编程时学习的不论是哪种语言，看到的第一个示例程序可能都是“Hello World”。在Hadoop中也有一个类似于Hello World的程序。这就是WordCount。本节会结合这个程序具体讲解与MapReduce程序有关的所有类。这个程序的内容如下：

package cn.edu.ruc.cloudcomputing.book.chapter03;
import java.io.IOException;
import java.util.*;

import org.apache.hadoop.fs.Path;
import org.apache.hadoop.conf.*;
import org.apache.hadoop.io.*;
import org.apache.hadoop.mapred.*;
import org.apache.hadoop.util.*;

public class WordCount {

   public static class Map extends MapReduceBase implements Mapper<LongWritable, Text, Text, IntWritable> {
     private final static IntWritable one = new IntWritable(1);
     private Text word = new Text();

     public void map(LongWritable key, Text value, OutputCollector<Text, IntWritable> output, Reporter reporter) throws IOException {
       String line = value.toString();
       StringTokenizer tokenizer = new StringTokenizer(line);
       while (tokenizer.hasMoreTokens()) {
         word.set(tokenizer.nextToken());
         output.collect(word, one);
       }
     }
   }

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

   public static void main(String[] args) throws Exception {
     JobConf conf = new JobConf(WordCount.class);
     conf.setJobName("wordcount");

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

     conf.setMapperClass(Map.class);
     conf.setReducerClass(Reduce.class);

     conf.setInputFormat(TextInputFormat.class);
     conf.setOutputFormat(TextOutputFormat.class);

     FileInputFormat.setInputPaths(conf, new Path(args[0]));
     FileOutputFormat.setOutputPath(conf, new Path(args[1]));

     JobClient.runJob(conf);
   }
}

同时，为了叙述方便，设定两个输入文件，如下：

echo "Hello World Bye World" > file01
echo "Hello Hadoop Goodbye Hadoop" > file02

看到这个程序，相信很多读者会对众多的预定义类感到很迷惑。其实这些类非常简单明了。首先，WordCount程序的代码虽多，但是执行过程却很简单，在本例中，它首先将输入文件读进来，然后交由Map程序处理，Map程序将输入读入后切出其中的单词，并标记它的数目为1，形成的形式，然后交由Reduce处理，Reduce将相同key值（也就是word）的value值收集起来，形成的形式，之后将这些1值加起来，即为单词的个数，最后将这个对以TextOutputFormat的形式输出到HDFS中。
针对这个数据流动过程，我挑出了如下几句代码来表述它的执行过程：

JobConf conf = new JobConf(MyMapre.class);
conf.setJobName("wordcount");

conf.setInputFormat(TextInputFormat.class);
conf.setOutputFormat(TextOutputFormat.class);

conf.setMapperClass(Map.class);
conf.setReducerClass(Reduce.class);

FileInputFormat.setInputPaths(conf, new Path(args[0]));
FileOutputFormat.setOutputPath(conf, new Path(args[1]));

首先讲解一下Job的初始化过程。Main函数调用Jobconf类来对MapReduce Job进行初始化，然后调用setJobName()方法命名这个Job。对Job进行合理的命名有助于更快地找到Job，以便在JobTracker和TaskTracker的页面中对其进行监视。接着就会调用setInputPath()和setOutputPath()设置输入输出路径。下面会结合WordCount程序重点讲解Inputformat()、OutputFormat()、Map()、Reduce()这4种方法。

1. InputFormat()和InputSplit
   InputSplit是Hadoop中用来把输入数据传送给每个单独的Map，InputSplit存储的并非数据本身，而是一个分片长度和一个记录数据位置的数组。生成InputSplit的方法可以通过Inputformat()来设置。当数据传送给Map时，Map会将输入分片传送到InputFormat()上，InputFormat()则调用getRecordReader()方法生成RecordReader，RecordReader再通过creatKey()、creatValue()方法创建可供Map处理的对，即。简而言之，InputFormat()方法是用来生成可供Map处理的对的。

Hadoop预定义了多种方法将不同类型的输入数据转化为Map能够处理的对，它们都继承自InputFormat，分别是：

BaileyBorweinPlouffe.BbpInputFormat 
 ComposableInputFormat
 CompositeInputFormat
 DBInputFormat
 DistSum.Machine.AbstractInputFormat
 FileInputFormat
其中，FileInputFormat又有多个子类，分别为：
 CombineFileInputFormat
 KeyValueTextInputFormat
 NLineInputFormat
 SequenceFileInputFormat
 TeraInputFormat
 TextInputFormat

其中，TextInputFormat是Hadoop默认的输入方法，在TextInputFormat中，每个文件（或其一部分）都会单独作为Map的输入，而这是继承自FileInputFormat的。之后，每行数据都会生成一条记录，每条记录则表示成形式：
key值是每个数据的记录在数据分片中的字节偏移量，数据类型是LongWritable；
value值是每行的内容，数据类型是Text。
也就是说，输入数据会以如下的形式被传入Map中

file01:
hello world bye world
file02
hello hadoop bye hadoop

因为file01和file02都会被单独输入到一个Map中，因此它们的key值都是0。

1. OutputFormat()
   对于每一种输入格式都有一种输出格式与其对应。同样，默认的输出格式是TextOutputFormat，这种输出方式与输入类似，会将每条记录以一行的形式存入文本文件。不过，它的键和值可以是任意形式的，因为程序内部会调用toString()方法将键和值转换为String类型再输出。最后的输出形式如下所示：

Bye 2
Hadoop 2
Hello 2
World 2
3. Map()和Reduce()
Map()方法和Reduce()方法是本章的重点，从前面的内容知道，Map()函数接收经过InputFormat处理所产生的<k1, v1>，然后输出<k2, v2>。WordCount的Map()函数如下：
public class MyMapre {
   public static class Map extends MapReduceBase implements Mapper<LongWritable, Text, Text, IntWritable> {
     private final static IntWritable one = new IntWritable(1);
     private Text word = new Text();

     public void map(LongWritable key, Text value, 
OutputCollector<Text, IntWritable> output, Reporter reporter) throws IOException {
       String line = value.toString();
       StringTokenizer tokenizer = new StringTokenizer(line);
       while (tokenizer.hasMoreTokens()) {
         word.set(tokenizer.nextToken());
         output.collect(word, one);
     }
   }
}

Map()函数继承自MapReduceBase，并且它实现了Mapper接口，此接口是一个范型类型，它有4种形式的参数，分别用来指定Map()的输入key值类型、输入value值类型、输出key值类型和输出value值类型。在本例中，因为使用的是TextInputFormat，它的输出key值是LongWritable类型，输出value值是Text类型，所以Map()的输入类型即为。如前面的内容所述，在本例中需要输出这样的形式，因此输出的key值类型是Text，输出的value值类型是IntWritable。
实现此接口类还需要实现Map()方法，Map()方法会负责具体对输入进行操作，在本例中，Map()方法对输入的行以空格为单位进行切分，然后使用OutputCollect收集输出的，即。
下面来看Reduce()函数：

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

与Map()类似，Reduce()函数也继承自MapReduceBase，需要实现Reducer接口。Reduce()函数以Map()的输出作为输入，因此Reduce()的输入类型是。而Reduce()的输出是单词和它的数目，因此，它的输出类型是。Reduce()函数也要实现Reduce()方法，在此方法中，Reduce()函数将输入的key值作为输出的key值，然后将获得的多个value值加起来，作为输出的value值。

1. 运行MapReduce程序
   读者可以在Eclipse里运行MapReduce程序，也可以在命令行中运行MapReduce程序，但是在实际应用中，还是推荐到命令行中运行程序。按照第2章介绍的步骤，首先安装Hadoop，然后输入编译打包生成的JAR程序，如下所示（以Hadoop-0.20.2为例，安装路径是~/hadoop）：

mkdir FirstJar
javac -classpath ~/hadoop/hadoop-0.20.2-core.jar -d  FirstJar
WordCount.java
jar -cvf wordcount.jar -C FirstJar/ .

首先建立FirstJar，然后编译文件生成.class，存放到文件夹FirstJar中，并将FirstJar中的文件打包生成wordcount.jar文件。
接着上传输入文件（输入文件是file01，file02，存放在~/input）：

~/hadoop/bin/hadoop dfs –mkdir input
~/hadoop/bin/hadoop dfs –put ~/input/file0* input

在此上传过程中，先建立文件夹input，然后上传文件file01、file02到input中。
最后运行生成的JAR文件，为了叙述方便，先将生成的JAR文件放入Hadoop的安装文件夹中（HADOOP_HOME），然后运行如下命令。

~/hadoop/bin/hadoop jar wordcount.jar WordCount input output
11/01/21 20:02:38 WARN mapred.JobClient: Use GenericOptionsParser for parsing the arguments. Applications should implement Tool for the same.
11/01/21 20:02:38 INFO mapred.FileInputFormat: Total input paths to process : 2
11/01/21 20:02:38 INFO mapred.JobClient: Running job: job_201101111819_0002
11/01/21 20:02:39 INFO mapred.JobClient: map 0% reduce 0%
11/01/21 20:02:49 INFO mapred.JobClient: map 100% reduce 0%
11/01/21 20:03:01 INFO mapred.JobClient: map 100% reduce 100%
11/01/21 20:03:03 INFO mapred.JobClient: Job complete: job_201101111819_0002
11/01/21 20:03:03 INFO mapred.JobClient: Counters: 18
11/01/21 20:03:03 INFO mapred.JobClient: Job Counters 
11/01/21 20:03:03 INFO mapred.JobClient: Launched reduce tasks=1
11/01/21 20:03:03 INFO mapred.JobClient: Launched map tasks=2
11/01/21 20:03:03 INFO mapred.JobClient: Data-local map tasks=2
11/01/21 20:03:03 INFO mapred.JobClient: FileSystemCounters
11/01/21 20:03:03 INFO mapred.JobClient: FILE_BYTES_READ=100
11/01/21 20:03:03 INFO mapred.JobClient: HDFS_BYTES_READ=46
11/01/21 20:03:03 INFO mapred.JobClient: FILE_BYTES_WRITTEN=270
11/01/21 20:03:03 INFO mapred.JobClient: HDFS_BYTES_WRITTEN=31
11/01/21 20:03:03 INFO mapred.JobClient: Map-Reduce Framework
11/01/21 20:03:04 INFO mapred.JobClient: Reduce input groups=4
11/01/21 20:03:04 INFO mapred.JobClient: Combine output records=0
11/01/21 20:03:04 INFO mapred.JobClient: Map input records=2
11/01/21 20:03:04 INFO mapred.JobClient: Reduce shuffle bytes=106
11/01/21 20:03:04 INFO mapred.JobClient: Reduce output records=4
11/01/21 20:03:04 INFO mapred.JobClient: Spilled Records=16
11/01/21 20:03:04 INFO mapred.JobClient: Map output bytes=78
11/01/21 20:03:04 INFO mapred.JobClient: Map input bytes=46
11/01/21 20:03:04 INFO mapred.JobClient: Combine input records=0
11/01/21 20:03:04 INFO mapred.JobClient: Map output records=8
11/01/21 20:03:04 INFO mapred.JobClient: Reduce input records=8

Hadoop命令（注意不是Hadoop本身）会启动一个JVM来运行这个MapReduce程序，并自动获取Hadoop的配置，同时把类的路径（及其依赖关系）加入到Hadoop的库中。以上就是Hadoop Job的运行记录，从这里面可以看到，这个Job被赋予了一个ID号：job_201101111819_0002，而且得知输入文件有两个（Total input paths to process : 2），同时还可以了解Map的输入输出记录（record数及字节数），以及Reduce的输入输出记录。比如说，在本例中，Map的task数量是2个，Reduce的Task数量是一个；Map的输入record数是2个，输出record数是8个等。
可以通过命令查看输出文件输出文件为：

bye 2
hadoop 2
hello 2
world 2
5. 新的API
从0.20.2版本开始，Hadoop提供了一个新的API。新的API是在org.apache.hadoop.mapreduce中的，旧版的API则在org.apache.hadoop.mapred中。新的API不兼容旧的API，WordCount程序用新的API重写如下：
package cn.ruc.edu.cloudcomputing.book.chaptero3;
import java.io.IOException;
import java.util.*;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.conf.*;
import org.apache.hadoop.io.*;
import org.apache.hadoop.mapreduce.*;
import org.apache.hadoop.mapreduce.lib.input.*;
import org.apache.hadoop.mapreduce.lib.output.*;
import org.apache.hadoop.util.*;
public class WordCount extends Configured implements Tool {
public static class Map extends Mapper<LongWritable, Text, Text, IntWritable> {
private final static IntWritable one = new IntWritable(1);
private Text word = new Text();
public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
           String line = value.toString();
           StringTokenizer tokenizer = new StringTokenizer(line);
           while (tokenizer.hasMoreTokens()) {
             word.set(tokenizer.nextToken());
             context.write(word, one);
           }
         }
       }
    
   public static class Reduce extends Reducer<Text, IntWritable, Text, IntWritable> {
      public void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
        int sum = 0;
         for (IntWritable val : values) {
           sum += val.get();
         }
         context.write(key, new IntWritable(sum));
    }
}
    
public int run(String [] args) throws Exception {
     Job job = new Job(getConf());
     job.setJarByClass(WordCount.class);
     job.setJobName("wordcount");

     job.setOutputKeyClass(Text.class);
     job.setOutputValueClass(IntWritable.class);
    
     job.setMapperClass(Map.class);
     job.setReducerClass(Reduce.class);

     job.setInputFormatClass(TextInputFormat.class);
     job.setOutputFormatClass(TextOutputFormat.class);

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

     boolean success = job.waitForCompletion(true);
     return success ? 0 : 1;
}
    
   public static void main(String[] args) throws Exception {
         int ret = ToolRunner.run(new WordCount(), args);
         System.exit(ret);
   }
}

从这个程序可以看到新旧API的几个区别：
在新的API中，Mapper与Reducer已经不是接口而是抽象类。而且Map函数与Reduce函数也已经不再实现Mapper和Reducer接口，而是继承Mapper和Reducer抽象类。这样做更容易扩展，因为添加方法到抽象类中更容易。
新的API中更广泛地使用了context对象，并使用MapContext进行MapReduce间的通信，MapContext同时充当OutputCollector和Reporter的角色。
Job的配置统一由Configurartion来完成，而不必额外地使用JobConf对守护进程进行配置。
由Job类来负责Job的控制，而不是JobClient，JobClient在新的API中已经被删除。
这些区别，都可以在以上的程序中看出。
此外，新的API同时支持“推”和“拉”式的迭代方式。在以往的操作中，对是被推入到Map中的，但是在新的API中，允许程序将数据拉入Map中，Reduce也一样。这样做更加方便程序分批处理数据。

3.2.3　MapReduce的数据流和控制流
前面已经提到了MapReduce的数据流和控制流的关系，本节将结合WordCount实例具体解释它们的含义。图3-2是上例中WordCount程序的执行流程。

由前面的内容知道，负责控制及调度MapReduce的Job的是JobTracker，负责运行MapReduce的Job的是TaskTracker。当然，MapReduce在运行时是分成Map Task和Reduce Task来处理的，而不是完整的Job。简单的控制流大概是这样的：JobTracker调度任务给TaskTracker，TaskTracker执行任务时，会返回进度报告。JobTracker则会记录进度的进行状况，如果某个TaskTracker上的任务执行失败，那么JobTracker会把这个任务分配给另一台TaskTracker，直到任务执行完成。
这里更详细地解释一下数据流。上例中有两个Map任务及一个Reduce任务。数据首先按照TextInputFormat形式被处理成两个InputSplit，然后输入到两个Map中，Map程序会读取InputSplit指定位置的数据，然后按照设定的方式处理该数据，最后写入到本地磁盘中。注意，这里并不是写到HDFS上，这应该很好理解，因为Map的输出在Job完成后即可删除了，因此不需要存储到HDFS上，虽然存储到HDFS上会更安全，但是因为网络传输会降低MapReduce任务的执行效率，因此Map的输出文件是写在本地磁盘上的。如果Map程序在没来得及将数据传送给Reduce时就崩溃了（程序出错或机器崩溃），那么JobTracker只需要另选一台机器重新执行这个Task就可以了。
Reduce会读取Map的输出数据，合并value，然后将它们输出到HDFS上。Reduce的输出会占用很多的网络带宽，不过这与上传数据一样是不可避免的。如果大家还是不能很好地理解数据流的话，下面有一个更具体的图（WordCount执行时的数据流），如图3-3所示。

相信看到图3-3，大家就会对MapReduce的执行过程有更深刻的了解了。
除此之外，还有两种情况需要注意：
1）MapReduce在执行过程中往往不止一个Reduce Task，Reduce Task的数量是可以程序指定的。当存在多个Reduce Task时，每个Reduce会搜集一个或多个key值。需要注意的是，当出现多个Reduce Task时，每个Reduce Task都会生成一个输出文件。
2）另外，没有Reduce任务的时候，系统会直接将Map的输出结果作为最终结果，同时Map Task的数量可以看做是Reduce Task的数量，即有多少个Map Task就有多少个输出文件。