分布式计算框架Hadoop

时间:2022-10-30 03:29:09

1 Hadoop是什么?

Hadoop是一种开源的适合大数据的分布式存储和处理的平台。

Hadoop 是一个由Apache基金会所开发的分布式系统基础架构。用户可以在不了解分布式底层细节的情况下,开发分布式程序。充分利用集群的威力进行高速运算和存储。

Apache Hadoop是一款支持数据密集型分布式应用并以Apache 2.0许可协议发布的开源软件框架。它支持在商品硬件构建的大型集群上运行的应用程序。Hadoop是根据Google公司发表的MapReduce和Google档案系统的论文自行实作而成。

Hadoop框架透明地为应用提供可靠性和数据移动。它实现了名为MapReduce的编程范式:应用程序被分割成许多小部分,而每个部分都能在集群中的任意节点上执行或重新执行。此外,Hadoop还提供了分布式文件系统,用以存储所有计算节点的数据,这为整个集群带来了非常高的带宽。MapReduce和分布式文件系统的设计,使得整个框架能够自动处理节点故障。它使应用程序与成千上万的独立计算的电脑和PB级的数据。 

 

2 Hadoop能够做什么?

1)搜索引擎:这也正是Doug Cutting设计Hadoop的初衷,为了针对大规模的网页快速建立索引;

2)大数据存储:利用Hadoop的分布式存储能力,例如数据备份、数据仓库等;

3)大数据处理:利用Hadoop的分布式处理能力,例如数据挖掘、数据分析等;

4)科学研究:Hadoop是一种分布式的开源框架,对于分布式系统有很大程度地参考价值。

 

3 Hadoop的三种模式及子项目

Hadoop有三种不同的模式操作,分别为单机模式、伪分布模式和全分布模式。

 

子项目:

l  Hadoop Common:在0.20及以前的版本中,包含HDFS、MapReduce和其他项目公共内容,从            0.21开始HDFS和MapReduce被分离为独立的子项目,其余内容为Hadoop Common

l  HDFS:Hadoop分布式文件系统(Distributed File System)-HDFS(Hadoop Distributed File System)

l  MapReduce:并行计算框架,0.20前使用org.apache.hadoop.mapred旧接口,0.20版本开始引入org.apache.hadoop.mapreduce的新API

l  Apache HBase:分布式NoSQL列数据库,类似谷歌公司BigTable。

l  Apache Hive:构建于hadoop之上的数据仓库,通过一种类SQL语言HiveQL为用户提供数据的归纳、查询和分析等功能。Hive最初由Facebook贡献。

l  Apache Mahout:机器学习算法软件包。

l  Apache Sqoop:结构化数据(如关系数据库)与Apache Hadoop之间的数据转换工具。

l  Apache ZooKeeper:分布式锁设施,提供类似Google Chubby的功能,由Facebook贡献。

l  Apache Avro:新的数据序列化格式与传输工具,将逐步取代Hadoop原有的IPC机制。

l   

4 Hadoop核心之分布式文件系统HDFS

源自于Google的论文,发表于2003年10月,HDFS是GFS的克隆。

Hadoop分布式文件系统(Hadoop Distributed File System,简称HDFS)是Hadoop的核心模块之一,它主要解决Hadoop的大数据存储问题,其思想来源与Google的文件系统GFS。HDFS有高容错性的特点,并且设计用来部署在低廉的(low-cost)硬件上;而且它提供高吞吐量(high throughput)来访问应用程序的数据,适合那些有着超大数据集(large data set)的应用程序。HDFS的主要特点:

l  保存多个副本,且提供容错机制,副本丢失或宕机自动恢复。默认存3份。

l  运行在廉价的机器上。

l  适合大数据的处理。HDFS默认会将文件分割成block,64M为1个block。然后将block按键值对存储在HDFS上,并将键值对的映射存到内存中。如果小文件太多,那内存的负担会很重。

HDFS中的两个重要角色:

[Namenode]

1)管理文件系统的命名空间。

2)记录 每个文件数据快在各个Datanode上的位置和副本信息。

3)协调客户端对文件的访问。

4)记录命名空间内的改动或者空间本省属性的改动。

5)Namenode 使用事务日志记录HDFS元数据的变化。使用映像文件存储文件系统的命名空间,包括文件映射,文件属性等。

从社会学来看,Namenode是HDFS里面的管理者,发挥者管理、协调、操控的作用。

[Datanode]

1)负责所在物理节点的存储管理。

2)一次写入,多次读取(不修改)。

3)文件由数据库组成,一般情况下,数据块的大小为64MB。

4)数据尽量散步到各个节点。

从社会学的角度来看,Datanode是HDFS的工作者,发挥按着Namenode的命令干活,并且把干活的进展和问题反馈到Namenode的作用。

客户端如何访问HDFS中一个文件呢?具体流程如下:

1)首先从Namenode获得组成这个文件的数据块位置列表。

2)接下来根据位置列表知道存储数据块的Datanode。

3)最后访问Datanode获取数据。

注意:Namenode并不参与数据实际传输。

数据存储系统,数据存储的可靠性至关重要。HDFS是如何保证其可靠性呢?它主要采用如下机理:

1)冗余副本策略,即所有数据都有副本,副本的数目可以在hdfs-site.xml中设置相应的复制因子。

2)机架策略,即HDFS的“机架感知”,一般在本机架存放一个副本,在其它机架再存放别的副本,这样可以防止机架失效时丢失数据,也可以提供带宽利用率。

3)心跳机制,即Namenode周期性从Datanode接受心跳信号和快报告,没有按时发送心跳的Datanode会被标记为宕机,不会再给任何I/O请求,若是Datanode失效造成副本数量下降,并且低于预先设置的阈值,Namenode会检测出这些数据块,并在合适的时机进行重新复制。

4)安全模式,Namenode启动时会先经过一个“安全模式”阶段。

5)校验和,客户端获取数据通过检查校验和,发现数据块是否损坏,从而确定是否要读取副本。

6)回收站,删除文件,会先到回收站/trash,其里面文件可以快速回复。

7)元数据保护,映像文件和事务日志是Namenode的核心数据,可以配置为拥有多个副本。

8)快照,支持存储某个时间点的映像,需要时可以使数据重返这个时间点的状态。

 

如上图所示,HDFS也是按照Master和Slave的结构。分NameNode、SecondaryNameNode、DataNode这几个角色。

NameNode:是Master节点,是大领导。管理数据块映射;处理客户端的读写请求;配置副本策略;管理HDFS的名称空间;

SecondaryNameNode:是一个小弟,分担大哥namenode的工作量;是NameNode的冷备份;合并fsimage和fsedits然后再发给namenode。

DataNode:Slave节点,奴隶,干活的。负责存储client发来的数据块block;执行数据块的读写操作。

热备份:b是a的热备份,如果a坏掉。那么b马上运行代替a的工作。

冷备份:b是a的冷备份,如果a坏掉。那么b不能马上代替a工作。但是b上存储a的一些信息,减少a坏掉之后的损失。

fsimage:元数据镜像文件(文件系统的目录树。)

edits:元数据的操作日志(针对文件系统做的修改操作记录)

namenode内存中存储的是=fsimage+edits。

SecondaryNameNode负责定时默认1小时,从namenode上,获取fsimage和edits来进行合并,然后再发送给namenode。减少namenode的工作量。

文件写入时:

Client向NameNode发起文件写入的请求。

NameNode根据文件大小和文件块配置情况,返回给Client它所管理部分DataNode的信息。

Client将文件划分为多个block块,并根据DataNode的地址信息,按顺序写入到每一个DataNode块中。

当文件读取:

Client向NameNode发起文件读取的请求。

NameNode返回文件存储的block块信息、及其block块所在DataNode的信息。

Client读取文件信息。

HDFS 数据备份

HDFS被设计成一个可以在大集群中、跨机器、可靠的存储海量数据的框架。它将所有文件存储成block块组成的序列,除了最后一个block块,所有的block块大小都是一样的。文件的所有block块都会因为容错而被复制。每个文件的block块大小和容错复制份数都是可配置的。容错复制份数可以在文件创建时配置,后期也可以修改。HDFS中的文件默认规则是write one(一次写、多次读)的,并且严格要求在任何时候只有一个writer。NameNode负责管理block块的复制,它周期性地接收集群中所有DataNode的心跳数据包和Blockreport。心跳包表示DataNode正常工作,Blockreport描述了该DataNode上所有的block组成的列表。

备份数据的存放:

备份数据的存放是HDFS可靠性和性能的关键。HDFS采用一种称为rack-aware的策略来决定备份数据的存放。通过一个称为Rack Awareness的过程,NameNode决定每个DataNode所属rack id。缺省情况下,一个block块会有三个备份,一个在NameNode指定的DataNode上,一个在指定DataNode非同一rack的DataNode上,一个在指定DataNode同一rack的DataNode上。这种策略综合考虑了同一rack失效、以及不同rack之间数据复制性能问题。

副本的选择:

