MapReduce数据流(二)

 输入块(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,这样你就可以控制你文件是如何被拆分(或不拆分)成文件块的。自定义的文件格式在第五部分有描述。
  输入格式定义了组成mapping阶段的map任务列表,每一个任务对应一个输入块。接着根据输入文件块所在的物理地址,这些任务会被分派到对应的系统节点上,可能会有多个map任务被分派到同一个节点上。任务分派好后,节点开始运行任务,尝试去最大并行化执行。节点上的最大任务并行数由mapred.tasktracker.map.tasks.maximum参数控制。
  记录读取器(RecordReader):InputSplit定义了如何切分工作,但是没有描述如何去访问它。 RecordReader类则是实际的用来加载数据并把数据转换为适合mapper读取的键值对。RecordReader实例是由输入格式定义的,默认的输入格式,TextInputFormat,提供了一个LineRecordReader,这个类的会把输入文件的每一行作为一个新的值,关联到每一行的键则是该行在文件中的字节偏移量。RecordReader会在输入块上被重复的调用直到整个输入块被处理完毕,每一次调用RecordReader都会调用Mapper的map()方法。
  Mapper:Mapper执行了MapReduce程序第一阶段中有趣的用户定义的工作。给定一个键值对,map()方法会生成一个或多个键值对,这些键值对会被送到Reducer那里。对于整个作业输入部分的每一个map任务(输入块),每一个新的Mapper实例都会在单独的Java进程中被初始化,mapper之间不能进行通信。这就使得每一个map任务的可靠性不受其它map任务的影响,只由本地机器的可靠性来决定。map()方法除了键值对外还会接收额外的两个参数(译注:在0.20.×后的版本,接口已变化,由Context对象代替这两个参数):

  • OutputCollector对象有一个叫collect()的方法,它可以利用该方法把键值对送到作业的reduce阶段。
  • Reporter对象提供当前任务的信息,它的getInputSplit()方法会返回一个描述当前输入块的对象,并且还允许map任务提供关于系统执行进度的额外信息。setStatus()方法允许你生成一个反馈给用户的状态消息,incrCounter()方法允许你递增共享的高性能计数器,除了默认的计数器外,你还可以定义更多的你想要的计数器。每一个mapper都可以递增计数器,JobTracker会收集由不同处理得到的递增数据并把它们聚集在一起以供作业结束后的读取。

  Partition & Shuffle:当第一个map任务完成后,节点可能还要继续执行更多的map任务,但这时候也开始把map任务的中间输出交换到需要它们的reducer那里去,这个移动map输出到reducer的过程叫做shuffle。每一个reduce节点会分派到中间输出的键集合中的一个不同的子集合,这些子集合(被称为“partitions”)是reduce任务的输入数据。每一个map任务生成的键值对可能会隶属于任意的partition,有着相同键的数值总是在一起被reduce,不管它是来自那个mapper的。因此,所有的map节点必须就把不同的中间数据发往何处达成一致。Partitioner类就是用来决定给定键值对的去向,默认的分类器(partitioner)会计算键的哈希值并基于这个结果来把键赋到相应的partition上,自定义的分类器在第五部分有详细描述。
  排序:每一个reduce任务负责归约(reduceing)关联到相同键上的所有数值,每一个节点收到的中间键集合在被送到具体的reducer那里前就已经自动被Hadoop排序过了。
  归约(Reduce):每个reduce任务都会创建一个Reducer实例,这是一个用户自定义代码的实例,负责执行特定作业的第二个重要的阶段。对于每一个已赋予到reducer的partition内的键来说,reducer的reduce()方法只会调用一次,它会接收一个键和关联到键的所有值的一个迭代器,迭代器会以一个未定义的顺序返回关联到同一个键的值。reducer也要接收一个OutputCollector和Report对象,它们像在map()方法中那样被使用。
  输出格式:提供给OutputCollector的键值对会被写到输出文件中,写入的方式由输出格式控制。OutputFormat的功能跟前面描述的InputFormat类很像,Hadoop提供的OutputFormat的实例会把文件写在本地磁盘或HDFS上,它们都是继承自公共的FileInputFormat类。每一个reducer会把结果输出写在公共文件夹中一个单独的文件内,这些文件的命名一般是part-nnnnn,nnnnn是关联到某个reduce任务的partition的id,输出文件夹通过FileOutputFormat.setOutputPath() 来设置。你可以通过具体MapReduce作业的JobConf对象的setOutputFormat()方法来设置具体用到的输出格式。下表给出了已提供的输出格式:


输出格式


描述


TextOutputFormat


默认的输出格式, 以 "key \t value" 的方式输出行


SequenceFileOutputFormat


输出二进制文件,适合于读取为子MapReduce作业的输入


NullOutputFormat


忽略收到的数据,即不做输出

表4.2: Hadoop提供的输出格式
  Hadoop提供了一些OutputFormat实例用于写入文件,基本的(默认的)实例是TextOutputFormat,它会以一行一个键值对的方式把数据写入一个文本文件里。这样后面的MapReduce任务就可以通过KeyValueInputFormat类简单的重新读取所需的输入数据了,而且也适合于人的阅读。还有一个更适合于在MapReduce作业间使用的中间格式,那就是SequenceFileOutputFormat,它可以快速的序列化任意的数据类型到文件中,而对应SequenceFileInputFormat则会把文件反序列化为相同的类型并提交为下一个Mapper的输入数据,方式和前一个Reducer的生成方式一样。NullOutputFormat不会生成输出文件并丢弃任何通过OutputCollector传递给它的键值对,如果你在要reduce()方法中显式的写你自己的输出文件并且不想Hadoop框架输出额外的空输出文件,那这个类是很有用的。
  RecordWriter:这个跟InputFormat中通过RecordReader读取单个记录的实现很相似,OutputFormat类是RecordWriter对象的工厂方法,用来把单个的记录写到文件中,就像是OuputFormat直接写入的一样。
  Reducer输出的文件会留在HDFS上供你的其它应用使用,比如另外一个MapReduce作业,或一个给人工检查的单独程序。

时间: 2024-11-06 10:01:12

MapReduce数据流(二)的相关文章

MapReduce数据流

MapReduce的输入一般来自HDFS中的文件,这些文件分布存储在集群内的节点上.运行一个MapReduce程序会在集群的许多节点甚至所有节点上运行mapping任务,每一个mapping任务都是平等的:mappers没有特定“标识物”与其关联.因此,任意的mapper都可以处理任意的输入文件.每一个mapper会加载一些存储在运行节点本地的文件集来进行处理(译注:这是移动计算,把计算移动到数据所在节点,可以避免额外的数据传输开销). 当mapping阶段完成后,这阶段所生成的中间键值对数据必

MapReduce数据流(一)

在上一篇文章中我们讲解了一个基本的MapReduce作业由那些基本组件组成,从高层来看,所有的组件在一起工作时如下图所示: 图4.4高层MapReduce工作流水线 MapReduce的输入一般来自HDFS中的文件,这些文件分布存储在集群内的节点上.运行一个MapReduce程序会在集群的许多节点甚至所有节点上运行mapping任务,每一个mapping任务都是平等的:mappers没有特定“标识物”与其关联.因此,任意的mapper都可以处理任意的输入文件.每一个mapper会加载一些存储在运

MapReduce教程(二)MapReduce框架Partitioner分区<转>

1 Partitioner分区 1.1 Partitioner分区描述 在进行MapReduce计算时,有时候需要把最终的输出数据分到不同的文件中,按照手机号码段划分的话,需要把同一手机号码段的数据放到一个文件中:按照省份划分的话,需要把同一省份的数据放到一个文件中:按照性别划分的话,需要把同一性别的数据放到一个文件中.我们知道最终的输出数据是来自于Reducer任务.那么,如果要得到多个文件,意味着有同样数量的Reducer任务在运行.Reducer任务的数据来自于Mapper任务,也就说Ma

理解MapReduce数据流

一.先理解MapReduce作业组成 一个完整的MapReduce作业称作job,它包括三部分: 输入数据 MapReduce程序 配置信息 Hadoop工作时会将job分成若干个task:map任务和reduce任务 有两类节点控制作业执行的过程:JobTracker和TaskTracker JobTracker:记录作业整体进度,对TaskTracker进行调度 TaskTracker:执行task任务并向JobTracker汇报 二.大块数据先流入map Hadoop会将输入数据划分成等长

mapreduce的二次排序实现方式

本文主要介绍下二次排序的实现方式 我们知道mapreduce是按照key来进行排序的,那么如果有有个需求就是先按照第一个字段排序,在第一个字段相等的情况下,按照第二个字段排序,这个其实就是二次排序. 下面就具体说一下二次排序的实现方式 1. 自定义一个key 为什么要自定义一个key,我们知道mapreduce中排序就是按照key来排序的,我们既然想要实现按照两个字段进行排序,默认的方式肯定不行,所以需要定义一个新的key,key里面有两个属性,也就是我们要排序的两个字段 首先,实现Writab

MapReduce实现二度好友关系

一.问题定义 我在网上找了些,关于二度人脉算法的实现,大部分无非是通过广度搜索算法来查找,犹豫深度已经明确了2以内:这个算法其实很简单,第一步找到你关注的人:第二步找到这些人关注的人,最后找出第二步结果中出现频率最高的一个或多个人(频率这块没完成),即完成. 但如果有千万级别的用户,那在运算时,就肯定会把这些用户的follow 关系放到内存中,计算的时候依次查找:先说明下我没有明确的诊断对比,这样做的效果一定没 基于hadoop实现的好:只是自己,想用hadoop实现下,最近也在学:若有不足的地

MapReduce编程(二) 排序

一.问题描述 文件中存储了商品id和商品价格的信息,文件中每行2列,第一列文本类型代表商品id,第二列为double类型代表商品价格.数据格式如下: pid0 334589.41 pid1 663306.49 pid2 499226.8 pid3 130618.22 pid4 513708.8 pid5 723470.7 pid6 998579.14 pid7 831682.84 pid8 87723.96 要求使用MapReduce,按商品的价格从低到高排序,输出格式仍为原来的格式:第一列为商

MapReduce处理二次排序(分区-排序-分组)

MapReduce二次排序原理 在map阶段,使用job.setInputFormatClass定义的InputFormat将输入的数据集分割成小数据块splites,同时InputFormat提供一个RecordReader的实现. 本例子中使用的时TextInputFormat,他提供的RecordReader会将文本的字节偏移量作为key,这一行的文本作为value. 这就是自定义Map的输入是<LongWritable,Text>的原因,然后调用自定义的Map的map方法,将一个个&l

mapreduce 的二次排序

一: 理解二次排序的功能, 使用自己理解的方式表达(包括自定义数据类型,分区,分组,排序) 二: 编写实现二次排序功能, 提供源码文件. 三:理解mapreduce join 的几种 方式,编码实现reduce join,提供源代码,说出思路. 一: 二次排序 使用自己理解的方式表达(包括自定义数据类型,分区,分组,排序) 1.1 二次排序的功能 1. 当客户端提交一个作业的时候,hadoop 会开启yarn 接受进行数据拷贝处理,之后交友有yarn 框架上的启动服务resourcemanage