赞
踩
这一篇里,开始对Mapper.class的子类进行解读。
先回忆一下。Mapper有setup(),map(),cleanup()和run()四个方法。其中setup()一般是用来进行一些map()前的准备工作,map()则一般承担主要的处理工作,cleanup()则是收尾工作如关闭文件或者执行map()后的K-V分发等。run()方法提供了setup->map->cleanup()的执行模板。
在MapReduce中,Mapper从一个输入分片中读取数据,然后经过Shuffle and Sort阶段,分发数据给Reducer,在Map端和Reduce端我们可能使用设置的Combiner进行合并,这在Reduce前进行。Partitioner控制每个K-V对应该被分发到哪个reducer[我们的Job可能有多个reducer],Hadoop默认使用HashPartitioner,HashPartitioner使用key的hashCode对reducer的数量取模得来。
我们先来看看三个Mapper的子类,它们位于src\mapred\org\apache\hadoop\mapreduce\lib\map中。
1、TokenCounterMapper
在MapReduce的“Hello world”:WordCount例子中,我们完全可以直接使用这个TokenCounterMapper作为MapperClass,仅需用job.setMapperClass(TokenCounterMapper.class)进行设置即可。
2、InverseMapper
3、MultithreadedMapper
这个类稍微有点复杂,它是使用多线程来执行一个Mapper。我们可以从类图中看到,它有一个mapClass属性,这个属性指定另一个Mapper类[暂称workMapper,由mapred.map.multithreadedrunner.class设置],实际干活的其实是这个Mapper类而不是MultithreadedMapper。runnsers是运行的线程的列表。
下面是MultithreadedMapper的run()方法,它重写了Mapper中的run()。
MapRunner继承了Thread,它包含了一个独享的Context:subcontext,以及用mapper指定了workMapper,然后throwable是在MultithreadMapper的run()中进行综合的异常处理的。
MultithreadedMapper适用于CPU密集型的任务,采用多个线程处理后,一个线程可以在另外的线程在执行时读取数据并执行,这样就使用了更多的CPU周期来执行任务,从而提高吞吐率。注意读写操作都是线程安全的,因此不难想象对于IO密集型的作业,采用MultithreadedMapper会适得其反,因为会有多个线程等待IO,IO成为限制吞吐率的关键。对于IO密集型的任务,我们应该采用增多task数量的方法来解决,因为这样在IO上就是并行的。
除非map()的确是CPU密集型的,否则不推荐使用MultithreadedMapper,而建议采用更多的map task。
Copyright © 2003-2013 www.wpsshop.cn 版权所有,并保留所有权利。