hadoop Map/Reduce教程(2)

Map/Reduce用户界面

本节为用户采用框架要面对的各个环节提供了详细的描述,旨在与帮助用户对实现、配置和调优进行详细的设置。然而,开发时候还是要对应着API进行相关操作。

首先我们需要了解Mapper和Reducer接口,应用通常需要提供map和reduce方法以实现他们。

接着我们需要对JobConf, JobClient,Partitioner,OutputCollector,Reporter,InputFormat,OutputFormat,OutputCommitter等进行讨论。

最后,我们将通过讨论框架中一些有用的功能点(例如:DistributedCache,IsolationRunner等等)。

核心功能描述

应用程序通常会通过提供map和reduce来实现Mapper和Reducer接口,它们组成作业的核心。

Mapper

mapper对输入的键值对映射成一组中间格式的键值对。

映射是一个独立的任务,它将输入记录集转换为中间格式记录集。这种转换的中间格式记录不需要与输入的记录集类型一致,一个输入的键值可以对应多个输出的键值对。

Hadoop Map/Reduce为每个由作业中InputFormat产生的InputSplit产生一个map任务。

总的来说,对Mapper的实现者需要重写JobConfigurable.configure(JobConf)方法,此方法需要传递JobConf参数,以此来完成mapper的初始化。接着,框架调用map方法,对任务中的InputSplit中每个key/value pair调用以此。应用程序通过重写Closable.close()来执行必要的清理工作。

输出的键值对不需要跟输入的键值对的类型一致。输入的键值可能映射成0到多个输出的键值对,然后调用OutputCollector.collect(WritableComparable,Writable)来收集输出的键值对。

应用程序可以使用Reporter报告进度,设定应用级别的状态消息,更新Counters(计数器),或者仅是表明自己运行正常。

框架随后会把与一个特定key关联的所有中间过程的值(value)分成组,然后把它们传给Reducer以产出最终的结果。用户可以通过 JobConf.setOutputKeyComparatorClass(Class)来指定具体负责分组的Comparator。

Mapper的输出被排序后,就被划分给每个Reducer。分块的总数目和一个作业的reduce任务的数目是一样的。用户可以通过实现自定义的Partitioner来控制哪个key被分配给哪个Reducer。

用户可选择通过 JobConf.setCombinerClass(Class)指定一个combiner,它负责对中间过程的输出进行本地的聚集,这会有助于降低从Mapper到Reducer数据传输量。

这些被排好序的中间过程的输出结果保存的格式是(key-len, key, value-len, value),应用程序可以通过JobConf控制对这些中间结果是否进行压缩以及怎么压缩,使用哪种 CompressionCodec。

需要多少个Map?

Map的数目通常是由输入数据的大小决定的,一般就是所有输入文件的总块(block)数。

Map正常的并行规模大致是每个节点(node)大约10到100个map,对于CPU 消耗较小的map任务可以设到300个左右。由于每个任务初始化需要一定的时间,因此,比较合理的情况是map执行的时间至少超过1分钟。

这样,如果你输入10TB的数据,每个块(block)的大小是128MB,你将需要大约82,000个map来完成任务,除非使用 setNumMapTasks(int)(注意:这里仅仅是对框架进行了一个提示(hint),实际决定因素见)将这个数值设置得更高。

Reducer

对于reducer,官方给出的说明为:

Reducer reduces a set of intermediate values which share a key to a smaller set of values.

大意是Reducer对中间的值集合转换成一个key对应一个更小的数据集。

Reducer的个数取决于用户设置,用户通过JobConf.setNumReduceTasks(int)来设置。

总的来说,Reducer的实现需要通过重写JobConfigurable.configure(JobConf)方法,这个方法需要传递一个JobConf参数,目的是完成Reducer的初始化工作。然后,框架为成组的输入数据中的每个<key, (list of values)>对调用一次 reduce(WritableComparable, Iterator, OutputCollector, Reporter)方法。之后,应用程序可以通过重写Closeable.close()来执行相应的清理工作。

Reducer有3个主要阶段:shuffle、sort和reduce。

Shuffle

reducer的输入对应的是mapper的已排序的输出。

Sort

框架在此阶段根据输入key的值对reducer的输入进行分组(因为不同mapper的输出中可能会有相同的key);

Shuffle和sort两个阶段是同时进行的;map的输出也是边取回边合并的。

Secondary Sort

如果需要中间过程对key的分组规则和reduce前对key的分组规则不同,那么可以通过 JobConf.setOutputValueGroupingComparator(Class)来指定一个Comparator。再加上 JobConf.setOutputKeyComparatorClass(Class)可用于控制中间过程的key如何被分组,所以结合两者可以实现按值的二次排序。

Reduce

本阶段框架为已分组的输入数据中的每个 <key, (list of values)>对调用一次 reduce(WritableComparable, Iterator, OutputCollector, Reporter)方法。

reduce任务的输出通常是通过调用OutputCollector.collect(WritableComparable, Writable)来写入文件系统的。

应用可以利用Reporter来报告进度,设置程序级别状态消息和更新计数器,或是仅仅告知程序运行正常。

Reducer的输出没有排序处理。

需要多少Reduce

Reduce的数目建议是0.95或1.75乘以 (<no. of nodes> *mapred.tasktracker.reduce.tasks.maximum)。

用0.95,所有reduce可以在maps一完成时就立刻启动,,开始传输map的输出结果。用1.75,速度快的节点可以在完成第一轮reduce任务后,可以开始第二轮,这样可以得到比较好的负载均衡的效果。

增加reduce的数目会增加整个框架的开销,但可以改善负载均衡,降低由于执行失败带来的负面影响。

上述比例因子比整体数目稍小一些是为了给框架中的推测性任务(speculative-tasks) 或失败的任务预留一些reduce的资源。

无Reducer

如果没有归约要进行,那么设置reduce任务的数目为零是合法的。

这种情况下,map任务的输出会直接被写入由setOutputPath(Path)指定的输出路径。框架在把它们写入FileSystem之前没有对它们进行排序。

Partitioner

Partitioner对值空间进行划分。

绊脚石乃是进身之阶。

hadoop Map/Reduce教程(2)

相关文章:

你感兴趣的文章:

标签云: