分享

关于Mapper、Reducer的个人总结

nettman 发表于 2014-3-6 00:30:51 [显示全部楼层] 回帖奖励 阅读模式 关闭右栏 20 156357
本帖最后由 nettman 于 2014-3-6 00:37 编辑
我们了解mapreduce的作用是什么,我们还可以详细了解下面内容:
RecordReader的作用是什么?
job类的作用是什么?
Reducer包含几个步骤?
扩充:
Reducer所有步骤是不是必须的

加微信w3aboutyun,可拉入技术爱好者群

已有(20)人评论

跳转到指定楼层
nettman 发表于 2014-3-6 00:36:39
Mapper的处理过程:

1.1. InputFormat 产生 InputSplit,并且调用RecordReader将这些逻辑单元(InputSplit)转化为map task的输入。其中InputSplit是map task处理的最小输入单元的逻辑表示。

1.2. 在客户端代码中调用Job类来设置参数,并执行在hadoop集群的上的MapReduce程序。

1.3. Mapper类在Job中被实例化,并且通过MapContext对象来传递参数设置。可以调用Job.getConfiguration().set(“myKey”, “myVal”)来设置参数。

1.4. Mapper的run()方法调用了自身的setup()来设置参数。

1.5. Mapper的run()方法调用map(KeyInType, ValInType, Context)方法来依次处理InputSplit中输入的key/value数据对。并且可以通过Context参数的Context.write(KeyOutType, ValOutType)方法来发射处理的结果。用户程序还可以用Context来设置状态信息等。同时,用户还可以用Job.setCombinerClass(Class)来设置Combiner,将map产生的中间态数据在Mapper本地进行汇聚,从而减少传递给Reducer的数据。

1.6. Mapper的run()方法调用cleanup()方法。



一些说明:

所有的中间结果都会被MapReduce框架自动的分组,然后传递给Reducer(s)去产生最后的结果。用户可以通过Job.setGroupingComparatorClass(Class)来设置Comparator。如果这个Comparator没有被设置,那么所有有一样的key的数据不会被排序。

Mapper的结果都是被排序过的,并被划分为R个区块(R是Reducer的个数)。用户可以通过实现自定义的Partitioner类来指定哪些数据被划分给哪个Reducer。

中间态的数据往往用(key-len, key, value-len, value)的简单格式存储。用户程序可以指定CompressionCodec来压缩中间数据。

Map的数目由输入数据的总大小决定。一般来说,一个计算节点10-100个map任务有较好的并行性。如果cpu的计算量很小,那么平均每个计算节点300个map任务也是可以的。但是每个任务都是需要时间来初始化的,因此每个任务不能划分的太小,至少也要平均一个任务执行个一分钟。可以通过mapreduce.job.maps参数来设置map的数目。更进一步说,任务的数量是有InputFormat.getSplits()方法来控制的,用户可以重写这个方法。

------------------------------------------------------------------------------------------------------------------------------------------------

Reducer主要分三个步骤:

1.       Shuffle洗牌

这步骤是Reducer从相关的Mapper节点获取中间态的数据。

2.       Sort排序

在洗牌的同时,Reducer对获取的数据进行排序。在这个过程中,可以用Job.setGroupingComparatorClass(Class)来对同一个key的数据进行Secondary Sort二次排序。

3.       Reduce

Reduce的调用顺序和Map差不多,也是通过run()方法调用setup(),reduce(),cleanup()来实现的。Reducer输出的数据是没有经过排序的。

Reduce 的数目可以通过Job.setNumReduceTasks(int)来设置。一般来说,Reduce的数目是节点数的0.95到1.75倍。

Reduce的数目也可以设置为0,那么这样map的输出会直接写到文件系统中。

Reduce中还可以使用Mark-Reset的功能。简而言之就是可以在遍历map产生的中间态的数据的时候可以进行标记,然后在后面适当的时候用reset回到最近标记的位置。当然这是有一点限制的,如下面的例子,必须自己在Reduce方法中对reduce的Iterator重新new 一个MarkableIterator才能使用。

public void reduce(IntWritable key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
MarkableIterator<IntWritable> mitr = new MarkableIterator<IntWritable>(values.iterator());
// Mark the position
mitr.mark();
while (mitr.hasNext()) {
i = mitr.next();
// Do the necessary processing
}


// Reset
mitr.reset();
// Iterate all over again. Since mark was called before the first
// call to mitr.next() in this example, we will iterate over all
// the values now
while (mitr.hasNext()) {
i = mitr.next();
// Do the necessary processing
}
}



