Hadoop【基础知识 01+02】【分布式文件系统HDFS设计原理+特点+存储原理】（部分图片来源于网络）【分布式计算框架MapReduce核心概念+编程模型+combiner&partitioner+词频统计案例解析与进阶+作业的生命周期】（图片来源于网络）

  Hadoop是一个大数据解决方案。它提供了一套分布式系统基础架构。 核心内容包含 hdfs 和mapreduce。hadoop2.0 以后引入 yarn。hdfs 是提供数据存储的，mapreduce 是方便数据计算的。这篇主要说HDFS。

1. hdfs 对应 namenode 和 datanode。 namenode 负责保存元数据的基本信息，datanode 直接存放数据本身；
2. mapreduce 对应 jobtracker 和 tasktracker。 jobtracker 负责分发任务，tasktracker 负责执行具体任务；
3. 对应到 master/slave 架构，namenode 和 jobtracker 就应该对应到 【master】节点，而 datanode和 tasktracker 就应该对应到 【slave】节点。

   1.介绍

     HDFS （Hadoop Distributed File System）是 Hadoop 下的分布式文件系统，具有高容错、高吞吐量等特性，可以部署在低成本的硬件上。

2.设计原理

2.1 HDFS 架构

Client

  Client（代表用 户） 通过与 NameNode 和 DataNode 交互访问 HDFS 中的文件。 Client 提供了一个类似 POSIX 的文件系统接口供用户调用。

NameNode

  整个 Hadoop 集群中只有一个 NameNode。 它是整个系统的“总管”，【负责执行有关文件系统命名空间的操作，例如打开，关闭、重命名文件和目录等。它同时还负责集群元数据的存储，记录着文件中各个数据块的位置信息。】 负责管理 HDFS 的目录树和相关的文件元数据信息。 这些信息是以“fsimage”（ HDFS ==元数据镜像==文件）和“editlog”（HDFS 文件==改动日志==）两个文件形式存放在本地磁盘，当 HDFS 重启时重新构造出来的。此外， NameNode 还负责监控各个 DataNode 的健康状态， 一旦发现某个 DataNode 宕掉，则将该 DataNode 移出 HDFS 并重新备份其上面的数据。

Secondary NameNode

  Secondary NameNode 最重要的任务并不是为 NameNode 元数据进行热备份， 而是定期合并fsimage 和 edits 日志， 并传输给 NameNode。 这里需要注意的是，为了减小 NameNode 压力， NameNode 自己并不会合并 fsimage 和 edits， 并将文件存储到磁盘上， 而是交由Secondary NameNode 完成。

DataNode

  一般而言，每个Slave节点上安装一个 DataNode，【负责提供来自文件系统客户端的读写请求，执行块的创建，删除等操作。】 它负责实际的数据存储， 并将数据信息定期汇报给 NameNode。 DataNode 以固定大小的 block为基本单位组织文件内容， 默认情况下block 大小为 64MB。 当用户上传一个大的文件到 HDFS 上时， 该文件会被切分成若干个 block，分别存储到不同的 DataNode ；同时，为了保证数据可靠， 会将同一个 block 以流水线方式写到若干个（默认是 3，该参数可配置）不同的 DataNode 上。 这种文件切割后存储的过程是对用户
透明的。

2.2 文件系统命名空间

  HDFS 的 文件系统命名空间 的层次结构与大多数文件系统类似 (如 Linux)， 支持目录和文件的创建、移动、删除和重命名等操作，支持配置用户和访问权限，但不支持硬链接和软连接。 NameNode 负责维护文件系统名称空间，记录对名称空间或其属性的任何更改。

2.3 数据复制

  由于 Hadoop 被设计运行在廉价的机器上，这意味着硬件是不可靠的，为了保证容错性，HDFS 提供了数据复制机制。HDFS 将每一个文件存储为一系列块，每个块由多个副本来保证容错，块的大小和复制因子可以自行配置（默认情况下，块大小是 128M，默认复制因子是 3）。