为了降低整体的带宽消耗和读取延时,HDFS会尽量读取最近的副本。如果在同一个rack上有一个副本,那么就读该副本。如果一个HDFS集群跨越多个数据中心,那么将首先尝试读本地数据中心的副本。

安全模式:

系统启动后先进入安全模式,此时系统中的内容不允许修改和删除,直到安全模式结束。安全模式主要是为了启动检查各个DataNode上数据块的安全性。

 

5 Hadoop核心之MapReduce(分布式计算框架)

源自于Google的MapReduce论文,发表于2004年12月,Hadoop MapReduce是Google MapReduce的克隆版。

MapReduce是一种计算模型,用以进行大数据量的计算。其中Map对数据集上的独立元素进行指定的操作,生成键-值对形式中间结果。Reduce则对中间结果中相同“键”的所有“值”进行规约,以得到最终结果。MapReduce这样的功能划分,非常适合在大量计算机组成的分布式并行环境里进行数据处理。

MapReduce的思想就是“分而治之”。Mapper负责“分”,即把复杂的任务分解为若干个“简单的任务”来处理。“简单的任务”包含三层含义:一是数据或计算的规模相对原任务要大大缩小;二是就近计算原则,即任务会分配到存放着所需数据的节点上进行计算;三是这些小任务可以并行计算,彼此间几乎没有依赖关系。Reducer负责对map阶段的结果进行汇总。至于需要多少个Reducer,用户可以根据具体问题,通过在mapred-site.xml配置文件里设置参数mapred.reduce.tasks的值,缺省值为1。

 

JobTracker:Master节点,只有一个,管理所有作业,作业/任务的监控、错误处理等;将任务分解成一系列任务,并分派给TaskTracker。

TaskTracker:Slave节点,运行Map Task和Reduce Task;并与JobTracker交互,汇报任务状态。

Map Task:解析每条数据记录,传递给用户编写的map(),并执行,将输出结果写入本地磁盘(如果为map-only作业,直接写入HDFS)。

Reducer Task:从Map Task的执行结果中,远程读取输入数据,对数据进行排序,将数据按照分组传递给用户编写的reduce函数执行。

Mapreduce处理流程,以wordCount为例:

 

6 Hive(基于Hadoop的数据仓库)

由facebook开源,最初用于解决海量结构化的日志数据统计问题。

Hive定义了一种类似SQL的查询语言(HQL),将SQL转化为MapReduce任务在Hadoop上执行。

通常用于离线分析。

 

7、Hbase(分布式列存数据库)

源自Google的Bigtable论文,发表于2006年11月,HBase是Google Bigtable克隆版。

HBase是一个针对结构化数据的可伸缩、高可靠、高性能、分布式和面向列的动态模式数据库。和传统关系数据库不同,HBase采用了BigTable的数据模型:增强的稀疏排序映射表(Key/Value),其中,键由行关键字、列关键字和时间戳构成。HBase提供了对大规模数据的随机、实时读写访问,同时,HBase中保存的数据可以使用MapReduce来处理,它将数据存储和并行计算完美地结合在一起。

数据模型:Schema-->Table-->Column Family-->Column-->RowKey-->TimeStamp-->Value

 

8、Zookeeper(分布式协作服务)

源自Google的Chubby论文,发表于2006年11月,Zookeeper是Chubby克隆版

解决分布式环境下的数据管理问题:统一命名,状态同步,集群管理,配置同步等。

 

9、Sqoop(数据同步工具)

Sqoop是SQL-to-Hadoop的缩写,主要用于传统数据库和Hadoop之前传输数据。

数据的导入和导出本质上是Mapreduce程序,充分利用了MR的并行化和容错性。

 

10、Pig(基于Hadoop的数据流系统)

Pig是一种数据流语言,用来快速轻松的处理巨大的数据。Pig包含两个部分:Pig Interface,Pig Latin。Pig可以非常方便的处理HDFS和HBase的数据,和Hive一样,Pig可以非常高效的处理其需要做的,通过直接操作Pig查询可以节省大量的劳动和时间。

当你想在你的数据上做一些转换,并且不想编写MapReduce jobs就可以用Pig.。

通常用于进行离线分析。

 

11 storm(基于Hadoop的数据流框架)

分布式实时计算系统,Storm是一个任务并行连续计算引擎。 Storm本身并不在Hadoop集群上运行,它使用Apache ZooKeeper的和自己的主/从工作进程,协调拓扑,主机和工作者状态,保证信息的语义。无论如何, Storm必定还是可以从HDFS文件读取或者从文件写入到HDFS。

 

12 spark(基于Hadoop的数据流框架)

