hadoop 基本原理与应用

一、时代背景

2000 年左右的时候，中国互联网尚处在门户时代，世界互联网已经开始腾飞。

一方面，以 Google 为代表的主流互联网公司开始面临数据爆发的场景；另一方面，当时的互联网公司普遍选择使用配置低下的廉价服务器。 所以急剧增加的数据量与低下的计算存储能力之间的矛盾成为了当时的主流互联网公司面临的主要矛盾之一。 

超级计算机虽然能够解决一部分大数据计算的问题，但是超级计算机价格高昂，主要在实验室和科研机构中采用。工业界尚没有一个统一的且得到广泛商用的分布式计算框架，虽然当时已经也研究出来了一些分布式的计算框架，但是大部分都还停留在实验室或者科研机构里面。

没有得到推广的原因其实也比较简单，就是那些分布式框架设计的实在是太复杂了，根本就没有办法被很好的理解，更不用说写代码去做实现和做分布式计算了。

二、Hadoop 的诞生

在这个时代背景下，我们来看一下 Hadoop 是如何诞生的。

第一个时间点是 2003 年的 11 月份，Google 发表了第一篇论文《The Google File System》，后面我们简称 GFS。主要内容是如何解决分布式存储的问题。

第二个时间节点是 2004 年 12 月份，Google 发表了第二篇论文《MapReduce：Simplified Data Processing on Large Clusters》。讨论的是在大规模集群下一种简化了的数据处理模型，后面我们还会具体分析。 

顺便提一下，在 2006 年的时候，Google 发表了第三篇论文，叫《Bigtable: A Distributed Storage System for Structured Data》。

这三篇论文的发表基本上就标志着整个大数据时代的来临，我们一般也把它们称为是大数据领域的三驾马车。第一篇论文解决的是分布式存储的问题；第二篇论文解决的是分布式计算的问题；第三篇论文解决的是大规模结构化数据的存储和查询问题。

Google 在发表这些论文之前内部已经构建出了相应的系统，而且在公司内部得到了广泛的应用。同时 Google 是一家做搜索引擎的公司，核心问题是解决互联网的爬取和索引构建，当互联网上的数据特别大时，如何对这么多的数据快速的构建索引，在 MapReduce 诞生之前 Google 内部各个部门的具体实现也都不一致，但是碰到的问题和大概的解决思路是类似的，比如如何进行多台机器的任务划分、如何进行容错，但是大家基本都在做一些重复造轮子的工作，而且实现和维护成本都很高。这时 Google 内部的大神 Jeff Dean 就基于大家的实践抽象出了 MapReduce 的架构。

但是，Google 的这三篇论文和 Hadoop 有什么关系呢？这时就要提到第二个人 ——Doug Cutting，Doug Cutting 是一个开源系统爱好者，和 Google 差不多同期也实现了一个开源版本的搜索引擎。

但是，他的引擎始终无法做到 Google 这么快的搜索和索引构建能力，当 Google 发表了这两篇论文之后，Doug Cutting 受到了启发，就实现了开源版本的 GFS 和 MapReduce。一开始做这两个系统也主要是为了解决搜索引擎当中碰到的数据统计问题，但是人们很快发现，MapReduce 这个框架其实是很灵活而且非常通用的。

于是到 2006 年 2 月份，Hadoop 项目就正式成立了，它就包含了两个部分：一部分叫做 MapReduce，解决分布式计算的问题；另外一个部分叫做 HDFS，解决分布式存储的问题。

另外也提一下 Hadoop 这个名字是怎么来的，Doug Cutting 曾经解释说这个名字其实是他儿子的一个毛绒玩具大象的名字，包括现在大家看到的 Hadoop 的黄色大象 Logo，其实也是来自于这个玩具大象的启发。不过现在我们说玩转大数据的时候，常常说让大象能跳舞，也是在形象的强调 Hadoop 给大数据带来的灵活计算的能力。

三、HDFS 的特点

Hadoop 包含了 HDFS 和 MapReduce 两部分，我们先来具体看一下 HDFS 是如何设计的，它有哪些特点呢？

