RAID：SAN/NAS采用RAID磁盘阵列进行存储，而Hadoop框架通过复本来确保数据的可靠性和容错性。
DAS采用JBOD磁盘数组进行存储，如果Hadoop节点的内置存储容量较小，可以采用DAS做扩展。如果只是想通过Hadoop做数据归档，没有计算，好吧，SAN/NAS是个选择。
Hadoop是由Apache开源软件基金会开发的，运行于大规模普通服务器上的分布式系统基础架构，用于大规模数据的存储、计算、分析等。通过使用Hadoop平台用户可以在不了解分布式底层细节的情况下，开发分布式程序，充分利用集群的威力进行高速运算和存储。2007年雅虎发布了第一个Apache Hadoop版本0141；2008年雅虎用Hadoop做到全网尺度的搜索；2009年雅虎把内部版本全部开源，于是IBM也加入Hadoop的开发阵营；2010年Facebook宣布正式运行世界最大的Hadoop集群；2011年Apache Hadoop10版本发布；2012年Apache Hadoop20版本发布。下面具体介绍一下Hadoop系统的架构。
Hadoop由许多元素构成，如下图图所示，包括HBase、Hive、Pig、Chukwa、Oozie和ZooKeeper等，但是其核心组件为HDFS和MapReduce。
HDFS是Hadoop Distributed File System系统的缩写，是一个使用JAVA语言实现的、分布式的、可扩展的文件系统，它存储 Hadoop 集群中所有存储节点上的文件，由NameNode和DataNode两部分组成。HDFS的上一层是 MapReduce 引擎，该引擎由 JobTrackers 和 TaskTrackers 组成，用来对存储在HDFS上的数据进行计算分析。下面来具体介绍HDFS和MapReduce的工作原理及应用。
HDFS
HDFS采用master/slave架构。一个HDFS集群是由一个Namenode和一定数目的Datanodes组成。Namenode是一个中心服务器，负责管理文件系统的名字空间(namespace)以及客户端对文件的访问。集群中的Datanode是集群中的数据节点，用来存储实际的数据，并负责管理它所在节点上的数据存储。HDFS公开了文件系统的名字空间，用户能够以文件的形式在上面存储数据。从内部看，一个文件被分成一个或多个数据块，这些块存储在一组Datanode上。Namenode执行文件系统的名字空间 *** 作，比如打开、关闭、重命名文件或目录。它也负责确定数据块到具体Datanode节点的映射。Datanode负责处理文件系统客户端的读写请求。在Namenode的统一调度下进行数据块的创建、删除和复制，下面就具体来阐述HDFS系统中涉及的基本概念；
数据块（Block） HDFS和传统的分布式文件系统一样，也采用了数据块的概念，将数据分割成固定大小的数据块进行存储，默认大小为64MB，块的大小可针对每个文件配置，由客户端任意指定，并且每个块都有属于自己的全局ID，作为一个独立的单位存储在集群服务器上。与传统分布式文件系统不同的是，如果实际数据没有达到块大小时，则并不实际占用磁盘空间。HDFS元数据 HDFS元数据由文件系统目录树信息、文件和数据块的对应关系和块的存放位置三个部分组成，文件系统目录树信息包括文件名、目录名及文件和目录的从属关系，文件和目录的大小，创建及最后访问时间。文件和块的对应关系记录了文件由哪些块组成。此外元数据还记录了块的存放位置，包括存放块的机器名和块ID。NameNode HDFS对元数据和实际数据采取分别存储的方式，元数据存储在一台指定的服务器上，称为NameNode，实际数据则存储在集群中的其他机器上的文件系统中，称为DataNode。NameNode是用来管理文件系统命名空间的组件，并且一个HDFS集群只有一台NameNode，由于元数据存储在NameNode上，当NameNode出现故障时将导致整个集群无法工作。元数据保存在NameNode的内存当中，以便快速查询，1G内存大致可以存放1000000个块对应的元数据信息。DataNode DataNode用来存储块的实际数据，每个块会在本地文件系统产生两个文件，一个是实际的数据文件，另一个是块的附加信息文件，其中包括数据的校验和生成时间等信息。DataNode通过心跳包(Heartbeat)与NameNode通信，当客户端读取/写入数据的时候将直接与DataNode进行通信。Secondary NameNode Secondary NameNode在Hadoop集群中起到至关重要的作用，首先需要明确它并不是NameNode的备份节点，它和NameNode运行在不同的主机上，它主要的工作是阶段性地合并NameNode的日志文件，控制NameNode日志文件的大小。此外，在NameNode硬盘损坏的情况下，Secondary NameNode也可用作数据恢复，但恢复的只是部分数据。
HDFS架构及工作原理
下图为HDFS对数据存储的原理图，NameNode存储了DataNode节点所存储数据的元数据，即Hdfs和MapReduce两个文件的分块信息，假设单个文件的存储份数为3，即每个数据块有三份备份，那么数据在DataNode上的存储的原则为：相同的两个数据块存储在同一机架的不同的DataNode节点上；第三个数据块存储在不同机架上的DataNode节点上。这样就解决了当某个DataNode节点出现故障的时候数据丢失的问题，保障了存储在HDFS系统上数据的可用性。
Hadoop MapReduce
MapReduce是Google公司的核心计算模型，它将运行于大规模集群上的复杂的并行计算过程高度地抽象为两个函数：Map和Reduce。MapReduce也可以看成是一种解决问题的方法，它把一个复杂的任务分解成多个任务，Map负责把任务分解成多个任务，Reduce负责把分解后多任务处理的结果汇总起来。
Hadoop中的MapReduce是一个简易的软件框架，基于它写出来的应用程序能够运行在由上千台机器组成的大型集群上，并以一种可靠容错的方式并行处理TB级别的数据集，实现了Hadoop在集群上的数据和任务的并行计算与处理。在并行计算中其他的种种复杂的问题，如分布式存储、工作调度、负载均衡、容错处理、网络通信等均由MapReduce框架负责处理，编程人员可以不用关心。用MapReduce来处理的数据集必须具备这样的特点：待处理的数据集可以分解成许多小的数据集，并且每个小的数据集都可以完全并行地进行处理。
Hadoop MapReduce实现
Hadoop MapReduce是基于HDFS的MapReduce编程框架实现的，我们把MapReduce处理的问题称为作业 (Job)，并将作业分解为任务 (Task)，在MapReduce执行过程中需要有两种任务。
Map 把输入的键/值对转换成一组中间结果的键/值对Reduce 把Map任务产生的一组具有相同键的中间结果根据逻辑转换生成较小的最终结果。
Hadoop MapReduce的服务进程
Hadoop MapReduce有两个主要的服务进程，一个是单独运行在主节点上的JobTracker进程，另一个是运行在每个集群从节点上的TaskTracker进程。服务进程部署如下图所示。
JobTraker和NameNode运行在同一个服务器上，我们称为Hadoop集群的主节点，负责接收客户端提交的作业，并将任务分配到不同的计算节点TaskTracker上，同时监控作业的运行情况，完成作业的更新和容错处理；Tasktracker通常和DataNode装在一起，称为Hadoop集群的从节点，它调用Map和Reduce执行JobTracker指派的任务，并发送心跳消息给JobTracker，向JobTracker汇报可运行任务的数量。
Hadoop安全机制
Hadoop 一直缺乏安全机制，主要表现在以下几个方面。
User to Service：NameNode或者JobTracker缺乏安全认证机制；DataNode缺乏安全授权机制；JobTracker缺乏安全授权机制。Service to Service安全认证：Datanode与TaskTracker缺乏安全授权机制，这使得用户可以随意启动假的DataNode和TaskTracker。磁盘或者通信连接没有经过加密。
为了增强Hadoop的安全机制， 从2009年起Apache专门抽出一个团队为Hadoop增加安全认证和授权机制，Apache Hadoop 100版本之后的版本添加了安全机制，但是升级到该版本后可能会导致Hadoop的一些应用不可用。
可以只用一行代码来运行MapReduce作业：JobClientrunJon(conf)，Job作业运行时参与的四个实体：
1JobClient 写代码，配置作业，提交作业。
2JobTracker:初始化作业，分配作业，协调作业运行。这是一个java程序，主类是JobTracker。
3TaskTracker：运行作业划分后的任务，即分配数据分配上执行Map或Reduce任务。
4HDFS：保存作业数据、配置信息等，保存作业结果。
Map/Reduce 作业总体执行流程：
代码编写 ----> 作业配置 ----> 作业提交 ----> Map任务分配和执行 ----> 处理中间结果 ----> Reduce任务分配与执行 ----> 输出结果
而对于每个作业的执行，又包含：
输入准备 ----> 任务执行 ----> 输出结果
作业提交JobClient：
JobClient的runJob方法产生一个Jobclient实例并调用其submitJob方法，然后runJob开始循环吗，并在循环中调用getTaskCompetionEvents方法，获得TaskCompletionEvent实例，每秒轮询作业进度（后面有介绍进度和状态更新），把进度写到控制台，作业完成后显示作业计数器，若失败，则把错误记录到控制台。
submitJob方法作业提交的过程：
1向JobTracker请求一个新的JobId。
2检查作业相关路径，如果路径不正确就会返回错误。
3计算作业输入分片及其划分信息。
4将作业运行需要的资源（jar文件、配置文件等）复制到Shared HDFS，并
复制多个副本（参数控制，默认值为10）供tasktracker访问，也会将计算的分片复制到HDFS。
5调用JobTracker对象的submitJob()方法来真正提交作业，告诉JobTracker作业准备执行。
作业的初始化JobTracker：
JobTracker收到submitJob方法调用后，会把调用放入到一个内部队列，由作业调度器（Job scheduler）进行调度并对其初始化。Job初始化即创建一个作业对象。
当作业被调度后，JobTracker会创建一个代表这个作业的JobInProgress对象，并将任务和记录信息封装在这个对象中，以便跟踪任务状态和进程。
初始化过程就是JobInProgress对象的initTasks方法进行初始化的。
初始化步骤：
1从HDFS中读取作业对应的jobsplit信息，为后面的初始化做好准备。
2创建并初始化map和reduce任务。根据数据分片信息中的个数确定map task的个数，然后为每个map task生成一个TaskInProgress对象来处理数据分片，先将其放入nonRunningMapCache,以便JobTracker分配任务的时候使用。接下来根据JobConf中的mapredreducetasks属性利用setNumReduceTasks()方法设置reduce task的数量，然后同map task创建方式。
3最后就是创建两个初始化task，进行map和reduce的初始化。
任务的分配JobTracker：
消息传递HeartBeat： tasktracker运行一个简单循环定期发送心跳（heartbeat）给JobTracker。由心跳告知JobTracker自己是否存活，同时作为消息通道传递其它信息（请求新task）。作为心跳的一部分，tasktracker会指明自己是否已准备好运行新的任务，如果是，jobtracker会分配它一个任务。
分配任务所属于的作业：在Jobtracker分配任务前需先确定任务所在的作业。后面会介绍到各种作业调度算法，默认是一个FIFO的作业调度。
分配Map和Reduce任务：tasktracker有固定数量的任务槽,一个tasktracker可以同时运行多个Map和Reduce任务，但其准确的数量由tasktracker的核的数量和内存大小决定。默认调度器会先填满Map任务槽，再填Reduce任务槽。jobtracker会选择距离离分片文件最近的tasktracker，最理想情况下，任务是数据本地化（data-local）的,当然也可以是机架本地化（rack-local），如果不是本地化的，那么他们就需要从其他机架上检索数据。Reduce任务分配很简单，jobtracker会简单的从待运行的reduce任务列表中选取下一个来执行，不用考虑数据本地化。
任务的执行TaskTracker：
TaskTracker收到新任务后，就要在本地运行任务了，运行任务的第一步就是通过localizedJob将任务本地化所需要的注入配置、数据、程序等信息进行本地化。
1本地化数据：从共享文件系统将jobsplit 、jobjar (在分布式缓存中)复制本地，将job配置信息写入jobxml。
2新建本地工作目录：tasktracker会加压jobjar文件到本工作目录。
3调用launchTaskForJob方法发布任务（其中会新建TaskRunner实例运行任务），如果是Map任务就启用MapTaskRunner，对于Reduce就是ReduceTaskRunner。
在这之后，TaskRunner会启用一个新的JVM来运行每个Map/Reduce任务，防止程序原因而导致tasktracker崩溃，但不同任务间重用JVM还是可以的，后续会讲到任务JVM重用。
对于单个Map，任务执行的简单流程是：
1分配任务执行参数
2在Child临时文件中添加map任务信息（Child是运行Map和Reduce任务的主进程）
3配置log文件夹，配置map任务的通信和输出参数
4读取input split，生成RecordReader读取数据
5为Map生成MapRunnable,依次从RecordReader中接收数据，并调用Map函数进行处理。
6最后将map函数的输出调用collect收集到MapOutputBuffer（参数控制其大小）中。
Streaming和Pipes:
Streaming和Pipes都运行特殊的Map和Reduce任务，目的是运行用户提供的可执行程序并与之通信。
Streaming:使用标准输入输出Streaming与进程进行通信。
Pipes:用来监听套接字，会发送一个端口号给C++程序，两者便可建立链接。
进度和状态更新：
一个作业和它的任务都有状态（status），其中包括：运行成功失败状态、Map/Reduce进度、作业计数器值、状态消息。
状态消息与客户端的通信：
1对于Map任务Progress的追踪：progress是已经处理完的输入所占的比例。
2对于Reduce：稍复杂，reduce任务分三个阶段（每个阶段占1/3），复制、排序和Reduce处理，若reduce已执行一半的输入的话，那么任务进度便是1/3+1/3+1/6=5/6。
3任务计数器：任务有一组计数器，负责对任务运行各个事件进行计数。
4任务进度报告：如果任务报告了进度，便会设置一个标记以表明状态将被发送到tasktracker。有一个独立线程每隔三秒检查一次此标记，如果已设置，则告知tasktracker当前状态。
5tasktracker进度报告：tasktracker会每隔5秒（这个心跳是由集群大小决定，集群越大时间会越长）发送heartbeat到jobtracker，并且tasktracker运行的所有状态都会在调用中被发送到jobtracker。
6jobtracker合并各任务报告：产生一个表明所有运行作业机器所含任务状态的全局视图。
前面提到的JobClient就是通过每秒查询JobTracker来接收最新状态，而且客户端JobClient的getJob方法可以得到一个RunningJob的实例，其包含了作业的所以状态信息。
作业的完成：
当jobtracker收到作业最后一个任务已完成的通知后，便把作业状态设置成成功。JobClient查询状态时，便知道任务已成功完成，于是JobClient打印一条消息告知用户，然后从runJob方法返回。
如果jobtracker有相应设置，也会发送一个>
jobtracker情况作业的工作状态，指示tasktracker也清空作业的工作状态，如删除中间输出。
失败
实际情况下，用户的代码存在软件错误进程会崩溃，机器也会产生故障，但Hadoop能很好的应对这些故障并完成作业。
1任务失败
子任务异常：如Map/Reduce任务中的用户代码抛出异常，子任务JVM进程会在退出前向父进程tasktracker发送错误报告，错误被记录用户日志。tasktracker会将此次task attempt标记为tailed，并释放这个任务槽运行另外一个任务。
子进程JVM突然退出：可能由于JVM bug导致用户代码造成的某些特殊原因导致JVM退出，这种情况下，tasktracker会注意到进程已经退出，并将此次尝试标记为failed。
任务挂起：一旦tasktracker注意一段时间没有收到进度更新，便会将任务标记为failed，JVM子进程将被自动杀死。任务失败间隔时间通常为10分钟，可以以作业或者集群为基础设置过期时间，参数为mapredtasktimeout。注意：如果参数值设置为0，则挂起的任务永远不会释放掉它的任务槽，随着时间的推移会降低整个集群的效率。
任务失败尝试次数：jobtracker得知一个tasktracker失败后，它会重新调度该任务执行，当然，jobtracker会尝试避免重新调度失败过的tasktracker任务。如果一个任务尝试次数超过4次，它将不再被重试。这个值是可以设置的，对于Map任务，参数是mapredmapmaxattempts,对于reduce任务，则由mapredreducemaxattempts属性控制。如果次数超过限制，整个作业都会失败。当然，有时我们不希望少数几个任务失败就终止运行的整个作业，因为即使有些任务失败，作业的一些结果可能还是有用的，这种情况下，可以为作业设置在不触发作业失败情况下的允许任务失败的最大百分比，Map任务和Reduce任务可以独立控制，参数为mapredmaxmapfailurespercent 和mapredmaxreducefailurespercent。
任务尝试中止（kill）：任务终止和任务失败不同，task attempt可以中止是因为他是一个推测副本或因为它所处的tasktracker失败，导致jobtracker将它上面的所有task attempt标记为killed。被终止的task attempt不会被计入任务运行尝试次数，因为尝试中止并不是任务的错。
2tasktracker失败
tasktracker由于崩溃或者运行过慢而失败，他将停止向jobtracker发送心跳（或很少发送心跳）。jobtracker注意已停止发送心跳的tasktracker（过期时间由参数mapredtasktrackerexpiryinterval设置，单位毫秒），并将它从等待调度的tasktracker池中移除。如果是未完成的作业，jobtracker会安排次tasktracker上已经运行成功的Map任务重新运行，因为此时reduce任务已无法访问（中间输出存放在失败的tasktracker的本地文件系统上）。
即使tasktracker没有失败，也有可能被jobtracker列入黑名单。如果tasktracker上面的失败任务数量远远高于集群的平均失败任务次数，他就会被列入黑名单，被列入黑名单的tasktracker可以通过重启从jobtracker黑名单中移除。
3jobtracker失败
老版本的JobTracker失败属于单点故障，这种情况下作业注定失败。
作业调度：
早期作业调度FIFO：按作业提交顺序先进先出。可以设置优先级，通过设置mapredjobpriority属性或者JobClient的setJobPriority()方法制定优先级（优先级别：VERY_HIGH,HIGH,NORMAL,LOW,VERY_LOW）。注意FIFO调度算法不支持抢占（preemption）,所以高优先级作业仍然会被那些已经开始的长时间运行的低优先级作业所阻塞。
Fair Scheduler：目标是让每个用户公平地共享集群能力。当集群存在很多作业时，空闲的任务槽会以”让每个用户共享集群“的方式进行分配。默认每个用户都有自己的作业池。FairScheduler支持抢占，所以，如果一个池在特定的一段时间未得到公平地资源共享，它会终止池中得到过多的资源任务，以便把任务槽让给资源不足的池。FairScheduler是一个后续模块，使用它需要将其jar文件放在Hadoop的类路径下。可以通过参数mapredjobtrackertaskScheduler属性配置（值为orgapachehadoopmapredFairScheduler）
Capacity Scheduler：
集群由很多队列组成，每个队列都有一个分配能力，这一点与FairScheduler类似，只不过在每个队列内部，作业根据FIFO方式进行调度。本质上说，Capacity Scheduler允许用户或组织为每个用户模拟一个独立使用FIFO的集群。
shuffle和排序：
MapReduce确保每个Reducer的输入都是按键排序的。系统执行排序的过程-将map输出作为输入传给reducer的过程称为shuffle。shuffle属于不断被优化和改进的代码库的一部分，从许多方面来看，shuffle是MapReduce的心脏。
整个shuffle的流程应该是这样：
map结果划分partition 排序sort 分割spill 合并同一划分 合并同一划分 合并结果排序 reduce处理 输出
Map端：
写入缓冲区：Map函数的输出，是由collector处理的，它并不是简单的将结果写到磁盘。它利用缓冲的方式写到内存，并处于效率的考虑进行预排序。每个map都有一个环形的内存缓冲区，用于任务输出，默认缓冲区大小为100MB（由参数iosortmb调整），一旦缓冲区内容达到阈值（默认08），后台进程边开始把内容写到磁盘（spill），在写磁盘过程中，map输出继续被写到缓冲区，但如果缓冲区被填满，map会阻塞知道写磁盘过程完成。写磁盘将按照轮询方式写到mapredlocaldir属性制定的作业特定子目录中。
写出缓冲区：collect将缓冲区的内容写出时，会调用sortAndSpill函数，这个函数作用主要是创建spill文件，按照key值对数据进行排序，按照划分将数据写入文件，如果配置了combiner类，会先调用combineAndSpill函数再写文件。sortAndSpill每被调用一次，就会写一个spill文件。
合并所有Map的spill文件：TaskTracker会在每个map任务结束后对所有map产生的spill文件进行merge，merge规则是根据分区将各个spill文件中数据同一分区中的数据合并在一起，并写入到一个已分区且排序的map输出文件中。待唯一的已分区且已排序的map输出文件写入最后一条记录后，map端的shuffle阶段就结束了。
在写磁盘前，线程首先根据数据最终要传递到的reducer把数据划分成响应的分区（partition），在每个分区中，后台线程按键进行内排序，如果有一个combiner,它会在排序后的输出上运行。
内存达到溢出写的阈值时，就会新建一个溢出写文件，因为map任务完成其最后一个输出记录之后，会有几个溢出写文件。在任务完成前，溢出写文件会被合并成一个已分区且已排序的输出文件。配置属性iosortfacor控制一次最多能合并多少流，默认值是10。
如果已经指定combiner,并且写次数至少为3（通过minmumspillsforcombine设置）时，则combiner就会在输出文件写到磁盘之前运行。运行combiner的意义在于使map输出更紧凑，舍得写到本地磁盘和传给reducer的数据更少。
写磁盘时压缩：写磁盘时压缩会让写的速度更快，节约磁盘空间，并且减少传给reducer的数据量。默认情况下，输出是不压缩的，但可以通过设置mapredcompressmapoutput值为true，就可以启用压缩。使用的压缩库是由mapredmapoutputcompressioncodec制定。
reducer获得文件分区的工作线程：reducer通过>
Reduce端：
复制阶段：reduce会定期向JobTracker获取map的输出位置，一旦拿到输出位置，reduce就会从对应的TaskTracker上复制map输出到本地（如果map输出很小，则会被复制到TaskTracker节点的内存中，否则会被让如磁盘），而不会等到所有map任务结束（当然这个也有参数控制）。
合并阶段：从各个TaskTracker上复制的map输出文件（无论在磁盘还是内存）进行整合，并维持数据原来的顺序。
Reduce阶段：从合并的文件中顺序拿出一条数据进行reduce函数处理，然后将结果输出到本地HDFS。
Map的输出文件位于运行map任务的tasktracker的本地磁盘，现在，tasktracker要为分区文件运行reduce任务。每个任务完成时间可能不同，但是只要有一个任务完成，reduce任务就开始复制其输出，这就是reduce任务的复制阶段（copy phase）。reduce任务有少量复制线程，因此能够并行取得map输出。默认值是5个线程，可以通过mapredreduceparallelcopies属性设置。
Reducer如何得知从哪个tasktracker获得map输出：map任务完成后会通知其父tasktracker状态已更新，tasktracker进而通知（通过heart beat）jobtracker。因此，JobTracker就知道map输出和tasktracker之间的映射关系，reducer中的一个线程定期询问jobtracker以便获知map输出位置。由于reducer有可能失败，因此tasktracker并没有在第一个reducer检索到map输出时就立即从磁盘上删除它们，相反他会等待jobtracker告示它可以删除map输出时才删除，这是作业完成后最后执行的。
如果map输出很小，则会被直接复制到reduce tasktracker的内存缓冲区（大小由mapredjobshuffleinputbufferpercent控制，占堆空间的百分比），否则，map输出被复制到磁盘。一旦内存缓冲区达到阈值大小（由maprediobshufflemergepercent）
或达到map输出阈值大小（mapredinmemthreadhold），则合并后溢出写到磁盘中。
随着磁盘上副本增多，后台线程会将他们合并为更大的、排好序的文件。注意：为了合并，压缩的map输出必须在内存中被解压缩。
排序阶段：复制阶段完成后，reduce任务会进入排序阶段，更确切的说是合并阶段，这个阶段将合并map输出，维持其顺序排列。合并是循环进行的，由合并因子决定每次合并的输出文件数量。但让有可能会产生中间文件。
reduce阶段：在最后reduce阶段，会直接把排序好的文件输入reduce函数，不会对中间文件进行再合并，最后的合并即可来自内存，也可来自磁盘。此阶段的输出会直接写到文件系统，一般为hdfs。
细节：这里合并是并非平均合并，比如有40个文件，合并因子为10，我们并不是每趟合并10个，合并四趟。而是第一趟合并4个，后三趟合并10，在最后一趟中4个已合并的文件和余下6个未合并会直接并入reduce。
欢迎分享，转载请注明来源：内存溢出
微信扫一扫
支付宝扫一扫
评论列表（0条）