一种快速,通用引擎用于大规模数据处理,Spark是一个数据并行通用批量处理引擎。工作流中在一个类似的和怀旧风格的MapReduce中定义,但是,比传统Hadoop MapReduce的更能干。Apache Spark有其流API项目,该项目通过短间隔批次允许连续处理。Apache Spark本身并不需要Hadoop操作。但是,它的数据并行模式,需要稳定的数据优化使用共享文件系统。该稳定源的范围可以从S3,NFS或更典型地,HDFS。执行Spark应用程序并不需要Hadoop YARN。Spark有自己独立的主/服务器进程。然而,这是共同的运行使用YARN容器Spark的应用程序。此外,Spark还可以在Mesos集群上运行。

 

13、Mahout(数据挖掘算法库)

Mahout起源于2008年,最初是Apache Lucent的子项目,它在极短的时间内取得了长足的发展,现在是Apache的*项目。

Mahout的主要目标是创建一些可扩展的机器学习领域经典算法的实现,旨在帮助开发人员更加方便快捷地创建智能应用程序。Mahout现在已经包含了聚类、分类、推荐引擎(协同过滤)和频繁集挖掘等广泛使用的数据挖掘方法。除了算法,Mahout还包含数据的输入/输出工具、与其他存储系统(如数据库、MongoDB 或Cassandra)集成等数据挖掘支持架构。

 

14、Flume(日志收集工具)

Cloudera开源的日志收集系统,具有分布式、高可靠、高容错、易于定制和扩展的特点。

它将数据从产生、传输、处理并最终写入目标的路径的过程抽象为数据流,在具体的数据流中,数据源支持在Flume中定制数据发送方,从而支持收集各种不同协议数据。同时,Flume数据流提供对日志数据进行简单处理的能力,如过滤、格式转换等。此外,Flume还具有能够将日志写往各种数据目标(可定制)的能力。总的来说,Flume是一个可扩展、适合复杂环境的海量日志收集系统。

 

13、MapReduce作业运行流程

 

1.在客户端启动一个作业。

2.向JobTracker请求一个Job ID。

3.将运行作业所需要的资源文件复制到HDFS上,包括MapReduce程序打包的JAR文件、配置文件和客户端计算所得的输入划分信息。这些文件都存放在JobTracker专门为该作业创建的文件夹中。文件夹名为该作业的Job ID。JAR文件默认会有10个副本(mapred.submit.replication属性控制);输入划分信息告诉了JobTracker应该为这个作业启动多少个map任务等信息。

4.JobTracker接收到作业后,将其放在一个作业队列里,等待作业调度器对其进行调度,当作业调度器根据自己的调度算法调度到该作业时,会根据输入划分信息为每个划分创建一个map任务,并将map任务分配给TaskTracker执行。对于map和reduce任务,TaskTracker根据主机核的数量和内存的大小有固定数量的map槽和reduce槽。这里需要强调的是:map任务不是随随便便地分配给某个TaskTracker的,这里有个概念叫:数据本地化(Data-Local)。意思是:将map任务分配给含有该map处理的数据块的TaskTracker上,同时将程序JAR包复制到该TaskTracker上来运行,这叫“运算移动,数据不移动”。而分配reduce任务时并不考虑数据本地化。

5.TaskTracker每隔一段时间会给JobTracker发送一个心跳,告诉JobTracker它依然在运行,同时心跳中还携带着很多的信息,比如当前map任务完成的进度等信息。当JobTracker收到作业的最后一个任务完成信息时,便把该作业设置成“成功”。当JobClient查询状态时,它将得知任务已完成,便显示一条消息给用户。

 

15、Map、 Reduce任务中Shuffle和排序过程

 

Map端: 

1.每个输入分片会让一个map任务来处理,默认情况下,以HDFS的一个块的大小(默认为64M)为一个分片,当然我们也可以设置块的大小。map输出的结果会暂且放在一个环形内存缓冲区中(该缓冲区的大小默认为100M,由io.sort.mb属性控制),当该缓冲区快要溢出时(默认为缓冲区大小的80%,由io.sort.spill.percent属性控制),会在本地文件系统中创建一个溢出文件,将该缓冲区中的数据写入这个文件。

2.在写入磁盘之前,线程首先根据reduce任务的数目将数据划分为相同数目的分区,也就是一个reduce任务对应一个分区的数据。这样做是为了避免有些reduce任务分配到大量数据,而有些reduce任务却分到很少数据,甚至没有分到数据的尴尬局面。其实分区就是对数据进行hash的过程。然后对每个分区中的数据进行排序,如果此时设置了Combiner,将排序后的结果进行Combia操作,这样做的目的是让尽可能少的数据写入到磁盘。

