MapReduce经典例子WordCount运行详解.doc

wordHadoop MapReduce经典例子WordCount 运行详解1、MapReduce理论简介1.1 MapReduce编程模型 MapReduce 采用“分而治之的思想，把对大规模数据集的操作，分发给一个主节点管理下的各个分节点共同完成，然后通过整合各个节点的中间结果，得到最终结果。简单地说，MapReduce 就是“任务的分解与结果的汇总。在Hadoop 中，用于执行MapReduce 任务的机器角色有两个：一个是JobTracker；另一个是TaskTracker，JobTracker 是用于调度工作的，TaskTracker 是用于执行工作的。一个Hadoop集群中只有一台JobTracker。在分布式计算中，MapReduce 框架负责处理了并行编程中分布式存储、工作调度、负载均衡、容错均衡、容错处理以与网络通信等复杂问题，把处理过程高度抽象为两个函数：map 和reduce，map 负责把任务分解成多个任务，reduce 负责把分解后多任务处理的结果汇总起来。需要注意的是，用MapReduce 来处理的数据集或任务必须具备这样的特点：待处理的数据集可以分解成许多小的数据集，而且每一个小数据集都可以完全并行地进展处理。1.2 MapReduce处理过程在Hadoop 中，每个MapReduce 任务都被初始化为一个Job，每个 Job 又可以分为两种阶段：map 阶段和reduce 阶段。这两个阶段分别用两个函数表示，即map 函数和reduce 函数。map 函数接收一个<key,value>形式的输入，然后同样产生一个<key,value>形式的中间输出，Hadoop 函数接收一个如<key,(list of values)>形式的输入，然后对这个 value 集合进展处理，每个reduce 产生0 或1 个输出，reduce 的输出也是<key,value>形式的。MapReduce 处理大数据集的过程116 / 16创建时间：2012/3/1修改时间：2017/3/1 修改次数：02、运行WordCount程序单词计数是最简单也是最能表现MapReduce 思想的程序之一，可以称为MapReduce 版“Hello World，该程序的完整代码可以在Hadoop 安装包的“src/examples目录下找到。单词计数主要完成功能是：统计一系列文本文件中每个单词出现的次数，如如下图所示。2.1 准备工作现在以“hadoop普通用户登录服务器。1创建本地示例文件首先在“/home/hadoop目录下创建文件夹“file。接着创建两个文本文件file1.txt 和， file1.txt 内容为使“Hello World 而，的内容为“Hello Hadoop。2创建时间2012/3/1修改时间：2017/3/1 修改次数：02在 HDFS 上创建输入文件夹3上传本地 file 中文件到集群的input 目录下2.2 运行例子1在集群上运行 WordCount 程序备注：以input 作为输入目录，output 目录作为输出目录。已经编译好的WordCount 的Jar 在“/usr/hadoop下面， 就是，所以在下面执行命令时记得把路径写全了，不然会提示找不到该Jar 包。2MapReduce 执行过程显示信息Hadoop 命令会启动一个JVM 来运行这个MapReduce 程序，并自动获得Hadoop 的配置，同时把类的路径与其依赖关系参加到Hadoop 的库中。以上就是Hadoop Job 的运行记录，从这里可以看到，这个Job 被赋予了一个ID 号：job_201202292213_0002，而且得知输入文件有两个Total input paths to process : 2，同时还可以了解map 的输入输出记录record 数与字节数，以与reduce 输入输出记录。比如说，在本例中，map 的task 数量是2 个，reduce3创建时间：2012/3/1修改时间：2012/3/1 修改次数：0的task 数量是一个。map 的输入record 数是2 个，输出record 数是4 个等信息。2.3 查看结果1查看 HDFS 上output 目录内容从上图中知道生成了三个文件，我们的结果在“part-r-00000中。2查看结果输出文件内容3、WordCount源码分析3.1 特别数据类型介绍Hadoop 提供了如下内容的数据类型，这些数据类型都实现了Writableparable 接口，以便用这些类型定义的数据可以被序列化进展网络传输和文件存储，以与进展大小比拟。BooleanWritable：标准布尔型数值ByteWritable：单字节数值DoubleWritable：双字节数FloatWritable：浮点数IntWritable：整型数LongWritable：长整型数Text：使用 UTF8 格式存储的文本NullWritable：当<key,value>中的 key 或value 为空时使用3.2 旧的WordCount分析1源代码程序package org.apache.hadoop.examples;4创建时间：2012/3/1修改时间：2012/3/1 修改次数：0import java.io.IOException;import java.util.Iterator;import java.util.StringTokenizer;import org.apache.hadoop.fs.Path;import org.apache.hadoop.io.IntWritable;import org.apache.hadoop.io.LongWritable;import org.apache.hadoop.io.Text;import org.apache.hadoop.mapred.FileInputFormat;import org.apache.hadoop.mapred.FileOutputFormat;import org.apache.hadoop.mapred.JobClient;import org.apache.hadoop.mapred.JobConf;import org.apache.hadoop.mapred.MapReduceBase;import org.apache.hadoop.mapred.Mapper;import org.apache.hadoop.mapred.OutputCollector;import org.apache.hadoop.mapred.Reducer;import org.apache.hadoop.mapred.Reporter;import org.apache.hadoop.mapred.TextInputFormat;import org.apache.hadoop.mapred.TextOutputFormat;publicclassWordCountpublicprivateprivatestatic class Map extends MapReduceBaseMapper<LongWritable, Text, Text, IntWritable> finalstatic IntWritable one = newText word = new Text();implementsIntWritable( 1 );publicvoid map(LongWritable key, Text value,OutputCollector<Text, IntWritable> output, Reporter reporter)throws IOExceptionString line = value.toString();StringTokenizer tokenizer = new StringTokenizer(line);while (tokenizer.hasMoreTokens() word.set(tokenizer.nextToken(); output.collect(word, one);5创建时间：2012/3/1修改时间：2012/3/1 修改次数：0publicstatic class Reduce extends MapReduceBaseReducer<Text, IntWritable, Text, IntWritable>implementspublicvoid reduce(Text key, Iterator<IntWritable> values,OutputCollector<Text, IntWritable> output, Reporter reporter)throws IOExceptionint sum = 0 ;while (values.hasNext() sum += values.next().get();output.collect(key, new IntWritable(sum);public static void main(String args) throws Exception JobConf conf = new JobConf(WordCount. class ); conf.setJobName("wordcount" );conf.setOutputKeyClass(Text.class );conf.setOutputValueClass(IntWritable.class );conf.setMapperClass(Map.class );conf.setbinerClass(Reduce.class );conf.setReducerClass(Reduce.class );conf.setInputFormat(TextInputFormat.class );conf.setOutputFormat(TextOutputFormat.class );FileInputFormat.setInputPaths(conf, new Path(args 0 );FileOutputFormat.setOutputPath(conf, new Path(args 1 );JobClient.runJob(conf);3主方法Main 分析public staticvoidmain(String args) throws ExceptionJobConf conf = new JobConf(WordCount. class );conf.setJobName("wordcount" );6创建时间：2012/3/1修改时间：2012/3/1 修改次数：0conf.setOutputKeyClass(Text.class );conf.setOutputValueClass(IntWritable.class );conf.setMapperClass(Map.class );conf.setbinerClass(Reduce.class );conf.setReducerClass(Reduce.class );conf.setInputFormat(TextInputFormat.class );conf.setOutputFormat(TextOutputFormat.class );FileInputFormat.setInputPaths(conf, new Path(args 0 );FileOutputFormat.setOutputPath(conf, new Path(args 1 );JobClient.runJob(conf);首先讲解一下Job 的初始化过程。main 函数调用Jobconf 类来对MapReduce Job 进展初始化，然后调用setJobName()方法命名这个 Job。对Job 进展合理的命名有助于更快地找到Job，以便在 JobTracker 和Tasktracker 的页面中对其进展监视。JobConf conf = new JobConf(WordCount. class );conf.setJobName("wordcount" );接着设置Job 输出结果<key,value>的中key 和value 数据类型，因为结果是<单词,个数>，所以key 设置为“Text类型，相当于Java 中String 类型。Value 设置为“IntWritable，相当于Java 中的int 类型。conf.setOutputKeyClass(Text.class );conf.setOutputValueClass(IntWritable.class );然后设置Job 处理的Map拆分、biner中间结果合并以与Reduce合并的相关处理类。这里用Reduce 类来进展Map 产生的中间结果合并，防止给网络数据传输产生压力。conf.setMapperClass(Map.class );conf.setbinerClass(Reduce.class );conf.setReducerClass(Reduce.class );接着就是调用setInputPath()和 setOutputPath()设置输入输出路径。conf.setInputFormat(TextInputFormat.class );conf.setOutputFormat(TextOutputFormat.class );7创建时间：2012/3/1修改时间：2012/3/1 修改次数：01InputFormat 和InputSplitInputSplit 是Hadoop 定义的用来传送给每个单独的map 的数据，InputSplit 存储的并非数据本身，而是一个分片长度和一个记录数据位置的数组。生成InputSplit 的方法可以通过InputFormat()来设置。当数据传送给map 时，map 会将输入分片传送到InputFormat，InputFormat 如此调用方法getRecordReader()生成 RecordReader，RecordReader 再通过creatKey()、creatValue()方法创建可供map 处理的<key,value>对。简而言之，InputFormat()方法是用来生成可供 map处理的<key,value>对的。Hadoop 预定义了多种方法将不同类型的输入数据转化为map 能够处理的<key,value>对，它们都继承自InputFormat，分别是：InputFormat | |-posableInputFormat |-positeInputFormat |-DBInputFormat |-FileInputFormat |-bineFileInputFormat |-KeyValueTextInputFormat |-NLineInputFormat |-SequenceFileInputFormat |-TeraInputFormat |-TextInputFormat其中TextInputFormat 是Hadoop 默认的输入方法，TextInputFormat 中，在每个文件或其一局部都会单独地作为map 的输入，而这个是继承自FileInputFormat 的。之后，每行数据都会生成一条记录，每条记录如此表示成<key,value>形式：l key 值是每个数据的记录在数据分片中字节偏移量，数据类型是LongWritable；l value 值是每行的内容，数据类型是Text。2OutputFormat每一种输入格式都有一种输出格式与其对应。默认的输出格式是TextOutputFormat，这种输出方式与输入类似，会将每条记录以一行的形式存入文本文件。不过，它的键和值可以是任意形式的，因为程序内容会调用toString()方法将键和值转换为String 类型再输出。3Map 类中map 方法分析publicprivateprivate finalstatic IntWritable one = newText word = new Text();IntWritable( 1 );staticclass Map extends MapReduceBase implements Mapper<LongWritable, Text, Text, IntWritable>8创建时间：2012/3/1修改时间：2012/3/1 修改次数：0publicvoid map(LongWritable key, Text value,OutputCollector<Text, IntWritable> output, Reporter reporter)throws IOExceptionString line = value.toString();StringTokenizer tokenizer = new StringTokenizer(line);while (tokenizer.hasMoreTokens() word.set(tokenizer.nextToken(); output.collect(word, one); Map 类继承自MapReduceBase，并且它实现了Mapper 接口，此接口是一个规X类型，它有4 种形式的参数，分别用来指定map 的输入key 值类型、输入value 值类型、输出key值类型和输出value 值类型。在本例中，因为使用的是TextInputFormat，它的输出 key 值是LongWritable 类型，输出value 值是Text 类型，所以map 的输入类型为<LongWritable,Text>。在本例中需要输出<word,1>这样的形式，因此输出的key 值类型是Text，输出的 value 值类型是IntWritable。实现此接口类还需要实现map 方法，map 方法会具体负责对输入进展操作，在本例中，map 方法对输入的行以空格为单位进展切分，然后使用OutputCollect 收集输出的<word,1>。4Reduce 类中reduce 方法分析publicpublicvoid reduce(Text key, Iterator<IntWritable> values,OutputCollector<Text, IntWritable> output, Reporter reporter)throws IOExceptionstaticclass Reduce extends MapReduceBase implementsReducer<Text, IntWritable, Text, IntWritable>int sum = 0 ;while (values.hasNext() sum += values.next().get();output.collect(key, new IntWritable(sum);Reduce 类也是继承自MapReduceBase 的，需要实现Reducer 接口。Reduce 类以map的输出作为输入，因此Reduce 的输入类型是<Text，Intwritable>。而Reduce 的输出是单词9和它的数目，因此，它的输出类型是<Text,IntWritable>。Reduce 类也要实现reduce 方法，在此方法中，reduce 函数将输入的key 值作为输出的key 值，然后将获得多个value 值加起来，作为输出的值。3.3 新的WordCount分析1源代码程序package org.apache.hadoop.examples;import java.io.IOException;import java.util.StringTokenizer;import org.apache.hadoop.conf.Configuration;import org.apache.hadoop.fs.Path;import org.apache.hadoop.io.IntWritable;import org.apache.hadoop.io.Text;import org.apache.hadoop.mapreduce.Job;import org.apache.hadoop.mapreduce.Mapper;import org.apache.hadoop.mapreduce.Reducer;import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;import org.apache.hadoop.util.GenericOptionsParser;public class WordCount public static class TokenizerMapperextends Mapper<Object, Text, Text, IntWritable>private final static IntWritable one = new IntWritable(1);private Text word = new Text();public void map(Object key, Text value, Context context ) throws IOException, InterruptedException StringTokenizer itr = new StringTokenizer(value.toString(); while (itr.hasMoreTokens() word.set(itr.nextToken(); context.write(word, one); public static class IntSumReducerextends Reducer<Text,IntWritable,Text,IntWritable> 10创建时间：2012/3/1修改时间：2012/3/1 修改次数：0private IntWritable result = new IntWritable();public void reduce(Text key, Iterable<IntWritable> values, Context context ) throws IOException, InterruptedException int sum = 0; for (IntWritable val : values) sum += val.get(); result.set(sum); context.write(key, result);public static void main(String args) throws Exception Configuration conf = new Configuration(); String otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs(); if (otherArgs.length != 2) System.err.println("Usage: wordcount <in> <out>"); System.exit(2); Job job = new Job(conf, "word count"); job.setJarByClass(WordCount.class); job.setMapperClass(TokenizerMapper.class); job.setbinerClass(IntSumReducer.class); job.setReducerClass(IntSumReducer.class); job.setOutputKeyClass(Text.class); job.setOutputValueClass(IntWritable.class); FileInputFormat.addInputPath(job, new Path(otherArgs0); FileOutputFormat.setOutputPath(job, new Path(otherArgs1); System.exit(job.waitForpletion(true) ? 0 : 1);1Map 过程public static class TokenizerMapperextends Mapper<Object, Text, Text, IntWritable>private final static IntWritable one = new IntWritable(1);private Text word = new Text();public void map(Object key, Text value, Context context ) throws IOException, InterruptedException 11创建时间：2012/3/1修改时间：2012/3/1 修改次数：0StringTokenizer itr = new StringTokenizer(value.toString();while (itr.hasMoreTokens() word.set(itr.nextToken(); context.write(word, one); Map 过程需要继承org.apache.hadoop.mapreduce 包中Mapper 类，并重写其map 方法。通过在map 方法中添加两句把key 值和value 值输出到控制台的代码，可以发现map 方法中value 值存储的是文本文件中的一行以回车符为行完毕标记，而key 值为该行的首字母相对于文本文件的首地址的偏移量。然后StringTokenizer 类将每一行拆分成为一个个的单词，并将<word,1>作为map 方法的结果输出，其余的工作都交有MapReduce 框架处理。2Reduce 过程public static class IntSumReducerextends Reducer<Text,IntWritable,Text,IntWritable> private IntWritable result = new IntWritable();public void reduce(Text key, Iterable<IntWritable> values, Context context ) throws IOException, InterruptedException int sum = 0; for (IntWritable val : values) sum += val.get(); result.set(sum); context.write(key, result);Reduce 过程需要继承org.apache.hadoop.mapreduce 包中Reducer 类，并重写其reduce方法。Map 过程输出<key,values>中key 为单个单词，而values 是对应单词的计数值所组成的列表，Map 的输出就是Reduce 的输入，所以reduce 方法只要遍历values 并求和，即可得到某个单词的总次数。3执行 MapReduce 任务public static void main(String args) throws Exception Configuration conf = new Configuration(); String otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs(); if (otherArgs.length != 2) System.err.println("Usage: wordcount <in> <out>");某某工业大学软件工程与理论实验室编辑：虾皮12创建时间：2012/3/1修改时间：2012/3/1 修改次数：0System.exit(2);Job job = new Job(conf, "word count");job.setJarByClass(WordCount.class);job.setMapperClass(TokenizerMapper.class);job.setbinerClass(IntSumReducer.class);job.setReducerClass(IntSumReducer.class);job.setOutputKeyClass(Text.class);job.setOutputValueClass(IntWritable.class);FileInputFormat.addInputPath(job, new Path(otherArgs0);FileOutputFormat.setOutputPath(job, new Path(otherArgs1);System.exit(job.waitForpletion(true) ? 0 : 1);在MapReduce 中，由Job 对象负责管理和运行一个计算任务，并通过Job 的一些方法对任务的参数进展相关的设置。此处设置了使用TokenizerMapper 完成Map 过程中的处理和使用IntSumReducer 完成bine 和Reduce 过程中的处理。还设置了Map 过程和Reduce过程的输出类型：key 的类型为Text，value 的类型为IntWritable。任务的输出和输入路径如此由命令行参数指定，并由FileInputFormat 和FileOutputFormat 分别设定。完成相应任务的参数设定后，即可调用job.waitForpletion()方法执行任务。4、WordCount处理过程本节将对WordCount 进展更详细的讲解。详细执行步骤如下：1将文件拆分成 splits，由于测试用的文件较小，所以每个文件为一个 split，并将文件按行分割形成<key,value>对，如图4-1 所示。这一步由MapReduce 框架自动完成，其中偏移量即key 值包括了回车所占的字符数Windows 和Linux 环境会不同。图4-1 分割过程2将分割好的<key,value>对交给用户定义的 map 方法进展处理，生成新的<key,value>对，如图4-2 所示。13分割结果map方法输出<Hello,1><World,1> <Bye,1><World,1> <Hello,1><Hadoop,1> <Bye,1><Hadoop,1><0,“Hello World><12,“Bye World>map()<0,“Hello Hadoop> <13,“Bye Hadoop>map()图4-2 执行map 方法 3得到 map 方法输出的<key,value>对后，Mapper 会将它们按照key 值进展排序，并执行bine 过程，将key 至一样value 值累加，得到Mapper 的最终输出结果。如图4-3 所示。map方法输出<Hello,1><World,1> <Bye,1><World,1> <Hello,1><Hadoop,1> <Bye,1><Hadoop,1>排序结果 <Bye,1><Hello,1><World,1><World,1> <Bye,1><Hadoop,1><Hadoop,1><Hello,1>bine输出 <Bye,1><Hello,1><World,2>Map端排序bine过程Map端排序bine过程 <Bye,1><Hadoop,2> <Hello,1>图4-3 Map 端排序与bine