大数据存储与处理第三讲

1、本讲内容原理Hadoop 基本架构HDFS MapReduce 分布式计算框架实践安装与配置编写MapReduce介绍背景Hadoop起源成功的案例1你发现：目前的硬盘容量1T，速度100MB/s 。你发现：要把现在的硬盘数据刷一遍，要两个多小时，太可怕了。设想：硬盘读取速度永远满足要求！2而Hadoop就是一个并行处理海量数据的工具。我们只能采用并行计算3一个分布式文件系统和并行执行环境（框架）让用户便捷地处理海量数据Apache软件基金会下面的一个开源项目目前Yahoo!是最主要的贡献者Hadoop是什么？4Apache Lucene开源的高性能全文检索工具包Apache Nutch开源的

2、 Web 搜索引擎Google 三大论文MapReduce / GFS / BigTableApache Hadoop大规模数据处理Hadoop起源5扩容能力（Scalable）：能可靠地（reliably）存储和处理千兆字节（PB）数据。成本低（Economical）：可以通过普通机器组成的服务器群来分发以及处理数据。这些服务器群总计可达数千个节点。高效率（Efficient）：通过分发数据，hadoop可以在数据所在的节点上并行地（parallel）处理它们，这使得处理非常的快速。可靠性（Reliable）：hadoop能自动地维护数据的多份复制，并且在任务失败后能自动地重新部署（rede

3、ploy）计算任务。Hadoop的特点6Hadoop越来越火据估计，到2015年，全世界一半以上的数据将涉及Hadoop那我们就来看看一些成功的案例：不是可以解决数据中心和数据管理方面所有难题的灵丹妙药7雅虎北京全球软件研发中心IBMFacebookAmazonYahoo!中国移动研究院英特尔研究院百度、腾讯、新浪、搜狐、淘宝谁在用Hadoop？8Yahoo Hadoop应用揭秘中国移动联通海量上网日志数据Hadoop应用案例92010年10月时：Yahoo目前有超过38000台服务器，有超过4000个以上的服务器集群，数据总量达到了170PB，每日的数据增量在10TB以上。Yahoo Had

4、oop应用揭秘10Yahoo的Hadoop应用包含有搜索、日志处理（Analytics, Reporting, Buzz）、用户建模、内容优化，垃圾邮件过滤器以及广告计算等。如何利用Hadoop对海量数据进行优化处理是Yahoo正在致力于工作的内容。以网络分析为例，Yahoo目前有超过100亿个网页，1PB的网页数据内容，2万亿条链接，每日面临这300TB的数据输出。“在应用Hadoop前，实施这一过程我们大概需要1个月的时间，但应用后仅需要1周时间”Yahoo Hadoop应用揭秘11“再以Yahoo搜索为例，我们的服务器上保留有用户三年来的搜索记录，这个数据是由超过 10TB的自然语言文本

5、库所组成”，“如果数据重整，我们在应用Hadoop前需要1个月的时间进行处理，而在有了Hadoop后仅仅需要30分钟。”Hadoop也非万能，它采用Java实现，Java的IO处理虽然没有性能瓶颈，但是对于CPU密集型的任务是一个麻烦， 因此，有些算法效率不会提高很多。Yahoo Hadoop应用揭秘12中国移动大云在中国移动内部，Hadoop已经成为一个重要的数据挖掘工具，中国移动已经把Hadoop的群用来进行闭环的数据挖掘，和传统算法的数据挖掘性能相比，有很大的提升，而且成本非常低。中国移动13Hadoop生态系统HDFSMapReduceHivePigHbaseZookeeper14Ha

6、doop基本架构物理分布的Hadoop集群15Hadoop基本架构物理部署16为以流式数据访问模式存储超大文件而设计的文件系统超大文件指的是几百MB，几百GB，几百TB，甚至几百PB流式数据访问HDFS建立的思想是：一次写入、多次读取模式是最高效的。商用硬件HDFS不需要运行在昂贵并且高可靠的硬件上。HDFS文件系统设计目标17硬件错误是常态，因此需要冗余。程序采用“数据就近”原则分配节点执行。HDFS文件系统设计基础与目标18存储并管理PB级数据处理非结构化数据注重数据处理的吞吐量（延迟不敏感）应用模式为：write-once-read-many存取模式（无数据一致性问题）HDFS文件系统能

