Hadoop的核心组件在一起工作时如下图所示:
图4.4高层MapReduce工作流水线 MapReduce的输入一般来自HDFS中的文件,这些文件分布存储在集群内的节点上。运行一个MapReduce程序会在集群的许多节点甚至所有节点上运行mapping任务,每一个mapping任务都是平等的:mappers没有特定“标识物”与其关联。因此,任意的mapper都可以处理任意的输入文件。每一个mapper会加载一些存储在运行节点本地的文件集来进行处理(译注:这是移动计算,把计算移动到数据所在节点,可以避免额外的数据传输开销)。 当mapping阶段完成后,这阶段所生成的中间键值对数据必须在节点间进行交换,把具有相同键的数值发送到同一个reducer那里。Reduce任务在集群内的分布节点同mappers的一样。这是MapReduce中唯一的任务节点间的通信过程。map任务间不会进行任何的信息交换,也不会去关心别的map任务的存在。相似的,不同的reduce任务之间也不会有通信。用户不能显式的从一台机器封送信息到另外一台机器;所有数据传送都是由Hadoop MapReduce平台自身去做的,这些是通过关联到数值上的不同键来隐式引导的。这是Hadoop MapReduce的可靠性的基础元素。如果集群中的节点失效了,任务必须可以被重新启动。如果任务已经执行了有副作用(side-effect)的操作,比如说,跟外面进行通信,那共享状态必须存在可以重启的任务上。消除了通信和副作用问题,那重启就可以做得更优雅些。
近距离观察 在上一图中,描述了Hadoop MapReduce的高层视图。从那个图你可以看到mapper和reducer组件是如何用到词频统计程序中的,它们是如何完成它们的目标的。接下来,我们要近距离的来来看看这个系统以获取更多的细节。
图4.5细节化的Hadoop MapReduce数据流 图4.5展示了流线水中的更多机制。虽然只有2个节点,但相同的流水线可以复制到跨越大量节点的系统上。下去的几个段落会详细讲述MapReduce程序的各个阶段。
1.输入文件:文件是MapReduce任务的数据的初始存储地。正常情况下,输入文件一般是存在HDFS里。这些文件的格式可以是任意的;我们可以使用基于行的日志文件,也可以使用二进制格式,多行输入记录或其它一些格式。这些文件会很大—数十G或更大。 2. 输入格式:InputFormat类定义了如何分割和读取输入文件,它提供有下面的几个功能:
Hadoop自带了好几个输入格式。其中有一个抽象类叫FileInputFormat,所有操作文件的InputFormat类都是从它那里继承功能和属性。当开启Hadoop作业时,FileInputFormat会得到一个路径参数,这个路径内包含了所需要处理的文件,FileInputFormat会读取这个文件夹内的所有文件(译注:默认不包括子文件夹内的),然后它会把这些文件拆分成一个或多个的InputSplit。你可以通过JobConf对象的setInputFormat()方法来设定应用到你的作业输入文件上的输入格式。下表给出了一些标准的输入格式:
表4.1MapReduce提供的输入格式 默认的输入格式是TextInputFormat,它把输入文件每一行作为单独的一个记录,但不做解析处理。这对那些没有被格式化的数据或是基于行的记录来说是很有用的,比如日志文件。更有趣的一个输入格式是KeyValueInputFormat,这个格式也是把输入文件每一行作为单独的一个记录。然而不同的是TextInputFormat把整个文件行当做值数据,KeyValueInputFormat则是通过搜寻tab字符来把行拆分为键值对。这在把一个MapReduce的作业输出作为下一个作业的输入时显得特别有用,因为默认输出格式(下面有更详细的描述)正是按KeyValueInputFormat格式输出数据。最后来讲讲SequenceFileInputFormat,它会读取特殊的特定于Hadoop的二进制文件,这些文件包含了很多能让Hadoop的mapper快速读取数据的特性。Sequence文件是块压缩的并提供了对几种数据类型(不仅仅是文本类型)直接的序列化与反序列化操作。Squence文件可以作为MapReduce任务的输出数据,并且用它做一个MapReduce作业到另一个作业的中间数据是很高效的。
3. 数据片段(InputSplit):
一个输入块描述了构成MapReduce程序中单个map任务的一个单元。把一个MapReduce程序应用到一个数据集上,即是指一个作业,会由几个(也可能几百个)任务组成。Map任务可能会读取整个文件,但一般是读取文件的一部分。默认情况下,FileInputFormat及其子类会以64MB(与HDFS的Block默认大小相同,译注:Hadoop建议Split大小与此相同)为基数来拆分文件。你可以在hadoop-site.xml(译注:0.20.*以后是在mapred-default.xml里)文件内设定mapred.min.split.size参数来控制具体划分大小,或者在具体MapReduce作业的JobConf对象中重写这个参数。通过以块形式处理文件,我们可以让多个map任务并行的操作一个文件。如果文件非常大的话,这个特性可以通过并行处理大幅的提升性能。更重要的是,因为多个块(Block)组成的文件可能会分散在集群内的好几个节点上(译注:事实上就是这样),这样就可以把任务调度在不同的节点上;因此所有的单个块都是本地处理的,而不是把数据从一个节点传输到另外一个节点。当然,日志文件可以以明智的块处理方式进行处理,但是有些文件格式不支持块处理方式。针对这种情况,你可以写一个自定义的InputFormat,这样你就可以控制你文件是如何被拆分(或不拆分)成文件块的。自定义的文件格式在第五部分有描述。 4. 记录读取器(RecordReader)
InputSplit定义了如何切分工作,但是没有描述如何去访问它。 RecordReader类则是实际的用来加载数据并把数据转换为适合mapper读取的键值对。RecordReader实例是由输入格式定义的,默认的输入格式,TextInputFormat,提供了一个LineRecordReader,这个类的会把输入文件的每一行作为一个新的值,关联到每一行的键则是该行在文件中的字节偏移量。RecordReader会在输入块上被重复的调用直到整个输入块被处理完毕,每一次调用RecordReader都会调用Mapper的map()方法。 5. Mapper:
Mapper执行了MapReduce程序第一阶段中有趣的用户定义的工作。给定一个键值对,map()方法会生成一个或多个键值对,这些键值对会被送到Reducer那里。对于整个作业输入部分的每一个map任务(输入块),每一个新的Mapper实例都会在单独的Java进程中被初始化,mapper之间不能进行通信。这就使得每一个map任务的可靠性不受其它map任务的影响,只由本地机器的可靠性来决定。map()方法除了键值对外还会接收额外的两个参数(译注:在0.20.×后的版本,接口已变化,由Context对象代替这两个参数):
6. Partition & Shuffle:
当第一个map任务完成后,节点可能还要继续执行更多的map任务,但这时候也开始把map任务的中间输出交换到需要它们的reducer那里去,这个移动map输出到reducer的过程叫做shuffle。每一个reduce节点会分派到中间输出的键集合中的一个不同的子集合,这些子集合(被称为“partitions”)是reduce任务的输入数据。每一个map任务生成的键值对可能会隶属于任意的partition,有着相同键的数值总是在一起被reduce,不管它是来自那个mapper的。因此,所有的map节点必须就把不同的中间数据发往何处达成一致。Partitioner类就是用来决定给定键值对的去向,默认的分类器(partitioner)会计算键的哈希值并基于这个结果来把键赋到相应的partition上,自定义的分类器在第五部分有详细描述。 7. 排序:
每一个reduce任务负责归约(reduceing)关联到相同键上的所有数值,每一个节点收到的中间键集合在被送到具体的reducer那里前就已经自动被Hadoop排序过了。 8. 归约(Reduce):
每个reduce任务都会创建一个Reducer实例,这是一个用户自定义代码的实例,负责执行特定作业的第二个重要的阶段。对于每一个已赋予到reducer的partition内的键来说,reducer的reduce()方法只会调用一次,它会接收一个键和关联到键的所有值的一个迭代器,迭代器会以一个未定义的顺序返回关联到同一个键的值。reducer也要接收一个OutputCollector和Report对象,它们像在map()方法中那样被使用。 9.输出格式:
提供给OutputCollector的键值对会被写到输出文件中,写入的方式由输出格式控制。OutputFormat的功能跟前面描述的InputFormat类很像,Hadoop提供的OutputFormat的实例会把文件写在本地磁盘或HDFS上,它们都是继承自公共的FileInputFormat类。每一个reducer会把结果输出写在公共文件夹中一个单独的文件内,这些文件的命名一般是part-nnnnn,nnnnn是关联到某个reduce任务的partition的id,输出文件夹通过FileOutputFormat.setOutputPath() 来设置。你可以通过具体MapReduce作业的JobConf对象的setOutputFormat()方法来设置具体用到的输出格式。下表给出了已提供的输出格式:
表4.2: Hadoop提供的输出格式 |