第一个特点：错误检测和自动恢复机制是 HDFS 最核心的架构目标。为什么错误恢复会是当时的核心目标呢，这和当时的服务器配置也有很大关系，那个时候 Google 使用的典型服务器的配置可能是 2 核 4 G 内存这样一个配置，而且服务器很容易出故障，因为谷歌这种商业公司为了省钱往往使用廉价的商用服务器，甚至自己购买 CPU、内存、磁盘，自己组装服务器使用，这样组装出来的服务器可靠性是很低的。

而且在大规模集群下，这个故障率会被放大很多倍， 我们可以计算这样一道数学题：假如说我们一台服务器出故障的概率是 1%，当我们同时使用 100 台服务器的时候，至少有一台服务器出故障的概率是多少呢？答案是 63%，也就是我们有多数机会碰到服务器故障的问题。当我们使用的服务器资源进一步增多时，这个概率还会进一步增加。因此，错误检测和故障恢复是当时设计分布式系统时要考虑的核心目标之一。 

第二个特点是大规模的数据集，HDFS 上的典型文件大小在 GB ~ TB 级别。

第三个特点是流式的数据访问，就是说我们在读取 HDFS 上存储的数据时，一般是按照批量扫描的方式去读取，而不是随机性的访问部分数据。

第四个特点是简单的一致性模型，我们一般将数据一次性的写入 HDFS 上，后续可能会多次读取数据做分析，这是我们高吞吐能力设计的重要前提之一。

最后一个特点是移动计算比移动数据更加划算，这也是一个非常核心的思想，一般来讲，通过网络去读取数据成本要远高于通过本地磁盘读取数据的成本，然后通过本地磁盘读取数据的成本要远高于通过本地内存读取数据的成本。所以移动计算的思想是说，我们尽量把这个计算放在数据所在的位置，而不是将数据拉取到我们最终想要做计算的地方再统一处理，因为移动数据成本很高，但是将计算放在数据所在的地方，就可以让计算本地化，这样是一种非常高效而划算的计算方式。

四、HDFS 数据块

HDFS 数据块（block） 是 HDFS 的一个核心概念，数据块代表 HDFS 文件中读写的最小单位，典型的数据块大小在 64M 或者 128 M 左右。为什么需要引入数据块这个概念呢？

第一个原因是它可以最小化寻址空间。以机械磁盘为例，我们知道磁盘要读写数据的时候，先要通过磁臂的旋转找到对应的磁盘扇区，然后再通过磁头读写一定的数据块，所以其中磁臂旋转寻找位置的过程是非常耗时的，如果数据块足够大，我们将可以将数据连续的存储在一起，这样磁臂的寻址时间就会大大减少，整个磁盘读写的效率也是最高的。 

第二个原因是在我们引入了数据块的概念之后，HDFS 本身的文件大小，就可以直接看做有多少个数据块组成，理论上 HDFS 文件可以是无限大的。

第三个原因是在我们设计存储子系统的时候，整个架构也可以更加的简洁，我们只需要把服务器磁盘看做是一个个的数据块就可以了。

最后一个原因是我们在做数据的副本管理和容错时，也都是按照数据块为单位的，当我们一个数据块丢失或者损坏时，只需要对这当个的数据块进行修复就可以了，不会导致整个文件的损坏。

五、HDFS 整体架构

HDFS 的整体架构如图所示。HDFS 内部分为了两种角色：一种角色叫做 NameNode，另一种角色叫做 DataNode。在一个典型的 HDFS 集群当中，一般会有一个工作的 NameNode 和多个工作的 DataNode，所以它是一个典型的主从架构。

NameNode 会维护所有的元数据，这些元数据主要是指的是 HDFS 文件名、目录结构以及每个 HDFS 文件对应的文件块的真实存储位置等信息。

DataNode 节点一般在每台存储服务器上都会部署一个，主要负责管理存储在这台服务器上的文件的读写。比如说当我们去读取一个文件的时候，我们先去 NameNode 上请求文件的位置，再去对应的 DataNode 节点上读取实际的文件数据。写文件的时候也是类似，我们先请求 NameNode 分配对应的 DataNode 节点和位置信息，然后再去 DataNode 节点上写入数据。