2.4 数据复制的实现原理

  大型的 HDFS 实例在通常分布在多个机架的多台服务器上，不同机架上的两台服务器之间通过交换机进行通讯。在大多数情况下，==同一机架中的服务器间的网络带宽大于不同机架中的服务器之间的带宽==。因此 HDFS 采用机架感知副本放置策略，对于常见情况，当复制因子为 3 时，HDFS 的放置策略是：

  在写入程序位于 datanode 上时，就优先将写入文件的一个副本放置在该 datanode 上，否则放在随机 datanode 上。之后在另一个远程机架上的任意一个节点上放置另一个副本，并在该机架上的另一个节点上放置最后一个副本。此策略可以减少机架间的写入流量，从而提高写入性能。

  如果复制因子大于 3，则随机确定第 4 个和之后副本的放置位置，同时保持每个机架的副本数量低于上限，上限值通常为 （复制系数 - 1）/机架数量 + 2 ，需要注意的是不允许同一个 dataNode 上具有同一个块的多个副本。

  现在问题来了，一个3台服务器的集群，如果复制因子是1，会出现什么情况？情况是所有的数据都存放在nameNode所在的那台服务器上，这个应该没什么疑问吧。

2.5 副本的选择

  为了最大限度地减少带宽消耗和读取延迟，HDFS 在执行读取请求时，优先读取距离读取器最近的副本。如果在与读取器节点相同的机架上存在副本，则优先选择该副本。如果 HDFS 群集跨越多个数据中心，则优先选择本地数据中心上的副本。

2.6 架构的稳定性

1. 心跳机制和重新复制

  每个 DataNode 定期向 NameNode 发送心跳消息，如果超过指定时间没有收到心跳消息，则将DataNode 标记为死亡。NameNode 不会将任何新的 IO 请求转发给标记为死亡的 DataNode，也不会再使用这些 DataNode 上的数据。 由于数据不再可用，可能会导致某些块的复制因子小于其指定值，NameNode 会跟踪这些块，并在必要的时候进行重新复制。

1. 数据的完整性

  由于存储设备故障等原因，存储在 DataNode 上的数据块也会发生损坏。为了避免读取到已经损坏的数据而导致错误，HDFS 提供了数据完整性校验机制来保证数据的完整性，具体操作如下：

  当客户端创建 HDFS 文件时，它会计算文件的每个块的 ==校验和==，并将 校验和 存储在同一 HDFS 命名空间下的单独的隐藏文件中。当客户端检索文件内容时，它会验证从每个 DataNode 接收的数据是否与存储在关联校验和文件中的 校验和 匹配。如果匹配失败，则证明数据已经损坏，此时客户端会选择从其他 DataNode 获取该块的其他可用副本。

1. 元数据的磁盘故障

  FsImage 和 EditLog 是 HDFS 的核心数据，这些数据的意外丢失可能会导致整个 HDFS 服务不可用。为了避免这个问题，可以配置 NameNode 使其支持 FsImage 和 EditLog 多副本同步，这样FsImage 或 EditLog 的任何改变都会引起每个副本 FsImage 和 EditLog 的同步更新。

1. 支持快照

  快照支持在特定时刻存储数据副本，在数据意外损坏时，可以通过回滚操作恢复到健康的数据状态。

3.特点

• 高容错：由于 HDFS 采用数据的多副本方案，所以部分硬件的损坏不会导致全部数据的丢失。
• 高吞吐量：HDFS 设计的重点是支持高吞吐量的数据访问，而不是低延迟的数据访问。
• 大文件支持：HDFS 适合于大文件的存储，文档的大小应该是是 GB 到 TB 级别的。
• 简单一致性模型：HDFS 更适合于一次写入多次读取 (write-once-read-many) 的访问模型。支持将内容追加到文件末尾，但不支持数据的随机访问，不能从文件任意位置新增数据。
• 跨平台移植性：HDFS 具有良好的跨平台移植性，这使得其他大数据计算框架都将其作为数据持久化存储的首选方案。

4.存储原理

1. HDFS写数据原理
   
2. HDFS读数据原理
   
3. HDFS故障类型和其检测方法
   
   第二部分：读写故障的处理
   
   第三部分：DataNode 故障处理
   
   副本布局策略：
   

1. 概述

同 HDFS 一样，Hadoop MapReduce 也采用了 Master/Slave（M/S）架构，具体如图所示。它主要由以下几个组件组成：Client、JobTracker、TaskTracker 和 Task。 下面分别对这几个组件进行介绍。

Client