7、做什么？19存储小文件 (不建议使用)大量的随机读 (不建议使用)需要对文件的修改 (不支持)多用户写入(不支持)HDFS文件系统不适合做什么？20文件切分成块（默认大小64M），以块为单位，每个块有多个副本存储在不同的机器上，副本数可在文件生成时指定（默认3）NameNode是主节点，存储文件的元数据如文件名，文件目录结构，文件属性（生成时间,副本数,文件权限），以及每个文件的块列表以及块所在的DataNode等等DataNode在本地文件系统存储文件块数据，以及块数据的校验和可以创建、删除、移动或重命名文件，当文件创建、写入和关闭之后不能修改文件内容。HDFS文件系统文件21HDFS文件系

8、统文件22HDFS文件系统文件23HDFS文件系统主要组件的功能namenodedatanode 存储元数据 存储文件内容元数据保存在内存中文件内容保存在磁盘 保存文件,block ，datanode之间的映射关系维护了block id到datanode本地文件的映射关系24Namenode是一个中心服务器，单一节点（简化系统的设计和实现），负责管理文件系统的名字空间(namespace)以及客户端对文件的访问。文件操作，NameNode负责文件元数据的操作，DataNode负责处理文件内容的读写请求，跟文件内容相关的数据流不经过NameNode，只会询问它跟那个DataNode联系，否则Na

9、meNode会成为系统的瓶颈HDFS文件系统 NameNode25副本存放在那些DataNode上由NameNode来控制，根据全局情况做出块放置决定，读取文件时NameNode尽量让用户先读取最近的副本，降低带块消耗和读取时延Namenode全权管理数据块的复制，它周期性地从集群中的每个Datanode接收心跳信号和块状态报告(Blockreport)。接收到心跳信号意味着该Datanode节点工作正常。块状态报告包含了一个该Datanode上所有数据块的列表。 HDFS文件系统 NameNode26一个数据块在DataNode以文件存储在磁盘上，包括两个文件，一个是数据本身，一个是元数据包

10、括数据块的长度，块数据的校验和，以及时间戳DataNode启动后向NameNode注册，通过后，周期性（1小时）的向NameNode上报所有的块信息。 心跳是每3秒一次，心跳返回结果带有NameNode给该DataNode的命令如复制块数据到另一台机器，或删除某个数据块。如果超过10分钟没有收到某个DataNode 的心跳，则认为该节点不可用。集群运行中可以安全加入和退出一些机器HDFS文件系统 DataNode27HDFS文件系统系统架构28一个名字节点和多个数据节点数据复制（冗余机制）存放的位置（机架感知策略）故障检测数据节点心跳包（检测是否宕机）块报告（安全模式下检测）数据完整性检测（校

11、验和比较）名字节点（日志文件，镜像文件）空间回收机制HDFS文件系统保障可靠性的措施29HDFS如何写文件？Write packetCreate file Write packetWrite packetSend ackSend ackSend ackClose file30客户端联系namenode，在namenode命名空间中创建一个新文件，此时，namenode会检查文件是否存在和客户端是否有权限创建新文件，检查通过，就会创建一条记录。然后客户端写文件时，会取得合适的3个datanodes形成一个管线DataStreamer将数据包流式的传输到管线中第一个datanode，第一个data

12、node存储数据包并发送的第二个datanode, 第二个datanode存储数据包并发送的第三个datanode。当收到管道中所有datanodes的确认信息后对应数据包才会从确认队列中删除。如此反复，直到所有的数据包，都写完，最后向namenode报告写入完成。31HDFS如何读文件？Read blockOpen fileClose fileGet block location32客户端联系NameNode,得到所有数据块信息，以及数据块对应的所有数据服务器的位置信息尝试从某个数据块对应的一组数据服务器中选出一个，进行连接数据被一个包一个包发送回客户端，等到整个数据块的数据都被读取完了，就

13、会断开此链接，尝试连接下一个数据块对应的数据服务器，整个流程，依次如此反复，直到所有想读的都读取完了为止。Namenode并不实际参与数据传输。读文件流程33冗余备份策略机架策略心跳机制安全模式校验和回收站元数据保护HDFS的可靠性回顾3435什么是MapReduce？语言支持：JavaC+ (through pipes)script languages (through streaming)分布式技术框架35处理海量数据（1TB）上百/上千CPU实现并行处理简单地实现以上目的移动计算比移动数据更划算MapReduce设计目的大规模数据处理36自动实现分布式并行计算容错提供状态监控工具模型抽象