这里还有一个概念是机架，在一个大的 HDFS 集群中，读写数据都会考虑机架的影响，因为同一个机架往往对应着同一个交换机，如果读文件的时候，在同机架的机器上进行读取就会减少中间的网络传输。

反过来，写文件的时候在同一个机架上面写也会更加高效。但是同一个交换机的限制也会导致同一个机架一起出现故障的概率会比较高，因此在写文件的时候除了要考虑效率，还要兼顾文件的可靠性。

5.1 HDFS 读文件的过程

如图所示，当我们使用 HDFS 读取文件的时候，我们的客户端要做 1、3、6 这三个步骤。第 1 步，我们使用 open 操作打开一个文件；然后第 3 步使用 read 操作读取文件的内容；等我们读取完文件的内容之后，我们在第 6 步使用 close 操作关闭文件。

从使用者的角度来看，这和读取一个本地文件几乎没有任何区别，但是背后的执行过程是完全不同的。当第 1 步 open 操作执行之后，HDFS 客户端就会向 NameNode 发起一个请求去获取指定文件的实际存储位置信息，当 read 操作发起之后，HDFS 就会直接去访问对应的 DataNode 的数据，这时候就不再需要和 NameNode 打交道了，读取文件的时候，因为数据可能存储在多个 DataNode 上，客户端会依次从多个 DataNode 节点上读取数据。读取完成之后我们会将文件管道 close 掉，这时候一个完整的 HDFS 文件读取过程就完成了。

5.2 HDFS 写文件的过程

我们再来看一下 HDFS 写文件的过程，和读文件比较类似，写文件的时候从操作者的角度看也是分成 3 步。第 1 步，我们执行 create 操作创建一个文件。然后第 2 步，我们执行 write 操作不停的写入文件内容。第 3 步，在我们写入完成之后执行 close 操作将整个管道关闭。

后面实际发生了哪些事情呢？

首先客户端会向 NameNode 发起一个请求，NameNode 会检查客户端是否拥有相应的权限，检查通过后会分配相应的 DataNode 以及对应的位置信息，告诉客户端去哪里存储文件，客户端获取到这些信息之后，就会去对应的 DataNode 节点上写文件，在第一个 DataNode 节点上写完一个文件块之后，第一个 DataNode 节点还会把数据同步给第二个 DataNode 节点，同样的第二个节点还会把数据同步给第三个 DataNode 节点。

等数据写完之后，第三个节点会将确认消息回传给第二个节点，第二个节点回传给第一个节点，第一个节点将最终的确认消息回传给客户端确认数据写入完成。客户端会再向 NameNode 发送一个消息，告诉 NameNode 去更新相应的元数据信息。

这里为了保证数据的高可用，HDFS 也做了很多的容错机制，核心机制是副本机制。副本机制会保证默认情况下所有的数据都会按照 3 副本的方式进行存储，而且副本的存储位置也很有讲究，第一个节点一般会选择在客户端所在的机器存储第一个副本；然后选择和第一个节点在同一个机架上的另外一个节点上面存储第二个副本；最后选择和前两个节点在不同机架上的节点存储第三个副本。

这种选择一方面是考虑了存储数据的效率，另一方面也是为了尽可能的兼顾不同副本的可用性。另外 HDFS 在写文件的时候还会在本地提供 checksum 校验码机制，只有当所有节点上的数据的校验码一致才会认为数据被写入成功了，从而保证写入过程中的可用性。

六、MapReduce 基本原理

通过 HDFS 我们解决了分布式存储的问题，下面我们再来看一下 Hadoop 的另一个重要组成部分 MapReduce。

MapReduce 主要用来解决分布式计算的问题，它的第一个核心思想是分治法，就是将一个大的任务分成若干个相同的小任务，完成这些小的任务就相当于完成了这个大的任务。

第二个思想是移动计算而不是移动数据，这个思想我们在前边介绍 HDFS 的设计思想时也提过，目的是为了尽量让计算本地化。

第三个思想就是容忍部分节点故障，这个我们在 HDFS 部分也介绍过，在大规模集群在下硬件故障是非常普遍的事情，所以我们的计算框架层面一定要能够容忍这种部分节点的故障。

基于这三个基本思想我们可以看一下这面这个具体的例子：假如我们想要统计一篇文章中各个单词出现的频率，我们要怎么来实现呢？在不考虑大数据量的情况下，我们可以写一个简单的脚本就能实现，现在我们假设想要按照下面示例的伪代码的方法分布两个步骤来实现。

在第一个函数 map 里面，我们主要做的事情是把一篇文章按行读入进来，然后按照单词进行切割，切割完成之后直接输出每个单词，以及后面出现的次数 1。

在第二个函数 reduce 里面，我们输入的是 map 的输出，不过已经把相同的单词聚合到了一起，这时候，我们只需要对每个单词，迭代累加后面单词出现的次数就可以算出最终每个单词出现的次数。

这个例子的计算过程非常简单，现在假如我们把场景换成一个大数据量下的场景，要计算的文档的数据量可能是 TB 级别的，数据也可能是存储在多台机器上的，那我们怎么来完成 WordCount 的计算呢？我们就结合 MapReduce 的具体执行过程来看看。

我们假设大数据量文档文件已经存储在 HDFS 上面，现在我们把文件分割成 4 块，对应图中的 split1 - split4，其中的每个 map 任务会输入其中的一块文档数据，map 会按照我们上面的代码示例对文档的内容进行分割然后输出对应的 Key-Value 对，Key 就是我们要统计的单词，Value 是这个单词出现的次数，在 map 输出时 Value 值固定为 1。

map 输出的内容会进行分区并存储在本地磁盘上面，分区的目的是为了决定 map 输出的这部分内容会对应的发送给哪一个 reduce 进行处理。图中的黄色分区 1 内的数据最终都会发送给 reduce 1 作为输入。

reduce 的输入会是 Key-List<Value>，其中 Key 和 map 输出的 key 相同，List<Value> 是所有 map 输出的 value 的集合，在 reduce 阶段，我们对集合内的 value 重复累加就可以最终计算出想要的统计解决，reduce 的输出最终还会写入到 HDFS 存储系统当中。

在 map 的输出到 reduce 的输入这中间的过程我们称之为 Shuffle，Shuffle 的过程涉及到数据的排序和重新分发，是整个 MapReduce 最核心也是最复杂的过程。其中在 map 输出数据的时候一方面要对数据进行分区，同时在同一个分区内的数据还会按照 Key 进行排序。

map 阶段完成之后，reduce 会去对应的 map 任务的输出目录下拉取对应分区内的数据，同时 reduce 阶段为了保证多个 map 的输出结果是最终有序的还需要在输入阶段做一次归并排序。从整个 MapReduce 的执行过程中我们可以看到，如果是 map 阶段任务发生了失败，那只需要重新执行单个 map 任务，我们可以选择在任意一台节点上面重新读取对应分片的数据，重新将结果输出到本地磁盘。如果是 reduce 阶段任务发生了失败，那只需要执行单个 reduce 任务，同样的 reduce 任务也可以选择在任意一台新节点上面进行执行，重新去对应的 map 输出位置下拉取对应分区的数据，统计最终的计算结果并输出到磁盘当中。

七、Yarn 作业调度过程

前边我们介绍了 MapReduce 的基本原理和主要执行过程，其中还有一个很重要的部分是多个 map 和 reduce 的任务是如何分配的不同的节点上执行的，任务之间的协调工作又是如何做的呢，这里要引入 Yarn 的概念。

Yarn 是从 Hadoop 2 开始引入的新组件，主要用来解决任务资源管理和分配的问题。Yarn 内部也分为两种角色，ResourceManager 和 NodeManager。ResourceManager 主要用来协调 NodeManger 做全局的资源管理和分配。NodeManager 对应一个具体的执行节点，负责管理调度到该节点上的任务实例。一个典型的 MapReduce 作业执行调度的过程如上图所示：

（1）客户端启动 MapReduce 作业。