我们将编写的 MapReduce 程序通过 Client 提交到 JobTracker 端；同时也可通过 Client 提供的一些接口查看作业运行状态。在 Hadoop 内部用 “作业”（Job） 表示 MapReduce 程序。一个 MapReduce 程序可对应若干个作业，而每个作业会被分解成若干个 Map/Reduce 任务（Task）。

JobTracker

JobTracker 主要负责资源监控和作业调度。JobTracker 监控所有 TaskTracker 与作业的健康状况，一旦发现失败情况后，其会将相应的任务转移到其他节点；同时 JobTracker 会跟踪任务的执行进度、资源使用量等信息，并将这些信息告诉任务调度器，而调度器会在资源出现空闲时，选择合适的任务使用这些资源。在 Hadoop 中，任务调度器是一个可插拔的模块，用户可以根据自己的需要设计相应的调度器。

TaskTracker

TaskTracker 会周期性地通过 Heartbeat 将本节点上资源的使用情况和任务的运行进度汇报给JobTracker， 同时接收 JobTracker 发送过来的命令并执行相应的操作（如启动新任务、 杀死任务等）。TaskTracker 使用“slot” 等量划分本节点上的资源量。“slot” 代表计算资源（CPU、内存等）。一个 Task 获取到一个 slot 后才有机会运行，而 Hadoop 调度器的作用就是将各个TaskTracker 上的空闲 slot 分配给 Task 使用。 slot 分为 Map slot 和 Reduce slot 两种，分别供MapTask 和 ReduceTask 使用。 TaskTracker 通过 slot 数目（可配置参数）限定 Task 的并发度。

Task

Task 分为 Map Task 和 Reduce Task 两种， 均由 TaskTracker 启动。 HDFS 以固定大小的 block为基本单位存储数据， 而对于 MapReduce 而言， 其处理单位是 split。split 与 block 的对应关系如图所示。 split 是一个逻辑概念， 它只包含一些元数据信息， 比如数据起始位置、数据长度、数据所在节点等。它的划分方法完全由用户自己决定。 但需要注意的是，split 的多少决定了 Map Task 的数目 ，因为每个 split 会交由一个 Map Task 处理。

Map Task 先将对应的 split 迭代解析成一个个key/value 对，依次调用用户自定义的 map() 函数进行处理，最终将临时结果存放到本地磁盘上，其中临时数据被分成若干个 partition，每个 partition 将被一个 Reduce Task 处理。

ReduceTask执行过程

该过程分为三个阶段

1. 从远程节点上读取 MapTask 中间结果（称为“Shuffle 阶段”）；
2. 按照 key 对 key/value 对进行排序（称为“Sort 阶段”）；
3. 依次读取，调用用户自定义的 reduce() 函数处理，并将最终结果存到 HDFS上（称为“ Reduce 阶段）

Hadoop MapReduce 是一个分布式计算框架，用于编写批处理应用程序。编写好的程序可以提交到Hadoop 集群上用于并行处理大规模的数据集。

MapReduce 作业通过将输入的数据集拆分为独立的块，这些块由 map 以并行的方式处理，框架对map 的输出进行排序，然后输入到 reduce 中。MapReduce 框架专门用于 键值对处理，它将作业的输入视为一组 对，并生成一组 对作为输出。输出和输出的 key 和 value 都必须实现Writable 接口。

(input) <k1, v1> -> map -> <k2, v2> -> combine -> <k2, v2> -> reduce -> <k3, v3> (output)

2. 编程模型简述

这里以词频统计为例进行说明，MapReduce 处理的流程如下【两张图的侧重点不同】：

1. input : 读取文本文件；
2. splitting : 将文件按照行进行拆分，此时得到的 K1 行数， V1 表示对应行的文本内容；
3. mapping : 并行将每一行按照空格进行拆分，拆分得到的 List(K2,V2) ，其中 K2 代表每一个单词，由于是做词频统计，所以 V2 的值为 1，代表出现 1 次；
4. shuffling：由于 Mapping 操作可能是在不同的机器上并行处理的，所以需要通过 shuffling 将相同 key 值的数据分发到同一个节点上去合并，这样才能统计出最终的结果，此时得到 K2 为每一个单词， List(V2) 为可迭代集合， V2 就是 Mapping 中的 V2；
5. Reducing : 这里的案例是统计单词出现的总次数，所以 Reducing 对 List(V2) 进行归约求和操作，最终输出。MapReduce 编程模型中 splitting 和 shuffing 操作都是由框架实现的，需要我们自己编程实现的只有 mapping 和 reducing ，这也就是 MapReduce 这个称呼的来源。