3.     Partitioner

Partitioner 对map输出的中间态数据按照reduce的数目进行分区,一般通过hash来进行划分。默认的实现是HashPartitioner。

4.     Reporting Progress

MapReduce程序可以通过mapper或者reducer的Context来汇报应用程序的状态。

5.     Job

当我们在写MapReduce程序的时候,通常,在main函数里。建立一个Job对象,设置它的JobName,然后配置输入输出路径,设置我们的Mapper类和Reducer类,设置InputFormat和正确的输出类型等等。然后我们会使用job.waitForCompletion()提交到JobTracker,等待job运行并返回,这就是一般的Job设置过程。JobTracker会初始化这个Job,获取输入分片,然后将一个一个的task任务分配给TaskTrackers执行。TaskTracker获取task是通过心跳的返回值得到的,然后TaskTracker就会为收到的task启动一个JVM来运行。

Job其实就是提供配置作业、获取作业配置、以及提交作业的功能,以及跟踪作业进度和控制作业。Job类继承于JobContext类。JobContext提供了获取作业配置的功能,如作业ID,作业的Mapper类,Reducer类,输入格式,输出格式等等,它们除了作业ID之外,都是只读的。 Job类在JobContext的基础上,提供了设置作业配置信息的功能、跟踪进度,以及提交作业的接口和控制作业的方法。


一个Job对象有两种状态,DEFINE和RUNNING,Job对象被创建时的状态时DEFINE,当且仅当Job对象处于DEFINE状态,才可以用来设置作业的一些配置,如Reduce task的数量、InputFormat类、工作的Mapper类,Partitioner类等等,这些设置是通过设置配置信息conf来实现的;当作业通过submit()被提交,就会将这个Job对象的状态设置为RUNNING,这时候作业以及提交了,就不能再设置上面那些参数了,作业处于调度运行阶段。处于RUNNING状态的作业我们可以获取作业、maptask和reduce task的进度,通过代码中的*Progress()获得,这些函数是通过info来获取的,info是RunningJob对象,它是实际在运行的作业的一组获取作业情况的接口,如Progress。


  在waitForCompletion()中,首先用submit()提交作业,然后等待info.waitForCompletion()返回作业执行完毕。verbose参数用来决定是否将运行进度等信息输出给用户。submit()首先会检查是否正确使用了new API,这通过setUseNewAPI()检查旧版本的属性是否被设置来实现的,接着就connect()连接JobTracker并提交。实际提交作业的是一个JobClient对象,提交作业后返回一个RunningJob对象,这个对象可以跟踪作业的进度以及含有由JobTracker设置的作业ID。


  getCounter()函数是用来返回这个作业的计数器列表的,计数器被用来收集作业的统计信息,比如失败的map task数量,reduce输出的记录数等等。它包括内置计数器和用户定义的计数器,用户自定义的计数器可以用来收集用户需要的特定信息。计数器首先被每个task定期传输到TaskTracker,最后TaskTracker再传到JobTracker收集起来。这就意味着,计数器是全局的。



回复

使用道具 举报

tang 发表于 2015-4-3 21:45:24
回复

使用道具 举报

lmf19900120 发表于 2015-4-14 11:41:49
总结的挺好的,
回复

使用道具 举报

trodking 发表于 2015-4-29 15:36:56
回复

使用道具 举报

llp 发表于 2015-5-27 14:23:17
32个赞,真心是不错的
回复

使用道具 举报

小杨杨 发表于 2015-5-31 21:02:12
官网上对reduce阶段 还有个combiner
回复

使用道具 举报

小杨杨 发表于 2015-5-31 21:04:30
总结的相当赞啊 楼主  忍不住看了2遍
回复

使用道具 举报

hery 发表于 2015-7-3 16:26:07
回复

使用道具 举报

Rekklesly 发表于 2015-7-10 01:39:30
那么请问一下。。如果想重新计算分区函数的话。。REDUCE是不是就不能进行。。。因为分区函数还并没有确定
回复

使用道具 举报

123下一页
您需要登录后才可以回帖 登录 | 立即注册

本版积分规则

关闭

推荐上一条 /2 下一条