（2）客户端向 ResourceManager 发送请求，申请一个新的 Application id。

（3）客户端将执行作业所需要的代码和配置文件等本地资源拷贝到 HDFS 等分布式存储系统上。

（4）客户端再次向 ResourceManager 发送提交作业请求。

（5）ResourceManager 找到一个相对空闲的 NodeManager，由这台 NodeManager 启动一个容器，在这个容器内会启动 AppMaster 任务。

（6）AppMaster 会作为这个 MapReduce 作业生命周期内的管理中心，首先对作业进行初始化。

（7）AppMaster 去 HDFS 上获取作业的执行资源，计算出相应的分片信息，这时就可以知道这个 MapReduce 程序一定需要多少执行资源。

（8）AppMaster 向 ResourceManager 申请分配执行资源，这里的执行资源主要值得是 CPU 和内存等硬件资源，在 Yarn 里面实际是以 Container 为载体进行分配的。

（9）AppMaster 申请到相应的资源信息之后，就会去对应的 NodeManager 节点上启动相应的 Container 容器。

（10）NodeManager 启动新的 Container 容器并获取执行作业所需要的代码和分片配置等信息。

（11）在容器内真正开始执行一个 Map 或者 Reduce 的任务。

这里值得一提的是 Yarn 作为独立的组件，定位已经是一个通用的资源管理和协调中心，所以 Yarn 上面不仅能够执行 MapReduce 作业，对于新的 Spark 的作业和 Flink 的作业等也都支持了提交到 Yarn 上面进行管理和执行。

八、Hadoop 生态系统

前面我们介绍了 Hadoop 的两个核心部分：HDFS 解决了分布式存储的问题，MapReduce 解决了分布式计算的问题，最后还介绍了 Yarn 作为独立的资源管理和协调中心在 Hadoop 2 里面被引入。

在这些基础组件之上，整个开源生态圈围绕 Hadoop 基础组件又开发出来了越来越多的新组件，解决在大数据处理过程中碰到的各种各样的问题。现在当我们再泛泛地谈论 Hadoop 的时候，往往指的是整个 Hadoop 生态系统，这里面包含了越来越多的功能组件。

就拿最近比较火热的 Spark 和 Flink 来说，Spark 和 Flink 都是为了解决分布式计算的问题而诞生的，所以它们定位的位置是和 MapReduce 在同一个级别的。Spark 和 Flink 读取和输出的文件一般也都是存储在 HDFS 上面，Spark 和 Flink 作业一般也都可以在 Yarn 上面被执行和调度。

MapReduce 得到广泛应用之后，人们也发现了它存在着两个核心的短板：一个是 MapReduce 执行效率很低，只适合做离线计算，不适合用在时效性要求很高的计算场景；另一个是 MapReduce 架构过于简单，只支持 Map 和 Reduce 两个算子，很多复杂的计算操作都很难直接基于这两个算子进行实现。

而且 MapReduce 提供的是比较底层的 API 接口，依赖于程序开发中要去理解比较多的分布式相关的知识。一方面在 MapReduce 的基础上人们做了很多的修补性的工作，比如引入了 Hive；另一方面是出现了更多的新型计算框架，这里面杰出的就是 Spark 和 Flink。Spark 和 Flink 不仅大大提高了计算的性能问题，更是在流批一体、机器学习、SQL 查询等方面做到了底层架构的统一，暴露出来的是更多高级的上层 API，对于大数据程序开发者来讲也会更加友好。

虽然 MapReduce 会被 Spark、Flink 等新型计算框架逐渐替代，但是 MapReduce 的基本思想是得到了继承和发扬的，对于理解任何一个分布式计算系统来说理解清楚 MapReduce 的执行过程都是基础之一。

同时，Hadoop 作为大数据生态圈的底层基础以及围绕着 HDFS、Yarn 等开发出来的一系列分布式数据处理系统也让整个 Hadoop 生态圈变得越来越完善和活跃。Hadoop 的基础原理和应用也是值得我们每个入门大数据的同学进行学习和思考的内容。谢谢大家。

文渣来源：公众号神策技术社区