3. combiner & partitioner

3.1 InputFormat & RecordReaders

InputFormat 将输出文件拆分为多个 InputSplit ，并由 RecordReaders 将 InputSplit 转换为标准的键值对，作为 map 的输出。这一步的意义在于只有先进行逻辑拆分并转为标准的键值对格式后，才能为多个 map 提供输入，以便进行并行处理。

3.2 Combiner

combiner 是 map 运算后的可选操作，它实际上是一个本地化的 reduce 操作，它主要是在 map 计算出中间文件后做一个简单的合并重复 key 值的操作。这里以词频统计为例：

map 在遇到一个 hadoop 的单词时就会记录为 1，但是这篇文章里 hadoop 可能会出现 n 多次，那么 map 输出文件冗余就会很多，因此在 reduce 计算前对相同的 key 做一个合并操作，那么需要传输的数据量就会减少，传输效率就可以得到提升。

但并非所有场景都适合使用 combiner ，使用它的原则是 combiner 的输出不会影响到 reduce 计算的最终输入，例如：求总数，最大值，最小值时都可以使用 combiner ，但是做平均值计算则不能使用 combiner 。

不使用 combiner 的情况：

使用 combiner 的情况：

可以看到使用 combiner 的时候，需要传输到 reducer 中的数据由 12keys，降低到 10keys。降低的幅度取决于你 keys 的重复率，下文词频统计案例会演示用 combiner 降低数百倍的传输量。

3.3 Partitioner

partitioner 可以理解成分类器，将 map 的输出按照 key 值的不同分别分给对应的 reducer ，支持自定义实现，下文案例会给出演示。

4. MapReduce词频统计案例

4.1 项目简介

这里给出一个经典的词频统计的案例：统计如下样本数据中每个单词出现的次数。

Spark HBase
Hive Flink Storm Hadoop HBase Spark
Flink
HBase Storm
HBase Hadoop Hive Flink
HBase Flink Hive Storm
Hive Flink Hadoop
HBase Hive
Hadoop Spark HBase Storm
HBase Hadoop Hive Flink
HBase Flink Hive Storm
Hive Flink Hadoop
HBase Hive

为方便大家开发，我在项目源码中放置了一个工具类 WordCountDataUtils ，用于模拟产生词频统计的样本，生成的文件支持输出到本地或者直接写到 HDFS 上。

项目完整源码下载地址： hadoop-word-count

4.2 项目依赖

想要进行 MapReduce 编程，需要导入 hadoop-client 依赖：

<dependency>
  <groupId>org.apache.hadoop</groupId>
  <artifactId>hadoop-client</artifactId>
  <version>${hadoop.version}</version>
</dependency>

4.3 WordCountMapper

将每行数据按照指定分隔符进行拆分。这里需要注意在 MapReduce 中必须使用 Hadoop 定义的类型，因为 Hadoop 预定义的类型都是可序列化，可比较的，所有类型均实现了 WritableComparable 接口。

public class WordCountMapperV2 extends Mapper<LongWritable, Text, Text, IntWritable> {
   
   
    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
   
   
        String[] words = value.toString().split("\t");
        for (String word : words) {
   
   
            context.write(new Text(word), new IntWritable(1));
        }
    }
}

WordCountMapper 对应下图的 Mapping 操作：

WordCountMapper 继承自 Mappe 类，这是一个泛型类，定义如下：

WordCountMapper extends Mapper<LongWritable, Text, Text, IntWritable> 
    public class Mapper<KEYIN, VALUEIN, KEYOUT, VALUEOUT> {
   
   
 ......
}

• KEYIN : mapping 输入 key 的类型，即每行的偏移量 (每行第一个字符在整个文本中的位置)，Long 类型，对应 Hadoop 中的 LongWritable 类型；
• VALUEIN : mapping 输入 value 的类型，即每行数据； String 类型，对应 Hadoop 中 Text 类型；
• KEYOUT ： mapping 输出的 key 的类型，即每个单词； String 类型，对应 Hadoop 中 Text类型；
• VALUEOUT： mapping 输出 value 的类型，即每个单词出现的次数；这里用 int 类型，对应IntWritable 类型。

