MapReduce编程实例：单词计数

1、MapReduce编程实例：单词计数本节介绍如何编写基本的MapReduce程序实现数据分析。本节代码是基于Hadoop2.7.3开发的。任务准备单词计数(WordCou nt)的任务是对一组输入文档中的单词进行分别计数。假设文件的量 比较大，每个文档又包含大量的单词，则无法使用传统的线性程序进行处理，而这类问题正 是MapReduce可以发挥优势的地方。在前面MapReduce实例分析：单词计数教程中已经介绍了用MapReduce实现单词 计数的基本思路和具体执行过程。下面将介绍如何编写具体实现代码及如何运行程序。首先，在本地创建3个文件：file00l、file002和file003，文件

2、具体内容如表1所示。表1单词计数输入文件文件名file001fileO02file003文件内容Hello worldConn ected worldOne worldOne dreamHello HadoopHello MapHello Reduce再使用HDFS命令创建一个in put文件目录。hadoop fs -mkdir in put然后，把fileOOl、file002和fileOO3上传到HDFS中的in put目录下。hadoop fs -put file001 in puthadoop fs -put file002 in puthadoop fs -put file003

3、in put编写MapReduce程序的第一个任务就是编写Map程序。在单词计数任务中，Map需 要完成的任务就是把输入的文本数据按单词进行拆分，然后以特定的键值对的形式进行输 出。编写Map 程序Hadoop MapReduce框架已经在类Mapper中实现了 Map任务的基本功能。为了实 现Map任务，开发者只需要继承类Mapper，并实现该类的Map函数。为实现单词计数的Map任务，首先为类Mapper设定好输入类型和输出类型。这里， Map函数的输入是key,value形式，其中，key是输入文件中一行的行号，value是 该行号对应的一行内容。所以，Map函数的输入类型为LongWr

4、itable,Text。Map函数的功能为完成文本分割 工作，Map函数的输出也是key,value形式，其中，key是单词,value为该单词出 现的次数。所以，Map函数的输出类型为Text，LongWritable。以下是单词计数程序的Map任务的实现代码。public static class CoreMapper extends MapperObject,Text,Text,IntWritable private static final IntWritable one = new IntWritable(l);private static Text label = new Text

