中培教育谈Apache Hadoop

Apache Hadoop是根据Google公司公开的资料开发出来的类似于Google File System的Hadoop File System以及相应的Map/Reduce编程规范。Apache Hadoop也正在进一步开发类似于Google的Chubby系统以及相应的分布式数据库管理系统BigTable的功能。Hadoop是开放源*代码，开发人员可以使用它来开发云计算应用，尤其是云硬件平台。
Hadoop实现了一个分布式文件系统(Hadoop Distributed File System)，简称HDFS。
HDFS有着高容错性的特点，并且用来设计部署在低廉的硬件上。它提供高吞吐量来访问数
据，适合那些有着大量数据的应用程序。HDFS可以以流的形式访问文件系统中的数据。
Hadoop还实现了MapReduce分布式计算模型。MapReduce将应用程序的工作分解成很
多小的工作块。HDFS为了做到可靠性，创建了多份数据块的复制(Replicas)，并将它们放
置在服务器群的计算节点中，MapReduce就可以在它们所在的节点上处理这些数据。
1.Hadoop历史和特点
Hadoop原来是Apache Lucene下的一个子项目，它最初是从Nutch项目中分离出来的专门负责分布式存储以及分布式运算的项目。简单地说，Hadoop是一个处理大规模数据的软件平台，首先来看看Hadoop的整个历史。
2004年，最初的Hadoop版本(现在称为HDFS和MapReduce)是由Doug Cutting和Mike Cafarella开始实施的。2005年12月，Nutch移植到新的框架，Hadoop在20个节点上稳定运行。2006年1月，Doug Cutting加入雅虎公司。2006年2月，雅虎的网格计算团队采用Hadoop，并在同年5月建立了一个300个节点的Hadoop研究集群。2006年11月，研穷  增加到600个节点。2007年1月，研究集群达到900个节点。
2006年2月，Apache Hadoop项目正式启动以支持MapReduce和HDFS的独立发  2006年4月，标准排序（每个节点10 GB）在188个节点上运行47.9个小时。2006年5  标准排序在500个节点上运行42个小时（硬件配置比4月的更好）。2006年12月，朽  序在20个节点上运行1.8个小时、100个节点3.3小时、500个节点5.2小时、900个节，  个小时。
2007年4月，研究集群达到两个1000个节点的集群。2008年4月，Hadoop赢得垲  快，1TB数据排序在900个节点上用时209秒。2008年10月，研究集群每天装载10]  数据。2009年3月，17个集群总共24000台机器。2009年4月，Hadoop实现59秒陕  500 GB（在1400个节点上）和173分钟内排序100 TB数据（在3400个节点上）。
下面列举Hadoop主要的一些特点：
·扩展性强：能可靠地存储和处理千兆字节( GB)数据。
·成本低：可以通过普通机器组成的服务器群来分发以及处理数据，这些服务器群总计可达数千个节点。
·高效率：通过分发数据，Hadoop可以在数据所在的节点上并行地处理它们，非常的快速。
·可靠性好：Hadoop能自动地维护数据的多份复制，并且在任务失败后能自动地重新部署计算任务。
淘宝、百度、腾讯和网易等互联网公司都在使用Hadoop。Hadoop也被主流IT公司用作
其云计算环境中的重要基础软件，如雅虎除了资助Hadoop开发团队外，还在开发基于
Hadoop的开源项目Pig，这是一个专注于海量数据分析的分布式计算程序。Amazon公司基
于Hadoop推出了Amazon S3 (Amazon Simple Storage Service)，提供可靠、快速、可扩展的网络存储服务，以及一个商用的云计算平台Amazon EC2(Amazon Elastic Compute
Cloud)。在IBM公司的云计算项目中，Hadoop也是其中重要的基础软件。IBM在加州的硅谷
实验室研发了以Hadoop为基础的Big Data产品(http：//www.ibm.com/sofiware/data/info sphere/hadoop/)，很多数据库厂商也在做SQL Hadoop。
2.Hadoop的组成
Hadoop是由HDFS、MapReduce、HBase、Hive和ZooKeeper等组成，如图6-1所示。其中，HDFS和MapReduce是两个最基础、最重要的成员，其他子项目提供配套服务。

表6-1描述了各个组件的功能。

                            表6-1  组件的功能

组件	功能
Core	提供了一系列分布式文件系统和通用I/O的组件和接口（序列化、Java RPC和持久化数   据结构）
Avro	一种提供高效、跨语言RPC的数据序列系统，持久化数据存储
MapReduce	分布式数据处理模式和执行环境
HDFS	分布式文件系统
Pig	一种数据流语言和运行环境，用以检索非常大的数据集。Pig运行在MapReduce和   HDFS的集群上
Hbase	一个分布式的、列存储数据库。HBase使用HDFS作为底层存储，同时支持MapReduce   的批量式计算和点查询（随机读取）
ZooKeeper	一个分布式的、高可用性的协调服务。ZooKeeper提供分布式锁等用于构建分布式应用
Hive	分布式数据仓库。Hive管理HDFS中存储的数据，并提供基于SQL的查询语言（在运   行时引擎转换成MapReduce作业）用以查询数据
Chukwa	分布式数据收集和分析系统，它使用MapReduce来生成报告

HDFS是Google GFS的开源版本，是一个高度容错的分布式文件系统，它能够提供高吞
吐量的数据访问，适合存储PB级的大文件，其原理如图6-2所示。HDFS采用Master/Slave
结构；NameNode维护集群内的元数据，对外提供创建、打开、删除和重命名文件或目录的
功能；DataNode存储数据，并负责处理数据的读写请求。DataNode定期向NameNode上报
心跳，NameNode通过响应心跳来控制DataNode。

   

                           图6-2 HDFS架构
3.MapReduce
如图6-3所示，Map负责将数据打散，Reduce负责对数据进行集，用户只需要实现map和reduce两个接口，即可完成TB级数据计算，常见的应用包括日忐分析和数据挎掘等。Hadoop MapReduce的实现也采用了Master/Slave结构。Master叫做JobTracker，  而Slave叫做TaskTracker。用户提交的计算叫做Job，每一个Job会被划分成若干个Tasks。 JobTracker负责Job和Tasks的调度，而TaskTracker负责执行Tasks。

Hadoop Map/Reduce是一个软件框架，基于它写出来的应用程序能够运行在由上千个导组成的大型集群上，并以一种可靠容错的方式并行处理上T级别的数据集。一个Map/Reduce作业(job)通常会把输入的数据集切分为若干独立的数据块，由map任 task)以完全并行的方式处理它们。框架会对map的输出先进行排序，然后把结果输给reduce任务。通常，作业的输入和输出都会被存储在文件系统中。整个框架负责任务的调度和监控，以及重新执行已经失败的任务。

