首页 > 代码库 > WordCount示例深度学习MapReduce过程(1)
WordCount示例深度学习MapReduce过程(1)
我们都安装完Hadoop之后,按照一些案例先要跑一个WourdCount程序,来测试Hadoop安装是否成功。在终端中用命令创建一个文件夹,简单的向两个文件中各写入一段话,然后运行Hadoop,WourdCount自带WourdCount程序指令,就可以输出写入的那句话各个不同单词的个数。但是这不是这篇博客主要讲的内容,主要是想通过一个简单的Wordcount程序,来认识Hadoop的内部机制。并通过此来深入了解MapReduce的详细过程。在Thinking in BigDate(八)大数据Hadoop核心架构HDFS+MapReduce+Hbase+Hive内部机理详解中我们已经很大概梳理一下,Hadoop内部集群架构,并对MapReduce也有初步的了解,这里我们以WourdCount程序来深入的探讨MapReduce的过程。
利用命令行,测试WourdCount程序:
WourdCount程序就是统计文本中字母的个数
1、创建Wordcount示例文件
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 | zhangzhen@ubuntu:~/software$ mkdir input zhangzhen@ubuntu:~/software$ cd input/ zhangzhen@ubuntu:~/software/input$ echo "I am zhangzhen" >test1.txt zhangzhen@ubuntu:~/software/input$ echo "You are not zhangzhen" >test2.txt zhangzhen@ubuntu:~/software/input$ cd ../hadoop-1.2.1/ zhangzhen@ubuntu:~/software/hadoop-1.2.1$ cd bin zhangzhen@ubuntu:~/software/hadoop-1.2.1/bin$ ls hadoop slaves.sh start-mapred.sh stop-mapred.sh hadoop-config.sh start- all .sh stop- all .sh task-controller hadoop-daemon.sh start-balancer.sh stop-balancer.sh hadoop-daemons.sh start-dfs.sh stop-dfs.sh rcc start-jobhistoryserver.sh stop-jobhistoryserver.sh zhangzhen@ubuntu:~/software/hadoop-1.2.1/bin$ jps(确定Hadoop已经起来了) 7101 SecondaryNameNode 7193 JobTracker 7397 TaskTracker 9573 Jps 6871 DataNode 6667 NameNode zhangzhen@ubuntu:~/software/hadoop-1.2.1/bin$ cd .. zhangzhen@ubuntu:~/software/hadoop-1.2.1$ ls bin data hadoop-minicluster-1.2.1.jar libexec share build.xml docs hadoop-test-1.2.1.jar LICENSE.txt src c++ hadoop-ant-1.2.1.jar hadoop-tools-1.2.1.jar logs webapps CHANGES.txt hadoop-client-1.2.1.jar ivy NOTICE.txt conf hadoop-core-1.2.1.jar ivy.xml README.txt contrib hadoop-examples-1.2.1.jar lib sbin zhangzhen@ubuntu:~/software/hadoop-1.2.1$ bin/hadoop dfs -put ../input in //把文件上传的hdfa中的 in 目录中,其实这个说法有误 zhangzhen@ubuntu:~/software/hadoop-1.2.1$ bin/hadoop dfs -ls . in /* ls: Cannot access . in /*: No such file or directory. zhangzhen@ubuntu:~/software/hadoop-1.2.1$ bin/hadoop dfs -ls ./ in /* -rw-r --r-- 1 zhangzhen supergroup 15 2014-03-22 10:45 /user/zhangzhen/in/test1.txt -rw-r --r-- 1 zhangzhen supergroup 22 2014-03-22 10:45 /user/zhangzhen/in/test2.txt |
注意:Hadoop中是没有当前目录这个概念的。所以上传到hdfs中的文件,我们是不能通过cd命令、ls命令,查看目录中的文件。这里我们通过就是上面和下面命令查看hdfs中文件的方法。
在每个版本中,hadoop-examples-1.2.1.jar的位置不一样,在Hadoop1.2.1版本中,我们hadoop-examples-1.2.1.jar文件是在Hadoop目录中的,这里我们需要把这个hadoop-examples-1.2.1.jar拷贝到/bin 目录中。
执行:利用hadoop-examples-1.2.1.jar执行bin目录下in目录中的文件,并把结果写入到 put 的文件夹。
1 | zhangzhen@ubuntu:~/software$ bin/hadoop jar hadoop-examples-1.2.1.jar wordcount in put |
查看输出的结果:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 | zhangzhen@ubuntu:~/software/hadoop-1.2.1$ bin/hadoop dfs -ls Found 2 items drwxr-xr-x - zhangzhen supergroup 0 2014-03-22 10:45 / user /zhangzhen/ in drwxr-xr-x - zhangzhen supergroup 0 2014-03-22 10:56 / user /zhangzhen/put zhangzhen@ubuntu:~/software/hadoop-1.2.1$ bin/hadoop dfs -ls ./put Found 3 items -rw-r --r-- 1 zhangzhen supergroup 0 2014-03-22 10:56 /user/zhangzhen/put/_SUCCESS drwxr-xr-x - zhangzhen supergroup 0 2014-03-22 10:56 / user /zhangzhen/put/_logs 目录 -rw-r --r-- 1 zhangzhen supergroup 39 2014-03-22 10:56 /user/zhangzhen/put/part-r-00000 这是文件 zhangzhen@ubuntu:~/software/hadoop-1.2.1/hadoop dfs -cat ./put/* I 1 You 1 am 1 are 1 not 1 zhangzhen 2 cat: File does not exist: / user /zhangzhen/put/_logs zhangzhen@ubuntu:~/software/hadoop-1.2.1$ |
上面的结果,就基本可以证明Hadoop搭建是没有问题的。执行hadoop-examples-1.2.1.jar程序,其实是把java程序编译打成一个jar文件,然后直接运行,就可以得到结果。其实这也是以后我们运行java程序的一个方法。把程序编译打包上传,然后运行。还有另一种方面,eclipse连接Hadoop,可以联机测试。两种方法各有优点,不再详述。
运行的程序,我们可以在Hadoop的安装目录中找到源文件,WourdCount.java源代码。
1 2 3 | zhangzhen@ubuntu:~/software/hadoop-1.2.1/src/examples/org/apache/hadoop/examples$ pwd /home/zhangzhen/software/hadoop-1.2.1/src/examples/org/apache/hadoop/examples zhangzhen@ubuntu:~/software/hadoop-1.2.1/src/examples/org/apache/hadoop/examples$ |
下面是把源代码拷到eclipse程序中,利用此代码(并未修改)测试一下实际的数据并得到结果。(注释是对上以一行的解释)
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 | 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 TokenizerMapper extends Mapper<Object, Text, Text, IntWritable>{ //规定map中用到的数据类型,这里的Text相当于jdk中的String IntWritable相当于jdk的 int 类型, //这样做的原因主要是为了hadoop的数据序化而做的。 private final static IntWritable one = new IntWritable(1); //声时一个IntWritable变量,作计数用,每出现一个 key ,给其一个value=http://www.mamicode.com/1的值 private Text word = new Text();//用来暂存map输出中的 key 值,Text类型的 public void map(Object key , Text value, Context context ) throws IOException, InterruptedException { //这就是map函数,它是和Mapper抽象类中的相对应的,此处的Object key ,Text value的类型和上边的Object, //Text是相对应的,而且最好一样,不然的话,多数情况运行时会报错。 StringTokenizer itr = new StringTokenizer(value.toString()); //Hadoop读入的value是以行为单位的,其 key 为该行所对应的行号,因为我们要计算每个单词的数目, //默认以空格作为间隔,故用StringTokenizer辅助做字符串的拆分,也可以用string.split( "" )来作。 while (itr.hasMoreTokens()) { //遍历一下每行字符串中的单词 word. set (itr.nextToken()); //出现一个单词就给它设成一个 key 并将其值设为1 context.write(word, one); //输出设成的 key /value值 //上面就是map打散的过程 } } } public static class IntSumReducer extends Reducer<Text,IntWritable,Text,IntWritable> { //reduce的静态类,这里和Map中的作用是一样的,设定输入/输出的值的类型 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 ) { //由于map的打散,这里会得到如,{ key , values }={ "hello" ,{1,1,....}},这样的集合 sum += val.get(); //这里需要逐一将它们的value取出来予以相加,取得总的出现次数,即为汇和 } result. set ( sum ); //将 values 的和取得,并设成result对应的值 context.write( key , result); //此时的 key 即为map打散之后输出的 key ,没有变化,变化的时result,以前得到的是一个数字的集合, //已经给算出和了,并做为 key /value输出。 } } 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" ); //此程序的执行,在hadoop看来是一个Job,故进行初始化job操作 job.setJarByClass(Wordcount.class); //可以认为成,此程序要执行MyWordCount.class这个字节码文件 job.setMapperClass(TokenizerMapper.class); //在这个job中,我用TokenizerMapper这个类的map函数 job.setCombinerClass(IntSumReducer.class); job.setReducerClass(IntSumReducer.class); //在这个job中,我用IntSumReducer这个类的reduce函数 job.setOutputKeyClass(Text.class); //在reduce的输出时, key 的输出类型为Text job.setOutputValueClass(IntWritable.class); //在reduce的输出时,value的输出类型为IntWritable FileInputFormat.addInputPath(job, new Path(otherArgs[0])); //初始化要计算word的文件的路径 FileOutputFormat.setOutputPath(job, new Path(otherArgs[1])); //初始化要计算word的文件的之后的结果的输出路径 System.exit(job.waitForCompletion( true ) ? 0 : 1); //提交job到hadoop上去执行了,意思是指如果这个job真正的执行完了则主函数退出了,若没有真正的执行完就退出了。 } //参考:http://hi.baidu.com/erliang20088/item/ce550f2f088ff1ce0e37f930 } |
WourdCount程序中隐藏的秘密
1、具体流程:
1)文件拆分成splits,由于测试用的文件较小,所以每个文件为一个split,并将文件按行分割形成<key,value> 对,如下图。这一步由MapReduce框架自动完成,其中偏移量(即key值)包括了回车所占的字符数和Linux环境有关。
2)将分割好的<key,value>对交给用户定义的map方法进行处理,生成新的<key,value>对。
3)得到map方法输出的<key,value>对后,Mapper会将它们按照key值进行排序,并执行Combine过程,将key至相同value值累加,得到Mapper的最终输出结果。
2、Map Task的整体流程:
可以概括为5个步骤:
1)Read:Map Task通过用户编写的RecordReader,从输入InputSplit中解析出一个个key/value。
2)Map:该阶段主要将解析出的key/value交给用户编写的map()函数处理,并产生一系列的key/value。
3)Collect:在用户编写的map()函数中,当数据处理完成后,一般会调用OutputCollector.collect()输入结果。在该函数内部,它会将生成的key/value分片(通过Partitioner),并写入一个环形内存缓冲区中。
4)Spill:即“溢写”,当环形缓冲区满后,MapReduce会将数据写到本地磁盘上,生成一个临时文件。将数据写入本地磁盘之前,先要对数据进行一次本地排序,并在必要时对数据进行合并,压缩等操作。
5)Combine:当所有数据处理完成后,Map Task对所有临时文件进行一次合并,以确保最终只会生成一个数据文件。
3、Reduce的整体流程:
可以概括为5个步骤:
1)Shuffle:也称Copy阶段。Reduce Task从各个Map Task上远程拷贝一片数据,并针对某一片数据,如果其大小超过一定阀值,则写到磁盘上,否则直接放到内存中。
2)Merge:在远程拷贝的同时,Reduce Task启动了两个后台线程对内存和磁盘上的文件进行合并,以防止内存使用过多或者磁盘上文件过多。
3)Sort:按照MapReduce语义,用户编写的reduce()函数输入数据是按key进行聚集的一组数据。为了将key相同的数据聚在一 起,Hadoop采用了基于排序的策略。由于各个Map Task已经实现了对自己的处理结果进行了局部排序,因此,Reduce Task只需对所有数据进行一次归并排序即可。
4)Reduce:在该阶段中,Reduce Task将每组数据依次交给用户编写的reduce()函数处理。
5)Write:reduce()函数将计算结果写到HDFS。
通过一些博客对WourdCount的介绍示例,总结Map、Reduce的整个过程。加上Thinking in BigDate(八)大数据Hadoop核心架构HDFS+MapReduce+Hbase+Hive内部机理详解所将的内容,大致把整个文件数据处理的过程梳理一遍。但是还有很多细节没有讲明。如:Spill、Combine、Shuffle的过程,Shuffle整个MapReduce的核心。接下来,我们更深入了解MapReduce的过程,更深入的了解,便于我们在以后在操作Hadoop集群的过程中,有利于系统调优,甚至修改Hadoop源代码。
WordCount示例深度学习MapReduce过程(1)