14、简洁，程序员易用计算划到大量机器上组件间不允许任意数据共享，避免保保持同步而产生的通信开销，所有在MapReduce上的数据不可变节点间通信只在产生新输出MapReduce特性37MapReduce示例The weather is goodThis guyis a good manToday is goodGood manis goodthe 1weather 1is 1good 1today 1is 1good 1this 1guy 1is 1a 1good 1man 1good 1man 1is 1good 1a 1 good 1good 1good 1good 1good 1man 1m

15、an 1the 1weather 1today 1guy 1is 1is 1is 1is 1this 1a 1good 5guy 1is 4man 2the 1this 1today 1weather 138mapping任务平等， Mapper无特定标识，处理任意输入每个mapper本地进行中间值交换，同键值送到同一个reducer。唯一的结点通信数据传送由平台处理MapReduce数据流39MapReduce数据流40从Map到Reducer数据整体传输过程41输入块InputSplit：文件拆分成块Mapred-default.xml设置块大小输入格式定义了组成mapping阶段的map

16、任务列表，每个任务对应一个输入块。据InputSplit地址分配任务（最大任务数mapred.tasktracker.map.tasks.maximum）MapReduce数据流输入42每个map任务都有一个用来写入数据的循环内存缓冲区。这个缓冲区默认大小是100M当缓冲区中的数据量达到一定阀值（默认0.8），系统将会启动一个后台进程把缓冲区的数据spill到磁盘，spill过程中，map的输出会继续写入到缓冲区，但如果缓冲区已满，map就会被阻塞直到spill完成。MapReduce数据流 Map结果输出43当map开始产生输出时，并不是简单的写入到磁盘，先是写入到缓冲区，并进行预排序，以提

17、升效率。MapReduce数据流 shuffle和sort44当map开始产生输出时，并不是简单的写入到磁盘，先是写入到缓冲区，并进行预排序，以提升效率。MapReduce数据流 shuffle和sort45Partitons:每一个reduce节点会分派到中间输出的键集合中的一个不同的子集合Partitioner类决定键值对去向，默认分类器计算键哈希值 Spill线程把缓冲区数据写到磁盘前，会对它进行一个二次快速排序，首先根据数据属性的partition排序，然后每个partition中再按照key排序输出包括一个索引文件和数据文件MapReduce数据流 Partitons46如果设定了c

18、ombiner，将在排序输出的基础上运行Combiner就是一个 min reduce，它在执行map的节点本身执行，先对map的输出做一次简单的reduce，使得map的输出更紧凑，更少的数据被写入磁盘和传输给reduceMapReduce数据流 combiner47每当内存中的数据到达spill阀值的时候，都会产生一个新的spill文件，所以在map任务写完它的最后一个输出记录时，可能会有多个spill文件。在map任务完成前，所有的spill文件将会被归并排序为一个索引文件和数据文件。当spill文件归并完毕后，map将删除所有的临时spill文件，并告知tasktracker任务完成。

19、MapReduce数据流 map输出结束48Map的输出文件放置在map任务的tasktracker的本地磁盘上，它是运行reduce任务的tasktracker所需要的输入数据。Reduce任务的输入数据分布在集群的多个map任务的输出中，map任务可能会在不同的时间内完成，只有有其中的一个map任务完成，reduce任务就会开始拷贝它的输出，这个阶段称之为拷贝阶段Reduce任务拥有多个拷贝线程，可以并行的获取map输出，线程数默认是5MapReduce数据流 copy49拷贝来的数据叠加在磁盘上，有一个后台线程会将它们合并为了更大的排序文件，节省后期的归并时间。当所有的map输出都拷贝后

20、，reduce任务进入排序节点，这个阶段会对所有的map输出进行归并排序，这个工作会重复多次假设有50个map输出，并且归并因子是10，则需要5次归并，没每次会把10个文件归并为一个，最终生成5个中间文件。之后，系统不再把5个中间文件归并成一个文件，而是排序后直接交给reduce函数，省去了向磁盘写入数据这一步。MapReduce数据流 copy50每个reduce任务都会创建一个Reducer实例reducer的reduce()方法只会调用一次，它会接收一个键和关联到键的所有值的一个迭代器，迭代器会以一个未定义的顺序返回关联到同一个键的值。 MapReduce数据流 Reduce51键值对通

21、过OutputCollector被写到输出文件中，写入的方式由输出格式控制。 每一个reducer会把结果输出写在公共文件夹中一个单独的文件内，这些文件的命名一般是part-nnnnn，nnnnn是关联到某个reduce任务的partition的idMapReduce数据流输出输出格式描述TextOutputFormat默认的输出格式， 以 key t value 的方式输出行Sequence输出二进制文件，适合于读取为子MapReduce作业的输入NullOutputFormat忽略收到的数据，即不做输出52单一Reduce处理53多Reduce处理54无Redue处理55MapReduce

22、工作原理56JobTracker一直在等待JobClient提交作业TaskTracker每隔3秒向 JobTracker发送心跳询问有没有任务可做，如果有，让其派发任务给它执行这是一道pull过程: slave主动向master拉生意MapReduce工作原理57Hadoop 中的 MapReduce 58MapReduce 作业执行流程59默认为先进先出作业队列调度支持公平调度支持容量调度作业调度机制60推测执行：既如果jobtracker发现拖后腿的任务，会再启动相同的备份任务，然后那个先执行完成就会去kill掉另外一个。因此在监控网页上经常能看到正常执行完成的作业有被kill的任务。推

23、测执行默认打开，但是如果是代码的问题，并不能解决问题，而且会使集群更慢，通过mapred-side.xml配置文件中设置mapred.map.tasks.speculative.execution和mapred.reduce.tasks.speculative.execution可为map任务或reduce任务开启或者关闭推测执行。重用jvm，可以省去启动新的jvm消耗时间， mapred-side.xml文件中设置mapred.job.reuse.jvm.num.tasks单个jvm最大执行的任务数（1，1或-1表示没有限制）忽略模式：任务在读取任务数据失败2次之后，会把数据位置报告给job

24、tracker，然后重新启动该任务并在遇到所记录的坏数据位置时直接跳过（默认关闭，同SkipBadRecord方法打开）任务执行优化61硬件故障是指jobtracker故障或tasktracker故障Jobtracker是单点，若发生故障目前hadoop还无法处理，唯有选择最牢靠的硬件作为jobtrackerJobtracker通过心跳信号（周期1分钟）了解tasktracker是否发生故障或者负载过于严重Jobtracker将从任务节点列表中移除发生故障的tasktracker。如果故障节点在执行map任务并且尚未完成，jobtracker会要求其他节点重新执行此map任务如果故障节点在执行

25、reduce任务并且尚未完成，jobtracker会要求其他节点重新执行尚未完成的reduce任务。错误处理机制硬件故障62由于代码bug或进程崩溃引起任务失败Jvm自动退出，向tasktracker父进程发生错误信息，错误信息也会写入日志Tasktracker监听程序会发现进程退出，或者进程很久没有更新信息送回，将任务标记为失败标记任务失败后，任务计数器减1以便接受新的任务，并通过心跳信号通知jobtracker任务失败的信息Jobtracker获悉任务失败后，将把该任务重新放入调度队列，重新分配出去执行如果一个任务失败超过4次（可配置），将不会再执行，同时宣告任务作业失败。错误处理机制任务

26、失败63大纲原理Hadoop 基本架构HDFS 文件系统MapReduce 分布式计算框架实践安装与配置编写MapReduce程序介绍背景Hadoop起源成功的案例64Hadoop安装配置编写MapReduce程序实践65支持平台GNU/LinuxWindows（Cygwin）必需软件JDK1.6以上SSH无密码访问必要条件66格式化HDFSHadoop namenode -format运行脚本start-all.shstart-dfs.shstart-mapred.shjps命令查询java进程启动系统67通过浏览器和http访问jobtracker所在节点的50030端口监控jobtrac

27、ker通过浏览器和http访问namenode所在节点的50070端口监控集群通过Web了解hadoop的状况68Jobtracker69NameNode70命令行方式操作HDFS文件操作71Hadoop没有当前目录的概念，列出HDFS下的文件72上传文件到HDFS73将HDFS文件复制到本地74查看HDFS基本统计信息75在新节点安装好hadoop把namenode的相关配置文件复制到该节点修改masters和slaves文件，添加该节点设置ssh无密码登陆该节点单独启动该节点上的datanode和tasktracker（hadoop-daemon.sh start datanode/tasktracker)运行start-balancer.sh进行数据负载均衡怎么添加节点76Page 1：the weather is good.Page 2 : good weather is good. 举例：单词计数 (key,value) (the,1)(weather,1)(is,1)(good,1) (key,value) (good,1)(weather,1)(is,1)(good,1) (key,value) (the,1)(weather,2)(is,2)(good,3)Map1Map2Reduce77void map(LongWritable key