5、();public void map (Object key, Text value,MapperObject,Text,Text,IntWritable Context context)throws IOException,InterruptedException StringTokenizer tokenizer = new StringTokenizer(value.toString();while (tokenizer. hasMoreTokens() label. set(tokenizer. nextToken();context.writ e (label, one);在上述代码

6、中，实现Map任务的类为CoreMapper。该类首先将需要输出的两个变量one和label进行初始化。变量one的初始值直接设置为1,表示某个单词在文本中出现过。 Map函数的前两个参数是函数的输入参数，value为Text类型，是指每次读入文本的一行，key为Object类型，是指输入的行数据在文本中的行号。StringTokenizer类机器方法将value变量中文本的一行文字进行拆分，拆分后的单词放 在tokenizer列表中。然后程序通过循环对每一个单词进行处理，把单词放在label中， 把one作为单词计数。在函数的整个执行过程中，one的值一直是1。在该实例中，key没有被明显地

7、使用到。con text是Map函数的一种输出方式，通过使用该变量，可以直接将中间结果存储在其中。根据上述代码，Map任务结束后，3个文件的输出结果如表2所示。表2单词计数Map任务输出结果文件名/MapfileOO1/Map1file002/Map2file003/Map3Map任务输出结果编写 Reduce 程序编写MapReduce程序的第二个任务就是编写 Reduce程序。在单词计数任务中，Reduce需要完成的任务就是把输入结果中的数字序列进行求和从而得到每个单词的出现 次数。在执行完Map函数之后，会进入Shuffle阶段，在这个阶段中，MapReduce框架会自 动将Map阶段的

8、输出结果进行排序和分区，然后再分发给相应的Reduce任务去处理。经过Map端Shuffle阶段后的结果如表3所示。表3单词计数Map端Shuffle阶段输出结果文件名/MapfileOO1/Map1file002/Map2fileOO3/Map3Map端Shuffle阶段输出结果world, 0n e, Hello,Reduce端接收到各个Map端发来的数据后，会进行合并，即把同一个key,也就是同一 单词的键值对进行合并，形成key, 形式的输出。经过Map端Shuffle 阶段后的结果如表4所示。表4单词计数Reduce端Shuffle阶段输出结果Reduce 端Hello,Shuffl

9、e阶段输出结果O ne,world, Reduce阶段需要对上述数据进行处理从而得到每个单词的出现次数。从Reduce函数的 输入已经可以理解Reduce函数需要完成的工作就是首先对输入数据value中的数字序 列进行求和。以下是单词计数程序的Reduce任务的实现代码。public static class CoreReducer extendsReducerText, IntWritable, Text, IntWritable 2.private IntWritable count 二 new IntWritable ();public void reduce(Text key,Iter

10、ablevalues, Reducer Context context)throwsIOException, InterruptedException intsum = 0;for(IntWritableintWritable : values)sum+=intWritable. get();&count.set (sum);context.writ e (key, cou nt);与Map任务实现相似,Reduce任务也是继承Hadoop提供的类Reducer并实现其接 口。Reduce函数的输入、输出类型与Map函数的输出类型本质上是相同的。在Reduce函数的开始部分，首先设置sum参数

11、用来记录每个单词的出现次数，然后遍 历value列表，并对其中的数字进行累加，最终就可以得到每个单词总的出现次数。在输 出的时候，乃然使用context类型的变量存储信息。当Reduce阶段结束时，就可以得到最终需要的结果，如表5所示。表5单词计数Reduce任务输出结果Reduce任务输出结果编写main函数为了使用CoreMapper和CoreReducer类进行真正的数据处理，还需要在main函数 中通过Job类设置Hadoop MapReduce程序运行时的环境变量，以下是具体代码。public static void main(String args) throws Exceptio

12、n Configuration conf = new Configuration();String otherArgs = newGenericOptionsParser(conf, args). getRemainingArgsO;if (otherArgs. length != 2) Sys tem.err.pri ntln( Usage:wordcou ntinout)；Sys tem.ex it( 2);& Job job = new Job (conf, WordCount); /设置环境参数job.se tJarByClass (WordCoun t. class); /设置程序的

13、类名9.10.11.12.13.14.15.16.17.18.19.20.1.2.3.4.5.6.7.&9.10.job.se tM apperClass(CoreMapper. class); /添力口 Mapper 类job. se tReducerClass(CoreReducer. class); /添力口 Reducer 类job. se tOu tpu tKeyClass (Tex t. class); /设置输出 key 的类型job. setOutputValueClass (IntWritable. class);/设置输出value的类型FilelnputFormat.ad

14、dInputPath (job, new Path (otherArgs 0);/设置输入文件路径FileOutputFormat.setOutputPath (job, new Path (otherArgs 1);/设置输入文件路径System.exit(job.waitForCompletion(true) ? 0 : 1);代码首先检查参数是不是正确，如果不正确就提醒用户。随后，通过Job类设置环境参数， 并设置程序的类、Mapper类和Reducer类。然后，设置了程序的输出类型，也就是Reduce函数的输出结果key,value中key和value各自的类型。最后，根据程序运 行时

15、的参数，设置输入、输出文件路径。编写MapReduce程序需要引用Hadoop的以下几个核心组件包/它们实现了 HadoopMapReduce 框架。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. T

16、ext;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. FilelnputFormat;import org.apache.hadoop.mapreduce. lib. output.FileOutputFormat;import org.apache.hadoop. ut

17、il.GenericOptionsParser;这些核心组件包的基本功能描述如表6所示。表6 Hadoop Map Reduce核心组件包的基本功能包功能org.apache.hadoop.c onf定义了系统参数的配置文件处理方法org.apache.hadoop.fs定义了抽象的文件系统APIorg.apache.hadoop.mapreduceHadoop MapReduce框架的实现，包括任务的分发调度等org.apache.hadoop.io定义了通用的I/O API，用于网络、数据库和文件数据对象进行读写操作运行代码在运行代码前需要先把当前工作目录设置为/user/local/Ha

18、doop。编译WordCount程 序需要以下3个Jar，为了简便起见，把这3个Jar添加到CLASSPATH中。$exportCLASSPATH二/usr/local/hadoop/share/hadoop/comm on/hadoop-comm on-jar:$CLASSPATH$exportCLASSPATH二/usr/local/hadoop/share/hadoop/mapreduce/hadoop-mapreduce-jar:$CLAS SPATH$exportCLASSPATH二/usr/local/hadoop/share/hadoop/comm on/lib/comm on-cli-1.2.jar:$CLASSPATH使用JDK包中的工具对代码进行编译。$ javac WordCo