4.4 WordCountReducer

在 Reduce 中进行单词出现次数的统计：

public class WordCountReducerV2 extends Reducer<Text, IntWritable, Text, IntWritable> {
   
   
    @Override
    protected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
   
   
        int count = 0;
        for (IntWritable value : values) {
   
   
            count += value.get();
        }
        context.write(new Text(key), new IntWritable(count));
    }
}

如下图， shuffling 的输出是 reduce 的输入。这里的 key 是每个单词，values 是一个可迭代的数据
类型，类似 (1,1,1,...) 。

4.5 WordCountApp

组装 MapReduce 作业，并提交到服务器运行，代码如下：

public class WordCountJobV2 extends Configured implements Tool {
   
   

    public static final String HDFS_URL = "hdfs://aliyun:8020";
    private static final String HADOOP_USER_NAME = "root";

    public static void main(String[] args) throws Exception {
   
   
        //执行job作业任务类的对象是谁
        ToolRunner.run(new WordCountJobV2(), args);
    }

    @Override
    public int run(String[] args) throws Exception {
   
   
        int argsNum = 2;
        // 文件输入路径和输出路径由外部传参指定
        if (args.length < argsNum) {
   
   
            System.out.println("Input and output paths are necessary!");
            return 0;
        }
        System.setProperty("HADOOP_USER_NAME", HADOOP_USER_NAME);
        Configuration configuration = new Configuration();
        configuration.set("fs.defaultFS", HDFS_URL);

        // 创建job
        Job job = Job.getInstance(configuration);
        job.setJarByClass(WordCountJobV2.class);

        // 设置map和reduce
        job.setMapperClass(WordCountMapperV2.class);
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(IntWritable.class);

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

        // 设置 inputFormat 和 outputFormat
        job.setInputFormatClass(TextInputFormat.class);
        TextInputFormat.addInputPath(job, new Path(args[0]));

        // 如果输出目录已经存在，则必须先删除，否则重复运行程序时会抛出异常
        FileSystem fileSystem = FileSystem.get(new URI(HDFS_URL), configuration, HADOOP_USER_NAME);
        Path outputPath = new Path(args[1]);
        if (fileSystem.exists(outputPath)) {
   
   
            fileSystem.delete(outputPath, true);
        }
        job.setOutputFormatClass(TextOutputFormat.class);
        TextOutputFormat.setOutputPath(job, outputPath);

        // 将作业提交到群集并等待它完成，参数设置为 true 代表打印显示对应的进度
        boolean completion = job.waitForCompletion(true);

        // 关闭之前创建的 fileSystem
        fileSystem.close();

        // 根据作业结果,终止当前运行的 Java 虚拟机,退出程序
        System.exit(completion ? 0 : -1);
        return completion ? 0 : -1;
    }
}

需要注意的是：如果不设置 Mapper 操作的输出类型，则程序默认它和 Reducer 操作输出的类型相同。

4.6 提交到服务器运行

在实际开发中，可以在本机配置 hadoop 开发环境，直接在 IDE 中启动进行测试。这里主要介绍一下打包提交到服务器运行。由于本项目没有使用除 Hadoop 外的第三方依赖，直接打包即可：

# mvn clean package

使用以下命令提交作业：

hadoop jar /usr/appjar/hadoop-word-count-1.0.jar \
com.heibaiying.WordCountApp \
/wordcount/input.txt /wordcount/output/WordCountApp

作业完成后查看 HDFS 上生成目录：

# 查看目录
hadoop fs -ls /wordcount/output/WordCountApp
# 查看统计结果
hadoop fs -cat /wordcount/output/WordCountApp/part-r-00000

5 词频统计案例进阶之Combiner

5.1 代码实现

想要使用 combiner 功能只要在组装作业时，添加下面一行代码即可：

// 设置 Combiner
job.setCombinerClass(WordCountReducer.class);

5.2 执行结果

加入 combiner 后统计结果是不会有变化的，但是可以从打印的日志看出 combiner 的效果：

