mapreduce

private

IntWritable

result

=

new

IntWritable

public

void

reduce

(Textkey,Iterablevalues,

Contextcontext

)

throws

int

sum

=

0

for

sum+=val.get();

}

result.set(sum);

context.write(key,result);

}

}

public

static

void

main

(String[]args)

throws

Configuration

conf

=

new

Configuration

Job

job

=

“wordcount”

job.setJarByClass(WordCount.class);

job.setMapperClass(TokenizerMapper.class);

job.setCombinerClass(IntSumReducer.class);

job.setReducerClass(IntSumReducer.class);

job.setOutputKeyClass(Text.class);

job.setOutputValueClass(IntWritable.class);

FileInputFormat.addInputPath(job,new

Path

0

FileOutputFormat.setOutputPath(job,new

Path

1

System.exit(job.waitForCompletion(true

1

}

}

运行：

假设环境变量设置如下：

exportJAVA_HOME=

/

/

/

default

=

/

=

/

/

编译WordCount.java且创建一个jar:

$bin/hadoopcom.sun.tools.javac.MainWordCount.java

$jarcfwc.jarWordCount*.class

 假设：

/user/joe/wordcount/input–HDFS

/user/joe/wordcount/output–HDFS

作为输入的样例文本文件：

$bin/hadoopfs-ls/user/joe/wordcount/input//user/joe/wordcount/input/file01/user/joe/wordcount/input/file02

$bin/hadoopfs-cat/user/joe/wordcount/input/file01

HelloWorldByeWorld

$bin/hadoopfs-cat/user/joe/wordcount/input/file02

HelloHadoopGoodbyeHadoop

运行应用：

  $bin/hadoopjarwc.jarWordCount/user/joe/wordcount/input/user/joe/wordcount/output

输出：

 $bin/hadoopfs-cat/user/joe/wordcount/output/part-r-00000`

Bye1

Goodbye1

Hadoop2

Hello2

World2

应用能够通过使用选项–files来在该任务当前工作目录下指定多个以逗号分开的路径列表。-libjars选项允许应用给map和reduce的类路径增加jar。选项–archives允许传递以逗号分开的备份作为参数。

运行带有-libjars,-files和-archives的wordcount例子：

   bin/hadoopjarhadoop-mapreduce-examples-.jarwordcount-filescachefile.txt-libjarsmylib.jar-archivesmyarchive.zipinputoutput

其中，myarchive.zip会解压放在已“myarchive.zip”的名称的路径下。

用户能够使用

-files

-archives

比如：

bin/hadoopjarhadoop-mapreduce-examples-.jarwordcount-filesdir1/dict.txtdict1,dir2/dict.txtdict2-archivesmytar.tgztgzdirinputoutput

解析（walk-through

）

WordCount

public

void

map

(Objectkey,Textvalue,Contextcontext

)

throws

StringTokenizer

itr

=

new

StringTokenizer

while

word.set(itr.nextToken());

context.write(word,one);

}

}

Mapper通过map方法实现，在指定的textInputFormat的辅助下，一次处理一行。然后，通过StringTokenizer,将行按空格拆分成字符，且输出类似<,1>的键值对。

对于给定的例子，第一个map的输出：

第二个map输出：

我们已经了解到了为一个给定job而衍生的很多map.至于如何以一种细粒度的方式控制它们，将会在手册的后面介绍。

job.setCombinerClass(IntSumReducer.class);

WordCount同样指定了一个合并器。因此，在基于key值排序后，每一个map的输出都会传递给本地的合并器以实现本地聚合。

第一个map的输出：

`

第二个map的输出：

`

public

void

reduce

(Textkey,Iterablevalues,

Contextcontext

)

throws

int

sum

=

0

for

sum+=val.get();

}

result.set(sum);

context.write(key,result);

}

Reducer通过reduce方法实现，目的是对值进行求和，也就是是计算每个键出现的次数（比如，在这个例子中就是单词）

因此，job的输出为：

`

main方法指定job的不同方面，比如输入/输出路径（通过命令行传递），键/值类型，输入/输出格式等等。

在手册的后续我们将会学习更多关于Job,InputFormat,OutoutFormat和其他接口和类。

MapReduce-用户接口

这部分介绍MapReduce框架用户方面的一些细节。这能够帮助用户更细致的实施、配置和调节它们的job.但是，每个类/接口的javadoc依然是最详细可用的文档；这个仅仅是作为一个手册。

 让我们首先来看看Mapper和Reducer接口。应用通常通过提供map和reduce方法来实现它们。

然后我们将会讨论其他核心接口，包括Job,Partitioner,InputFormat,OutputFormat和其他等。

最后，我们会讨论框架的一些用户特性，比如DistributedCache,IsolationRunner等。

应用通常实现Mapper和Reducer接口来提供map和reduce方法。这些组成了job的核心。

Mapper

Mapper将输入的键/值对映射成一个中间键值对集。

映射是将输入记录转换成中间记录的单个行为。转换后的中间记录并不需要和输入记录类型相同。一个给定的输入对可能会映射成0个或更多的输出对。

HadoopMapReduce框架给通过InputFormat产生的每个InputSplit衍生一个map任务。

总的来说，Mapper实现通过Job.setMapperClass(Class)方法传递给Job。进而，框架为InputSplit中的每个键/值对调用map(WritableComparable,Writable,Context)。应用能够重载cleanup(Context)方法来操作任何需要的清理。

输出（键值）对的类型不需要和输入对的类型相同。一个给定的输入对可能会映射成0个或更多的输出对。通过调用context.write(WritableComparable,Writable)来收集输出对。

应用能够使用Counter来报告它们的统计。

所有关联一个给定输出键的中间值随后会被框架分组，进而传递给Reducer来决定最后的输出。用户能够通过Job.setGroupingComparatorClass(Class)指定一个Comparator来控制分组。

Mapper输出会被排序进而分发给Reducer。分块的数量和job中reduce任务一样。用户能够通过实现一个订制的Partitioner来控制那些键（也就是记录）分发给哪些Reducer。

用户能够选择通过Job.setCombinerClass(Class)指定一个combiner来对中间结果进行本地聚合，以帮助减少从Mapper传递给Reducer的数据的数量。

中间以及排序后的输出总是以一种简单的（键长，值长，值）格式存储。应用能够控制是否，以及如何对输出进行压缩。且通过Configuration使用CompressionCodec。

有多少map

Map的数量通常决定于输入的总尺寸，也就是输入文件的块的总数。

Map并行性的合适层级一般设置在每个节点大概10-100个maps，尽管对于很多不是很耗CPU资源的map任务可以设置高达300个map.任务启动需要花点时间，因此map花几分钟后再执行的话是最好的。

因此，如果有10TB的输入数据，且block(块)的尺寸为128MB的话，你将获得配备82000个map.除非你通过Configuration.set(MRJobConfig.NUM_MAPS,int)来将它设置的更高。

Reducer

Reducer减少共享一个键的中间值，得到一个更小的值集。

用户通过Job.setNumReduceTasks(int)来设置job的reduce的数量。

总的来说，实施的Reducer通过Job.setReducerClass(Class)传递给Job。然后，框架给分组后的输入中的每一对调用reduce(WritableCoparable,Iterable,Context)方法。应用能够重载cleanup(Context)方法来操作任何所需的清理。

Reducer有3个主要阶段：清洗、排序和简化。