Map/Reduce框架和分布式文件系统通常是运行在一组相同的节点上的，也就是说，计算点和存储节点通常在一起。这种配置允许框架在数据节点上高效地调度任务，可以非常高效地利用整个集群的网络带宽。Map/Reduce框架由一个单独的主(master) JobTracker和每个集群节点一个次( slave) TaskTracker共同组成。master负责调度构成一个作业的所有任务，这些任务分布在不同的slave上，master监控它们的执行，重新执行已经失败的任务；slave仅负责执行由master指派的任务。

应用程序至少应该指明输入／输出的位置（（路径），并通过实现合适的接口或抽象类提供 map和reduce函数，再加上其他作业的参数，就构成了作业配置(job configuration)。然后，Hadoop的job client提交作业(jar包／可执行程序等)和配置信息给JobTracker，后者负责分发这些软件和配置信息给slave，调度任务并监控它们的执行，同时提供状态和诊断信息 job client。还有一点要说明的是，Hadoop框架是用Java实现的，但Map/Reduce应用程，则不一定要用Java来写。
Hadoop Streaming是一种运行作业的实用工具，它允许用户创建和运行任何可执行程序（如Shell工具）来做为mapper和reducer。Hadoop Pipes是一个C++ API，它也可用于实现Map/Reduce应用程序。
MapReduce框架的核心步骤主要分成两个部分：Map和Reduce。当你向MapReduce框架提交一个计算作业时，它会首先把计算作业拆分成若干个Map任务，然后分配到不同的2上去执行。每一个Map任务处理输入数据中的一部分，当Map任务完成后，它会生成·中间文件，这些中间文件将会作为Reduce任务的输入数据。Reduce任务的主要目标就是把前面若干个Map的输出汇总到一起并输出。从高层抽象来看，MapReduce的数据流如图6-4听示。
Shuffle是指从Map产生输出开始，包括系统执行排序以及传送Map输出到Reducer 作为输入的过程。我们首先从Map端开始分析。当Map开始产生输出时，它并不是简单地把数据写入磁盘，因为频繁的磁盘操作会导致性能严重下降。它的处理过程更复杂，数据首：写入内存中的一个缓冲区，然后做了一些预排序，以提升效率。
MapReduce的工作过程分为两个阶段：map阶段和reduce阶段。每个阶段都有键／值对作为输入和输出，并且它们的类型可由程序员选择。程序员还具体定义了两个函数：map函数和reduce函数。图6-5说明了利用MapReduce来处理大数据集的过程，将大数据集分解为成百上千个小数据集，每个（或若干个）数据集分别有集群中的一个结点（一般就是一台普通的计算机）进行处理并生成中间结果，这些中间结果又由大量的结点进行合并，形成最终结果。

                      图6-4  MapReduce高层数据流图