没有加入 combiner 的打印日志：

加入 combiner 后的打印日志如下：

这里我们只有一个输入文件并且小于 128M，所以只有一个 Map 进行处理。可以看到经过 combiner 后，records 由 3519 降低为 6 (样本中单词种类就只有 6 种)，在这个用例中 combiner 就能极大地降低需要传输的数据量。

6 词频统计案例进阶之Partitioner

6.1 默认的Partitioner

这里假设有个需求：将不同单词的统计结果输出到不同文件。这种需求实际上比较常见，比如统计产品的销量时，需要将结果按照产品种类进行拆分。要实现这个功能，就需要用到自定义 Partitioner 。

这里先介绍下 MapReduce 默认的分类规则：在构建 job 时候，如果不指定，默认的使用的是HashPartitioner ：对 key 值进行哈希散列并对 numReduceTasks 取余。其实现如下：

public class HashPartitioner<K, V> extends Partitioner<K, V> {
   
   
    public int getPartition(K key, V value, int numReduceTasks) {
   
   
        return (key.hashCode() & Integer.MAX_VALUE) % numReduceTasks;
    }
}

6.2 自定义Partitioner

这里我们继承 Partitioner 自定义分类规则，这里按照单词进行分类：

    public class CustomPartitioner extends Partitioner<Text, IntWritable> {
   
   
    public int getPartition(Text text, IntWritable intWritable, int numPartitions) {
   
   
        return WordCountDataUtils.WORD_LIST.indexOf(text.toString());
    }
}

在构建 job 时候指定使用我们自己的分类规则，并设置 reduce 的个数：

// 设置自定义分区规则
job.setPartitionerClass(CustomPartitioner.class);
// 设置 reduce 个数
job.setNumReduceTasks(WordCountDataUtils.WORD_LIST.size());

6.3 执行结果

执行结果如下，分别生成 6 个文件，每个文件中为对应单词的统计结果：

7 作业的生命周期

1. 作业提交与初始化

提交作业后， 首先由 JobClient 实例将作业相关信息， 比如将程序 jar 包、作业配置文件、 分片元信息文件等上传到分布式文件系统（ 一般为 HDFS）上，其中，分片元信息文件记录了每个输入分片的逻辑位置信息。 然后 JobClient 通过 RPC 通知 JobTracker。

JobTracker 收到新作业提交请求后， 由作业调度模块对作业进行初始化：为作业创建一个JobInProgress 对象以跟踪作业运行状况， 而 JobInProgress 则会为每个 Task 创建一个TaskInProgress 对象以跟踪每个任务的运行状态，TaskInProgress 可能需要管理多个“Task 运行尝试”（ 称为“ Task Attempt”）。

1. 任务调度与监控

前面提到，任务调度和监控的功能均由 JobTracker 完成。TaskTracker 周期性地通过Heartbeat 向 JobTracker 汇报本节点的资源使用情况， 一旦出现空闲资源， JobTracker会按照一定的策略选择一个合适的任务使用该空闲资源， 这由任务调度器完成。 任务调度器是一个可插拔的独立模块， 且为双层架构， 即首先选择作业， 然后从该作业中选择任务， 其中，选择任务时需要重点考虑数据本地性。 此外，JobTracker 跟踪作业的整个运行过程，并为作业的成功运行提供全方位的保障。 首先， 当 TaskTracker 或者 Task 失败时， 转移计算任务 ； 其次， 当某个 Task 执行进度远落后于同一作业的其他 Task 时，为之启动一个相同Task， 并选取计算快的 Task 结果作为最终结果。

1. 任务运行环境准备

运行环境准备包括 JVM 启动和资源隔离， 均由 TaskTracker 实现。 TaskTracker 为每个Task 启动一个独立的 JVM 以避免不同 Task 在运行过程中相互影响 ； 同时，TaskTracker 使用了操作系统进程实现资源隔离以防止Task滥用资源。

1. 任务执行

TaskTracker 为 Task 准备好运行环境后， 便会启动 Task。 在运行过程中， 每个 Task 的最新进度首先由 Task 通过 RPC 汇报给 TaskTracker， 再由 TaskTracker 汇报给 JobTracker。

1. 作业完成。

待所有 Task 执行完毕后， 整个作业执行成功。