3.当map任务输出最后一个记录时,可能会有很多的溢出文件,这时需要将这些文件合并。合并的过程中会不断地进行排序和combia操作,目的有两个:1.尽量减少每次写入磁盘的数据量;2.尽量减少下一复制阶段网络传输的数据量。最后合并成了一个已分区且已排序的文件。为了减少网络传输的数据量,这里可以将数据压缩,只要将mapred.compress.map.out设置为true就可以了。

4.将分区中的数据拷贝给相对应的reduce任务。有人可能会问:分区中的数据怎么知道它对应的reduce是哪个呢?其实map任务一直和其父TaskTracker保持联系,而TaskTracker又一直和JobTracker保持心跳。所以JobTracker中保存了整个集群中的宏观信息。只要reduce任务向JobTracker获取对应的map输出位置就ok了哦。

到这里,map端就分析完了。那到底什么是Shuffle呢?Shuffle的中文意思是“洗牌”,如果我们这样看:一个map产生的数据,结果通过hash过程分区却分配给了不同的reduce任务,是不是一个对数据洗牌的过程呢?

 

Reduce端: 

1.Reduce会接收到不同map任务传来的数据,并且每个map传来的数据都是有序的。如果reduce端接受的数据量相当小,则直接存储在内存中(缓冲区大小由mapred.job.shuffle.input.buffer.percent属性控制,表示用作此用途的堆空间的百分比),如果数据量超过了该缓冲区大小的一定比例(由mapred.job.shuffle.merge.percent决定),则对数据合并后溢写到磁盘中。

2.随着溢写文件的增多,后台线程会将它们合并成一个更大的有序的文件,这样做是为了给后面的合并节省时间。其实不管在map端还是reduce端,MapReduce都是反复地执行排序,合并操作,现在终于明白了有些人为什么会说:排序是hadoop的灵魂。

3.合并的过程中会产生许多的中间文件(写入磁盘了),但MapReduce会让写入磁盘的数据尽可能地少,并且最后一次合并的结果并没有写入磁盘,而是直接输入到reduce函数。

 

16 Google三驾马车:GFS、MapReduce和Bigtable

MapReduce模仿了Google的MapReduce, 

HDFS模仿了Google File System, 

HBase模仿了BigTable,

HDFS和HBase是依靠外存(即硬盘)的存储模型和实现。HDFS是一个云存储的文件系统,它会把一个文件分块并分别保存,取用时分别取出再合并。重要的是,这些分块通常会在3台节点(即机群内的电脑)上有3个备份,所以即使出现少数电脑的失效(硬盘损坏、掉电等),文件也不会失效。如果说HDFS是文件级别的存储,那HBase则是表级别的存储。HBase是一个表模型,但比SQL数据库的表要简单的多,没有连接、聚集等功能。HBase表的物理存储是依赖HDFS的,比如把一个表分成4个文件,存到HDFS中。由于HDFS级会做备份,所以HBase级不再备份。

MapReduce则是一个计算模型,而不是存储模型;MapReduce与HDFS紧密配合,而非HBase。举个场景:如果你的手机通话信息保存在一个HDFS的文件callList.txt中,你想找到你与你同事A的所有通话记录并排序。因为HDFS会把callLst.txt分成几块分别存,比如说5块,那么对应的Map过程就是找到这5块所在的5台节点,让他们分别找自己存的那块中关于A的通话记录,对应的Reduce过程就是把5个节点过滤过的通话记录合并在一块并按时间排序。可见MapReduce的计算模型需要HDFS,但与HBase没有任何关系。

ZooKeeper本身是一个非常牢靠的记事本,最好用于记录一些概要信息。Hadoop依靠这个记事本来记录当前哪些节点正在用,哪些已掉线,哪些还备用等,以此来管理机群。

BigTable与GFS的关系: 

     GFS是分布式文件系统,BigTable 是建立在GFS之上的。就像文件系统需要数据库来存储结构化数据一样,GFS也需要Bigtable来存储结构化数据,每个Table都是一个多维的稀疏图,为了管理巨大的Table,把Table根据行分割,这些分割后的数据统称为:Tablets。每个Tablets大概有 100-200 MB,每个机器存储100个左右的 Tablets。底层的架构是:GFS。由于GFS是一种分布式的文件系统,采用Tablets的机制后,可以获得很好的负载均衡。比如:可以把经常响应的表移动到其他空闲机器上,然后快速重建。

BigTable是一个key/value分布式数据库,而GFS主要是一个分布式文件系统。bigTable存储结构化数据,GFS存储的是非结构化数据。另外BigTable的数据是存储在GFS上的。