输入
HDFS

                图6-5  多个reduce任务的MapReduce数据流
计算模型的核心是Map和Reduce两个函数，这两个函数由程序员负责实现，功能是按一定的映射规则将输入的<KEY, value>对转换成另一个或一批<KEY, value>对输出'如表6-2所示。

                     表6-2  Map和Reduce函数

函数	输入	输出	说明
Map	<K1,V1>	List(<K2,V2>)	首先将小数据集进一步解析成一批<KEY,VALUE>对，输入Map函数中进行处理。然后，每一个输入的<K1,V1>会输出一批<K2,V2>。<K2,V2>是计算的中间结果
Reduce	<K2,LIST(V2)>	<K3,V3>	输入的中间结果<K2< span>，List（v2）>中的List（v2）表示一批属于同一个k2的值

以一个计算文本文件中每个单词出现的次数的程序为例，<KL< span>，vl>可以是<行在文件中的
偏移位置，文件中的一行>，经Map函数映射之后，形成一批中间结果<单词，出现次数>，
而Reduce函数则可以对中间结果进行处理，将相同单词的出现次数进行累加，得到每个单词
的总的出现次数。
基于MapReduce计算模型编写分布式并行程序并不难，程序员的主要编码工作就是实现
Map和Reduce函数，其它的并行编程中的种种复杂问题，如分布式存储、工作调度、负载平
衡、容错处理和网络通信等，均由MapReduce框架（如Hadoop）负责处理，程序员完全不

用操心。

4.Hive
Hive是基于Hadoop构建的一套数据仓库分析系统，它提供了丰富的SQL查询方式来分析存储在Hadoop分布式文件系统中的数据。Hive将数据存储在数据仓库中，通过自己的 SQL去查询分析数据内容，这套SQL简称Hive SQL。它与关系型数据库的SQL略有不同，但支持了绝大多数的语句，如DDL, DML以及常见的聚合函数、连接查询、条件查询。
如图6-6所示，Hive在Hadoop的架构体系中承担了一个SQL解析的过程，它提供了对外的入口来获取用户的指令然后对指令进行分析，解析出一个MapReduce程序组成可执行计划，并按照该计划生成对应的MapReduce任务提交给Hadoop集群处理，获取最终的结果。    以xml格式所生成的Plan计划保存在HDFS文件系统中，共有两份，一份保存在HDFS中（不删除），一份保存在}玎）FS缓存区内，执行结束后会删除。任务计划由根任务与子任务构成，整个任务计划可能会包含多个MapReduce任务和非MapReduce任务，一个 MapReduce任务中的执行计划也会包括多个子任务。当该MapReduce任务做为一个Job提交的时候，会根据执行计划里的任务流程进行MapReduce处理，然后汇总进行下一步操作。在整个任务的执行中，HiveSql任务经历了“语法解析一生成执行Task树一生成执行计划一分发任务-MapReduce任务执行任务计划”的这样一个过程。

                       图6-6  Hive任务流程图
5 .HDFS在Web开发中的应用
互联网每时每刻都在产生数据，经过长期的积累，这些数据总量非常庞大。存储这些数据需要投入巨大的硬件资源，但是如果能把已有空闲磁盘集群利用起来，就减少了硬件成本。分布式存储这种方案解决了这个问题。
HDFS是一个开源的在Apache上的分布式文件系统框架，它提供了命令行模式和api模式来操作。HDFS文件系统部署在多节点上后，可以上传任意的文件到HDFS中，无需关心文件究竟存储在哪个节点上，只要通过api访问文件即可（流操作）。
如图6-7所示，我们可以在每台数据节点上都部署上Apache服务器，并发布同样的Web程序。当一个请求的数据发现在某个节点上时，我们只需要将URL重定向到节点所在的机器域名，就可以使得当前这个节点的Apache服务器同浏览器建立连接，并直接发送数据。
假设我们拥有100台机器，每台机器都部署了DataNode节点和Apache Web服务器，对外提供一个固定的域名服务器，客户通过浏览器直接访问的是该服务器，在该服务器上的程序接受来自浏览器的请求，通过调用HDFS的接口来判断请求的文件所存放的节点是否为当前节点，如果不是，则根据得到的节点地址去重定向请求到存储该文件的节点的Apache服务器上。由于HDFS中默认切割块为64M，通常图片文件大小不过几M，文件的存放就在一个Block中，如果Block数大于1，因为浏览器不支持分段获取数据，那么需要直接从远程将

