hadoop中mapreduce的常用类(1)(3)

Map数

通常map数由输入数据总大小决定，也就是所有输入文件的blocks数目决定。

每个节点并行的运行的map数正常在10到100个。由于Map任务初始化本身需要一段时间所以map运行时间至少在1分钟为好。

如此，如果有10T的数据文件，每个block大小128M，最大使用为82000map数，除非使用setNumMapTasks(int)(这个方法仅仅对MR框架提供一个建议值)将map数值设置到更高。

Reducer

Reducer根据key将中间数据集合处理合并为更小的数据结果集。

用户可以通过JobConf.setNumReduceTasks(int)设置作业的reducer数目。

整体而言，Reducer实现类通过JobConfigurable.configure(JobConf)方法将JobConf对象传入,并为Job设置和初始化Reducer。MR框架调用 reduce(WritableComparable, Iterator, OutputCollector, Reporter) 来处理以key被分组的输入数据。应用可以覆盖Closeable.close()处理必要的清理操作。

Reducer由三个主要阶段组成:shuffle，sort，reduce。

❈shuffle

输入到Reducer的输入数据是Mapper已经排过序的数据.在shuffle阶段，根据partition算法获取相关的mapper地址，并通过Http协议将mapper的相应输出数据由reducer拉取到reducer机器上处理。

❈sort

框架在这个阶段会根据key对reducer的输入进行分组(因为不同的mapper输出的数据中可能含有相同的key)。

shuffle和sort是同时进行的，同时reducer仍然在拉取map的输出。

❈Secondary Sort

如果对中间数据key进行分组的规则和在处理化简阶段前对key分组规则不一致时，可以通过 JobConf.setOutputValueGroupingComparator(Class)设置一个Comparator。因为中间数据的分组策略是通过 JobConf.setOutputKeyComparatorClass(Class) 设置的，可以控制中间数据根据哪些key进行分组。而JobConf.setOutputValueGroupingComparator(Class)则可用于在数据连接情况下对value进行二次排序。

Reduce(化简)

这个阶段框架循环调用 reduce(WritableComparable, Iterator, OutputCollector, Reporter) 方法处理被分组的每个kv对。

reduce 任务一般通过 OutputCollector.collect(WritableComparable, Writable)将输出数据写入文件系统FileSystem。应用可以使用Reporter汇报作业执行进度、设置应用层级的状态信息并更新计数器(Counter),或者只是提示作业在运行。

注意，Reducer的输出不会再进行排序。

Reducer数目

合适的reducer数目可以这样估算：(节点数目mapred.tasktracker.reduce.tasks.maximum)乘以0.95 或 乘以1.75。因子为0.95时，当所有map任务完成时所有reducer可以立即启动，并开始从map机器上拉取数据。因子为1.75时,最快的一些节点将完成第一轮reduce处理，此时框架开始启动第二轮reduce任务，这样可以达到比较好的作业负载均衡。提高reduce数目会增加框架的运行负担，但有利于提升作业的负载均衡并降低失败的成本。上述的因子使用最好在作业执行时框架仍然有reduce槽为前提，毕竟框架还需要对作业进行可能的推测执行和失败任务的处理。

不使用Reducer

如果不需要进行化简处理，可以将reduce数目设为0。这种情况下，map的输出会直接写入到文件系统。输出路径通过setOutputPath(Path)指定。框架在写入数据到文件系统之前不再对map结果进行排序。

Partitioner

Partitioner对数据按照key进行分区，从而控制map的输出传输到哪个reducer上。默认的Partitioner算法是hash(哈希。分区数目由作业的reducer数目决定。HashPartitioner 是默认的Partitioner。

Reporter

Reporter为MR应用提供了进度报告、应用状态信息设置，和计数器(Counter)更新等功能.

Mapper和Reducer实现可以使用Reporter汇报进度或者提示作业在正常运行。在一些场景下，应用在处理一些特殊的kv对时耗费了过多时间，这个可能会因为框架假定任务超时而强制停止了这些作业。为避免该情况，可以设置mapred.task.timeout 为一个比较高的值或者将其设置为0以避免超时发生。

应用也可以使用Reporter来更新计数(Counter)。

OutputCollector

OutputCollector是MR框架提供的通用工具来收集Mapper或者Reducer输出数据(中间数据或者最终结果数据)。

Hadoop MapReduce提供了一些经常使用的mapper、reducer和partioner的实现类供我们进行学习。

以上有关configuration和job的部分在新的API中有所改变，简单说就是在Mapper和Reducer中引入了MapContext和ReduceContext，它们封装了configuration和outputcollector，以及reporter。

---

---