搜索
HadoopMap/Reduce教程•目的•先决条件•概述•输入与输出•例子：WordCountv1.0•源代码•用法•解释•Map/Reduce-用户界面•核心功能描述•Mapper•Reducer•Partitioner•Reporter•OutputCollector•作业配置•任务的执行和环境•作业的提交与监控•作业的控制•作业的输入•InputSplit•RecordReader•作业的输出•任务的Side-EffectFile•RecordWriter•其他有用的特性•Counters•DistributedCache•Tool•IsolationRunner•Profiling•调试•JobControl•数据压缩•例子：WordCountv2.0•源代码•运行样例•程序要点目的这篇教程从用户的角度出发，全面地介绍了HadoopMap/Reduce框架的各个方面。先决条件请先确认Hadoop被正确安装、配置和正常运行中。更多信息见：•Hadoop快速入门对初次使用者。•Hadoop集群搭建对大规模分布式集群。概述HadoopMap/Reduce是一个使用简易的软件框架，基于它写出来的应用程序能够运行在由上千个商用机器组成的大型集群上，并以一种可靠容错的方式并行处理上T级别的数据集。一个Map/Reduce作业（job）通常会把输入的数据集切分为若干独立的数据块，由map任务（task）以完全并行的方式处理它们。框架会对map的输出先进行排序，然后把结果输入给reduce任务。通常作业的输入和输出都会被存储在文件系统中。整个框架负责任务的调度和监控，以及重新执行已经失败的任务。通常，Map/Reduce框架和分布式文件系统是运行在一组相同的节点上的，也就是说，计算节点和存储节点通常在一起。这种配置允许框架在那些已经存好数据的节点上高效地调度任务，这可以使整个集群的网络带宽被非常高效地利用。Map/Reduce框架由一个单独的masterJobTracker和每个集群节点一个slaveTaskTracker共同组成。master负责调度构成一个作业的所有任务，这些任务分布在不同的slave上，master监控它们的执行，重新执行已经失败的任务。而slave仅负责执行由master指派的任务。应用程序至少应该指明输入/输出的位置（路径），并通过实现合适的接口或抽象类提供map和reduce函数。再加上其他作业的参数，就构成了作业配置（jobconfiguration）。然后，Hadoop的jobclient提交作业（jar包/可执行程序等）和配置信息给JobTracker，后者负责分发这些软件和配置信息给slave、调度任务并监控它们的执行，同时提供状态和诊断信息给job-client。虽然Hadoop框架是用JavaTM实现的，但Map/Reduce应用程序则不一定要用Java来写。•HadoopStreaming是一种运行作业的实用工具，它允许用户创建和运行任何可执行程序（例如：Shell工具）来做为mapper和reducer。•HadoopPipes是一个与SWIG兼容的C++API（没有基于JNITM技术），它也可用于实现Map/Reduce应用程序。输入与输出Map/Reduce框架运转在key,value键值对上，也就是说，框架把作业的输入看为是一组key,value键值对，同样也产出一组key,value键值对做为作业的输出，这两组键值对的类型可能不同。框架需要对key和value的类(classes)进行序列化操作，因此，这些类需要实现Writable接口。另外，为了方便框架执行排序操作，key类必须实现WritableComparable接口。一个Map/Reduce作业的输入和输出类型如下所示：(input)k1,v1-map-k2,v2-combine-k2,v2-reduce-k3,v3(output)例子：WordCountv1.0在深入细节之前，让我们先看一个Map/Reduce的应用示例，以便对它们的工作方式有一个初步的认识。WordCount是一个简单的应用，它可以计算出指定数据集中每一个单词出现的次数。这个应用适用于单机模式，伪分布式模式或完全分布式模式三种Hadoop安装方式。源代码WordCount.java1.packageorg.myorg;2.3.importjava.io.IOException;4.importjava.util.*;5.6.importorg.apache.hadoop.fs.Path;7.importorg.apache.hadoop.conf.*;8.importorg.apache.hadoop.io.*;9.importorg.apache.hadoop.mapred.*;10.importorg.apache.hadoop.util.*;11.12.publicclassWordCount{13.14.publicstaticclassMapextendsMapReduceBaseimplementsMapperLongWritable,Text,Text,IntWritable{15.privatefinalstaticIntWritableone=newIntWritable(1);16.privateTextword=newText();17.18.publicvoidmap(LongWritablekey,Textvalue,OutputCollectorText,IntWritableoutput,Reporterreporter)throwsIOException{19.Stringline=value.toString();20.StringTokenizertokenizer=newStringTokenizer(line);21.while(tokenizer.hasMoreTokens()){22.word.set(tokenizer.nextToken());23.output.collect(word,one);24.}25.}26.}27.28.publicstaticclassReduceextendsMapReduceBaseimplementsReducerText,IntWritable,Text,IntWritable{29.publicvoidreduce(Textkey,IteratorIntWritablevalues,OutputCollectorText,IntWritableoutput,Reporterreporter)throwsIOException{30.intsum=0;31.while(values.hasNext()){32.sum+=values.next().get();33.}34.output.collect(key,newIntWritable(sum));35.}36.}37.38.publicstaticvoidmain(String[]args)throwsException{39.JobConfconf=newJobConf(WordCount.class);40.conf.setJobName(wordcount);41.42.conf.setOutputKeyClass(Text.class);43.conf.setOutputValueClass(IntWritable.class);44.45.conf.setMapperClass(Map.class);46.conf.setCombinerClass(Reduce.class);47.conf.setReducerClass(Reduce.class);48.49.conf.setInputFormat(TextInputFormat.class);50conf.setOutputFormat(TextOutputFormat.class);.51.52.FileInputFormat.setInputPaths(conf,newPath(args[0]));53.FileOutputFormat.setOutputPath(conf,newPath(args[1]));54.55.JobClient.runJob(conf);57.}58.}59.用法假设环境变量HADOOP_HOME对应安装时的根目录，HADOOP_VERSION对应Hadoop的当前安装版本，编译WordCount.java来创建jar包，可如下操作：$mkdirwordcount_classes$javac-classpath${HADOOP_HOME}/hadoop-${HADOOP_VERSION}-core.jar-dwordcount_classesWordCount.java$jar-cvf/usr/joe/wordcount.jar-Cwordcount_classes/.假设：•/usr/joe/wordcount/input-是HDFS中的输入路径•/usr/joe/wordcount/output-是HDFS中的输出路径用示例文本文件做为输入：$bin/hadoopdfs-ls/usr/joe/wordcount/input//usr/joe/wordcount/input/file01/usr/joe/wordcount/input/file02$bin/hadoopdfs-cat/usr/joe/wordcount/input/file01HelloWorldByeWorld$bin/hadoopdfs-cat/usr/joe/wordcount/input/file02HelloHadoopGoodbyeHadoop运行应用程序：$bin/hadoopjar/usr/joe/wordcount.jarorg.myorg.WordCount/usr/joe/wordcount/input/usr/joe/wordcount/output输出是：$bin/hadoopdfs-cat/usr/joe/wordcount/output/part-00000Bye1Goodbye1Hadoop2Hello2World2应用程序能够使用-files选项来指定一个由逗号分隔的路径列表，这些路径是task的当前工作目录。使用选项-libjars可以向map和reduce的classpath中添加jar包。使用-archives选项程序可以传递档案文件做为参数，这些档案文件会被解压并且在task的当前工作目录下会创建一个指向解压生成的目录的符号链接（以压缩包的名字命名）。有关命令行选项的更多细节请参考Commandsmanual。使用-libjars和-files运行wordcount例子：hadoopjarhadoop-examples.jarwordcount-filescachefile.txt-libjarsmylib.jarinputoutput解释WordCount应用程序非常直截了当。Mapper(14-26行)中的map方法(18-25行)通过指定的TextInputFormat(49行)一次处理一行。然后，它通过StringTokenizer以空格为分隔符将一行切分为若干tokens，之后，输出word,1形式的键值对。对于示例中的第一个输入，map输出是：Hello,1World,1Bye,1World,1第二个输入，map输出是：Hello,1Hadoop,1Goodbye,1Hadoop,1关于组成一个指定作业的map数目的确定，以及如何以更精细的方式去控制这些map，我们将在教程的后续部分学习到更多的内容。WordCount还指定了一个combiner(46行)。因此，每次map运