数据拉到一个对外服务器上合并成完整的流，然后推送到浏览器中。整个流程如图6-8所示。

                          图6-7 HDFS

                              图6-8读数据流程

NameNode和DataNode被设计成可以在普通的机器上运行，这些机器一般运行着Linux操作系统。HDFS采用Java语言开发，因此任何支持Java的机器都可以部署NameNode或 DataNode。由于采用了可移植性极强的Java语言，使得HDFS可以部署到多种类型的机器上。一个典型的部署场景是一台机器上只运行一个NameNode实例，而集群中的其它机器分别运行一个DataNode实例。这种架构并不排斥在一台机器上运行多个DataNode，只不过这样盼隋况比较少见。集群中单- NameNode的结构大大简化了系统的架构。NameNode是所有HDFS元数据的仲裁者和管理者，用户数据不会流过NameNode。

图6-9描述了客户端如何从HDFS中读取数据。客户端通过调用DistributedFileSystem对象的open()来打开文件（步骤1）。对于HDFS来说，这个对象是分布式文件系统的一个实例。分布式文件系统( DistributedFileSystem)通过使用RPC来调用名称节点，以确定文件开头部分的块的位置（步骤2）。对于每一个块，名称节点返回具有该块的数据节点地址。此外，这些数据节点根据它们与客户端的距离来排序（根据网络集群的拓扑）。如果该客户端本身就是一个数据节点（比如在一个MapReduce任务中），便从本地数据节点中读取。分布式文件系统返回一个FSDatalnputStream对象（一个支持文件定位的输入流）给客户端读取数据。FSDataInputStream转而包装了一个DFSInputStream对象。
接着，客户端对这个输入流调用read()操作（步骤3）。存储着文件开头部分的块的数据节点地址的DFSInputStream随即与这些块最近的数据节点相连接。通过在数据流中重复调用 read()，数据会从数据节点返回客户端（步骤4）。到达块的末端时，DFSInputStream会关闭与数据节点间的联系，然后为下一个块找到最佳的数据节点（步骤5）。客户端只需要读取一个连续的流，这些对于客户端来说都是透明的。客户端从流中读取数据时，块是按照 DFSInputStream打开与数据节点的新连接的顺序读取的。它也会调用名称节点来检索下一组需要的块的数据节点的位置。一旦客户端完成读取，就对文件系统数据输入流调用close()操作（步骤6）。

                    图6-9客户端从HDFS中读取数据
图6-10描述了客户端对HDFS写入数据。客户端通过在分布式文件系统    ( DistributedFileSystem)中调用create()来创建文件（步骤1）。分布式文件系统通过一个RPC去调用名称节点，在文件系统的命名空间中创建一个新的文件（步骤2），这时没有块与之相联系。名称节点执行各种不同的检查以确保这个文件不存在，并且客户端有可以创建文件的适当的许可。如果这些检查通过，名称节点就会生成一个新文件的记录；否则，文件创建失败并向客户端抛出一个IOException异常。分布式文件系统返回一个文件系统数据输出流，让客户端开始写入数据。就像读取一样，文件系统数据输出流控制一个    DFSOutputStream，负责处理数据节点和名称节点之间的通信。在客户端写入数据时(步骤    3)，DFSOutputStream将它分成一个个的包，写入内部的队列，称为数据队列。数据队列随    数据流流动，数据流的责任是根据适合的数据节点的列表来要求这些节点为副本分配新的块。这个数据节点的列表形成一个管线。我们假设这个副本数是3，那么有3个节点在管线   中。数据流将包分流给管线中第一个的数据节点，这个节点会存储包并且发送给管线中的第    二个数据节点。同样地，第二个数据节点存储包并且传给管线中第三个（也是最后一个）数
据节点（步骤4）。DFSOutputStream也有一个内部的包队列来等待数据节点确认，称为确认    队列。一个包只有在被管线中所有节点确认后才会被移出确认队列（步骤5）。

                      图6-10客户端对HDFS写入数据
在一个块被写入期间，多个数据节点发生故障的可能性虽然有但很少见。只要 dfs.replication.min的副本（默认为1）被写入，写操作就是成功的，并且这个块会在集群中被异步复制，直到满足其目标副本数（dfs.replication的默认设置为3）。
客户端完成数据的写入后，就会在流中调用close()操作（步骤6）。在向名称节点发送完信息之前，此方法会将余下的所有包放入数据节点管线并等待确认（步骤7）。名称节点已经知道文件由哪些块组成，所以它只需在返回成功前等待块进行最小量的复制。