hadoop--简介及原理.ppt

Hadoop介绍,1,Hadoop介绍,2,MapReduce简介,Mapreduce是一中用于数据处理的编程模型。例子:用Hadoop处理气象数据，找出每年最高气温MapReduce工作过程分为两个阶段：map阶段和reduce阶段每个阶段都有键/值对作为输入和输出。程序员还定义两个函数map函数和reduce函数。用map函数来找出年份和气温。本例map函数只是一个数据准备阶段。通过这种方式建立数据，使得reducer函数能在此基础上进行工作，找出每年最高气温。思考下面的输入数据,3,MapReduce简介(续),4,这些行以键/值对的方式来表示map函数,MapReduce简介(续),Map函数的功能仅仅提取年份和气温(粗体显示)，并将其作为输出被发送(气温值已被解释为整数)(1950,0)(1950,22)(1950,-11)(1949,211)(1949,78)Map函数的输出先有MapReduce框架处理，然后再被发送到reduce函数。这一处理过程根据键来对键/值进行排序和分组。因此，reduce函数会看到如下输入：(1949,111,78)(1950,0,22,-11)每年的年份后都有一系列气温度数。所有reduce函数现在必须重复这个列表并从中找出最大的度数：(1949,111)(1950,22)这是最后的输出：全球气温记录中每年的最高气温,5,MapReduce简介(续),上面的数据流如下图所示。在图的底部是Unix的管道，模拟整个MapReduce的流程。,6,图1MapReduce的逻辑数据流,MapReduce简介(续),MapReduce的分布式应用MapReduce作业(job)是客户端执行的基本单位：包括输入数据,MapReduce程序和配置信息。Hadoop通过把作业分成若干个小任务(task)来工作，包括。两种类型的任务:map任务和reduce任务。有两种类型的节点控制着作业执行的过程：jobtracker和多个tasktracker。Jobtracker通过调度任务在tasktracker上运行来协调所有运行在系统上的作业。Tasktracker运行任务的同时，把进度报告传送到jobtracker，jobtracker则记录着每项任务的整体进展情况。如果其中一个任务失败，jobtracker可以重新调度任务到另外一个tasktracker。Hadoop把输入数据划分成等长的小数据发送到MapReduce，成为输入分片(inputsplit)或分片。Hadoop为每个分片(split)创建一个map任务，由它来运行用户自定义的map函数来分析每个分片的记录。,7,MapReduce简介(续),Map任务把输出写入本地硬盘，而不是HDFS，因为map的输出作为中间输出，中间输出被reduce处理后产生最终的输出，一旦作业完成，map的输出就可以删除了。如果该节点运行的map任务在map输出给reduce任务处理之前崩溃，那么hadoop将在另一个节点上重新运行map任务以再次产生map输出。Reduce任务并不具备数据本地读取的优势。一个单一的reduce任务往往来自于所有mapper的输出。因此，有序map的输出必须通过网络传输到reduce任务运行的节点，在那里进行合并，然后传递到用户定义的reduce函数中。Reduce的输出通常存储在HDFS中。对于每个reduce输出的HDFS块，第一个副本存储在本地节点，其他副本存储在其他机架节点中。,8,MapReduce简介(续),一个单一的reduce任务的整个数据流如图2所示虚线框表示节点，虚线箭头表示数据传输到一个节点上实线箭头表示节点之间的数据传输,9,图2MapReduce中单一reduce任务的数据流图,MapReduce简介(续),Reduce任务数目不是由输入的大小决定的，而是单独具体指定的。如果有多个reducer,map任务会对其输出进行分区(partition)，为每个reduce任务创建一个分区(partition)。每个分区包含许多键(及其关联的值)，但每个键的记录都在同一个分区中。分区可以通过用户定义的partitioner来控制，通常是用默认的分区工具，使用hash函数来形成“木桶”键/值。多个reduce任务数据流图如图3所示。,10,图3多个reduce任务的MapReduce数据流,MapReduce简介(续),因为处理可以并行进行，也有可能不存在reduce任务，不需要shuffle的时候。这种情况下，唯一的非本地节点数据传输是当map任务写入到HDFS中。如图4。,11,图4MapReduce中没有reduce的数据流,MapReduce简介(续),CombinerFunction,12,MapReduce简介(续),CombinerFunction集群的带宽限制了MapReduce作业的数量，因此需要减小map和reduce任务之间传输数据量。Hadoop允许用户声明一个combiner，运行在map的输出上，该函数输出作为reduce函数的输入。Combiner是一个优化方法。例：1950年的读数由两个map处理(在不同的分片中)，假设第一个i饿map的输出如下:(1950,0)(1950,20)(1950,10)第二个map输出如下:(1950,25)(1950,15)Reduce函数再调用时被传入以下数字:(1950,0,20,10,25,15)因为25是输入中的最大值，所以输出如下:(1950,25)当使用了combiner，像reduce函数那样，为每个map输出找到最高气温。Reduce函数被调用的时候将被传入如下数值:(1950,20,25)Reduce输出结果和以前一样。注：combiner并不能取代reduce函数，因为reduce函数仍然需要处理来自不同的map给出的相同键记录。但它可以帮助减少map和reduce之间的数据传输量。,13,MapReduce简介(续),MapReduce的工作流如何将一个数据处理问题转换成MapReduce模型？前面的数据处理处理的问题都很简单，如何处理复杂的问题？(这里的复杂是指有更多的MapReduce作业，而不是更复杂的map和reduce函数，即增加更多的作业而不是增加作业的复杂度。)对于复杂问题，可以考虑使用MapReduce之上的高级语言，如Pig,Hive或Cascading。直接的好处是免于处理到MapReduce作业的转换，而是将中心集中于正在进行的分析上。例：假设我们想找到每个气象台年度每天平均气温最高的记录。例如：要就算029070-99999气象台的1月1日的平均每日最高气温的记录，我们将从这个气象台的1901年1月1日，1902年1月1日直到2000年的1月1日的气温中找出每日平均气温的最大值。该如何使用MapReduce来计算呢？,14,MapReduce简介(续),计算分为下面两个部分1.计算每个station-data对的每日最高气温.2.计算每个station-day-month键(key)的平均每日最高气温Mapper从上一个作业得到输出记录(station-data,最高气温值)，并丢掉年份将其值赋予到记录(station-day-month,最高气温值)。Reduce函数为每个station-day-month键(key)得到一个平均最高气温。第一部分的输出结果为：029070-99999190101010029070-9999919020101-94前两个字段是键(key),最后一列指的是从指定气象台和日期读入的最高气温。第二部分计算这些年的最高气温的平均值029070-999990101-68以上是气象台029070-99999整个世纪1月1日平均每日最高气温-6.8,15,MapReduce工作原理,Hadoop如何运行一个job？,16,MapReduce工作原理,17,图5Hapoop运行MapReduce作业的工作原理,在图中上层有四个实体:客户端，提交MapReduce作业Jobtracker，协调作业的运行。Jobtracker是Java的一个应用程序，主要类是JobTracker。Tasktracker，运行作业划分后的任务。Tasktracker是一个Java应用程序，主要类是TaskTracker。分布式文件系统(HDFS)，用来在其他实体间共享作业文件。,Hadoop运行作业的过程,JobSubmission(作业提交)JobClient的runjob()用于方法是产生JobClient实例和调用其submitJob()方法的简便方法(step1)。提交作业后，runjob()方法每秒轮训作业的进度，如果发现与上一个记录不同，便把报告显示到控制台。作业完成后，如果成功，显示作业计数器。否则，控制台会显示导致作业失败的错误。JobClient的submitJob()方法实现了作业提交。过程如下：向jobtracker请求一个新的作业ID(通过调用JobTracker的getNewJobId()(step2)。检查作业的输出说明。比如，如果没有指定输出目录或者它已经存在，作业就不会提交，所有错误返回给MapReduce程序。计算作业的输入划分。如果划分无法计算，比如因为输入路径不存在，作业就不会被提交，并有错误返回给MapReduce程序。将运行作业所需要的资源包括作业JAR文件，配置文件和计算的输入划分(computedinputsplits)复制到jobtracker文件系统中一个以作业ID号命名的目录中。由于作业JAR副本较多，在tasktracker运行作业时，集群能够为它们提供许多副本进行访问(step3)。告诉jobtracker作业准备执行(调用JobTracker的submitjob()方法)(step4)。,18,Hadoop运行作业的过程(续),JobInitialization(作业初始化)JobTracker接收到对其submitJob()方法的调用后，会把此调用方法一个内部对列中，交由作业调度器进行调度，并对其进行初始化。初始化包括创建一个代表该正在运行的作业的对象，它封装任务和记录信息，以便跟踪任务的状态和进程(step5)。要创建运行任务列表，作业调度器首先从共享文件系统中获取JobClient已计算好的输入划分信息(step6)。然后为每个划分(split)创建一个map任务。TaskAssignment(任务分配)TaskTracker执行一个简单的循环，定期发送心跳(heartbeat)方法调用Jobtracker。心跳方法告诉jobtracker,traktracker是否还存活，同时也充当两者之间的消息通道。作为心跳方法调用的一部分，tasktracker会指明他是否已经准备运行新的任务，如果是，jobtracker会为它分配一个任务，并使用心跳方法的返回值与tasktracker进行通信(step7)。,19,Hadoop运行作业的过程(续),要选择一个reduce任务，jobtracker只是简单的从尚未运行的reduce任务列表中选取下一个来执行，并没有考虑数据的本地化。然而，对于一个map任务，它考虑的是tasktracker的网络位置和选取一个距离其输入划分文件最近的tasktracker。在理想的情况下，任务是data-local(数据本地化)的，与分割文件所在的节点运行在相同的节点上。同样，任务也可能是rack-local(机架本地化)的：和分割文件在同一个机架，但不在同一节点。一些任务既不是数据本地化的，也不是机架本地化的，从与它们自身运行的不同机架上检索数据。可以通过查看作业的计数器得知每种类型任务的比例。,20,Hadoop运行作业的过程(续),TaskExecution(任务执行)Tasktracker本地化作业的JAR文件，将它从共享文件系统复制到tasktracker所在的文件系统，同时，将应用程序所需要的全部文件从分布式缓存复制到本地磁盘(step8)。然后，为任务新建一个本地工作目录，并把JAR文件中的内容解压到这个文件夹中。第3步，新建一个taskrunner实例来运行任务。TaskRunner启动一个新的Java虚拟机(step9)来运行每个任务(step10)，使得用户定义的map和reduce函数的任何缺陷都不会影响tasktracker。子进程通过unbilical接口与父进程进行通信。它每隔几秒便告诉父进程它的进度，直到任务完成。,21,Hadoop运行作业的过程(续),ProgressandStatusUpdates(进度和状态更新)MapReduce作业是一个长时间运行的批量作业，能够得知作业的进展是很重要的。一个作业和每个任务都有一个状态，这些状态信息怎样与客户端进行通信呢？任务在运行时，对任务进度(即任务完成率)保持追踪。对于map任务，是已处理完输入的百分比。对于reduce任务，是估计reduce输入已处理的百分比。JobCompletion(作业完成)Jobtracker收到作业最后一个任务已完成的通知后，便把作业的状态设置为“成功”。然后，在JobClient查询状态时,它将得知任务已成功完成，所以便显示一条消息告诉用户，然后从runJob()方法返回。如果jobtracker有相应的设置，也会发送一个HTTP作业通知，希望收到回调的客户端可以通过job.end.notification.url属性来设置。最后，jobtracker清空作业的工作状态，提示tasktracker也清空作业的工作状态。,22,作业的调度,FIFO调度算法(早期版本)FairScheduler,23,Shuffle和排序,MapReduce保证每个reduce输入都已经按键排序。系统执行排序的过程map输出传到reduce作为后者的输入即成为shuffle(混洗或称洗牌)。学习shuffle如何工作，对于需要优化MapReduce程序大有帮助。,24,图6MapReduce的shuffle和排序,MapReduce的类型与格式,Hadoop如何处理格式化数据，非格式化数据，半格式化数据？Hadoop可以处理很多格式的数据，从一般的文本文件到一整个数据库。InputFormats(输入格式)InputSplitsandRecord(输入分片与记录)TextInput(文本输入)BinaryInput(二进制输入)MultipleInput(多种输入)DatabaseInput(数据库输入),25,MapReduce的类型与格式,26,图7InputFormat类的层次结构,MapReduce的类型与格式,OutputFormats(输出格式)TextOutput(文本输出)BinaryOutput(二进制输出)MultipleOutput(多种输出)LazyOutput(延迟输出)DatabaseOutput(数据库输出),27,MapReduce的类型与格式,28,图8OutputFormat类的层次结构,Pig简介,什么是Pig?Pig是SQL-like语言,是在MapReduce上构建的一种高级查询语言,把一些运算编译进MapReduce模型的Map和Reduce中,并且用户可以定义自己的功能。Pig有以下两部分构成：表达数据流的语言，称为PigLatin。运行PigLatin程序的执行环境。目前有两种环境：单个JVM本地执行和在Hadoop集群上分布运行。PigLatin程序有一系列操作或者转换组成，用于将输入数据生成输出。从总体上看，这些操作描述了一个数据流，Pig执行环境将这个数据流转换成可执行的语句然后运行。实际上，Pig将转换变成一系列的MapReduce作业，它能是人们专注于数据而不是执行本质。,29,Pig简介,Pig的特点Pig是一种用于搜索大型数据集的脚本语言。MapReduce缺点之一就是开发周期很长。编写mapper和reducer，编译和打包代码，提交作业和结果检索是一个耗时的工作。Pig的优点就是它能够通过从控制台发出6条PigLatin的方法简单处理数TB的数据。由于Pig提供了几条用于反复检查程序中数据结构的命令，所以它非常支持程序员编写查询语句。Pig被设计成是可扩展的。处理路径中所有部分几乎都是可定制的：装载，存储，过滤，分组，排序和链接都可以换用用户定义函数(UDF)。这些功能在Pig内嵌数据模型上工作，所以他们可以与Pig的平台紧密结合在一起。,30,Pig简介,PigPig运行的是一个客户端应用程序。在Hapoop集群上运行Pig也不需要在集群