收藏本站

快速投稿

联系我们

广告服务

基石导航

峰会活动

基石数据

社区

开启辅助访问

搜索
大数据中国 › 首页 › 大数据技术 › 查看内容
Hadoop集群系列7：WordCount运行详解
2014-4-19 17:13 |来自: 博客园| 查看: 4104| 评论: 0


3、WordCount源码分析

3.1 特别数据类型介绍

Hadoop提供了如下内容的数据类型，这些数据类型都实现了WritableComparable接口，以便用这些类型定义的数据可以被序列化进行网络传输和文件存储，以及进行大小比较。

BooleanWritable：标准布尔型数值

ByteWritable：单字节数值

DoubleWritable：双字节数

FloatWritable：浮点数

IntWritable：整型数

LongWritable：长整型数

Text：使用UTF8格式存储的文本

NullWritable：当<key,value>中的key或value为空时使用

3.2 旧的WordCount分析

1）源代码程序

  1. package org.apache.hadoop.examples; 
  2.  
  3. import java.io.IOException; 
  4. import java.util.Iterator; 
  5. import java.util.StringTokenizer; 
  6.  
  7. import org.apache.hadoop.fs.Path; 
  8. import org.apache.hadoop.io.IntWritable; 
  9. import org.apache.hadoop.io.LongWritable; 
  10. import org.apache.hadoop.io.Text; 
  11. import org.apache.hadoop.mapred.FileInputFormat; 
  12. import org.apache.hadoop.mapred.FileOutputFormat; 
  13. import org.apache.hadoop.mapred.JobClient; 
  14. import org.apache.hadoop.mapred.JobConf; 
  15. import org.apache.hadoop.mapred.MapReduceBase; 
  16. import org.apache.hadoop.mapred.Mapper; 
  17. import org.apache.hadoop.mapred.OutputCollector; 
  18. import org.apache.hadoop.mapred.Reducer; 
  19. import org.apache.hadoop.mapred.Reporter; 
  20. import org.apache.hadoop.mapred.TextInputFormat; 
  21. import org.apache.hadoop.mapred.TextOutputFormat; 
  22.  
  23. public class WordCount { 
  24.  
  25.     public static class Map extends MapReduceBase implements 
  26.             Mapper<LongWritable, Text, Text, IntWritable> { 
  27.         private final static IntWritable one = new IntWritable(1); 
  28.         private Text word = new Text(); 
  29.  
  30.         public void map(LongWritable key, Text value, 
  31.                 OutputCollector<Text, IntWritable> output, Reporter reporter) 
  32.                 throws IOException { 
  33.             String line = value.toString(); 
  34.             StringTokenizer tokenizer = new StringTokenizer(line); 
  35.             while (tokenizer.hasMoreTokens()) { 
  36.                 word.set(tokenizer.nextToken()); 
  37.                 output.collect(word, one); 
  38.             } 
  39.         } 
  40.     } 
  41.  
  42.     public static class Reduce extends MapReduceBase implements 
  43.             Reducer<Text, IntWritable, Text, IntWritable> { 
  44.         public void reduce(Text key, Iterator<IntWritable> values, 
  45.                 OutputCollector<Text, IntWritable> output, Reporter reporter) 
  46.                 throws IOException { 
  47.             int sum = 0; 
  48.             while (values.hasNext()) { 
  49.                 sum += values.next().get(); 
  50.             } 
  51.             output.collect(key, new IntWritable(sum)); 
  52.         } 
  53.     } 
  54.  
  55.     public static void main(String[] args) throws Exception { 
  56.         JobConf conf = new JobConf(WordCount.class); 
  57.         conf.setJobName("wordcount"); 
  58.  
  59.         conf.setOutputKeyClass(Text.class); 
  60.         conf.setOutputValueClass(IntWritable.class); 
  61.  
  62.         conf.setMapperClass(Map.class); 
  63.         conf.setCombinerClass(Reduce.class); 
  64.         conf.setReducerClass(Reduce.class); 
  65.  
  66.         conf.setInputFormat(TextInputFormat.class); 
  67.         conf.setOutputFormat(TextOutputFormat.class); 
  68.  
  69.         FileInputFormat.setInputPaths(conf, new Path(args[0])); 
  70.         FileOutputFormat.setOutputPath(conf, new Path(args[1])); 
  71.  
  72.         JobClient.runJob(conf); 
  73.     } 
  74. } 

3）主方法Main分析

  1. public static void main(String[] args) throws Exception { 
  2.     JobConf conf = new JobConf(WordCount.class); 
  3.     conf.setJobName("wordcount"); 
  4.  
  5.     conf.setOutputKeyClass(Text.class); 
  6.     conf.setOutputValueClass(IntWritable.class); 
  7.  
  8.     conf.setMapperClass(Map.class); 
  9.     conf.setCombinerClass(Reduce.class); 
  10.     conf.setReducerClass(Reduce.class); 
  11.  
  12.     conf.setInputFormat(TextInputFormat.class); 
  13.     conf.setOutputFormat(TextOutputFormat.class); 
  14.  
  15.     FileInputFormat.setInputPaths(conf, new Path(args[0])); 
  16.     FileOutputFormat.setOutputPath(conf, new Path(args[1])); 
  17.  
  18.     JobClient.runJob(conf); 
  19. } 

首先讲解一下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（拆分）、Combiner（中间结果合并）以及Reduce（合并）的相关处理类。这里用Reduce类来进行Map产生的中间结果合并，避免给网络数据传输产生压力。

conf.setMapperClass(Map.class );

conf.setCombinerClass(Reduce.class );

conf.setReducerClass(Reduce.class );

接着就是调用setInputPath()和setOutputPath()设置输入输出路径。

conf.setInputFormat(TextInputFormat.class );

conf.setOutputFormat(TextOutputFormat.class );

（1）InputFormat和InputSplit

InputSplit是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，分别是：

  1. InputFormat 
  2.  
  3.     | 
  4.  
  5.     |---BaileyBorweinPlouffe.BbpInputFormat 
  6.  
  7.     |---ComposableInputFormat 
  8.  
  9.     |---CompositeInputFormat 
  10.  
  11.     |---DBInputFormat 
  12.  
  13.     |---DistSum.Machine.AbstractInputFormat 
  14.  
  15.     |---FileInputFormat 
  16.  
  17.         |---CombineFileInputFormat 
  18.  
  19.         |---KeyValueTextInputFormat 
  20.  
  21.         |---NLineInputFormat 
  22.  
  23.         |---SequenceFileInputFormat 
  24.  
  25.         |---TeraInputFormat 
  26.  
  27.         |---TextInputFormat 

其中TextInputFormat是Hadoop默认的输入方法，在TextInputFormat中，每个文件（或其一部分）都会单独地作为map的输入，而这个是继承自FileInputFormat的。之后，每行数据都会生成一条记录，每条记录则表示成<key,value>形式：

  • key值是每个数据的记录在数据分片中字节偏移量，数据类型是LongWritable；

value值是每行的内容，数据类型是Text。

（2）OutputFormat

每一种输入格式都有一种输出格式与其对应。默认的输出格式是TextOutputFormat，这种输出方式与输入类似，会将每条记录以一行的形式存入文本文件。不过，它的键和值可以是任意形式的，因为程序内容会调用toString()方法将键和值转换为String类型再输出。

3）Map类中map方法分析

  1. public static class Map extends MapReduceBase implements 
  2.         Mapper<LongWritable, Text, Text, IntWritable> { 
  3.     private final static IntWritable one = new IntWritable(1); 
  4.     private Text word = new Text(); 
  5.  
  6.     public void map(LongWritable key, Text value, 
  7.             OutputCollector<Text, IntWritable> output, Reporter reporter) 
  8.             throws IOException { 
  9.         String line = value.toString(); 
  10.         StringTokenizer tokenizer = new StringTokenizer(line); 
  11.         while (tokenizer.hasMoreTokens()) { 
  12.             word.set(tokenizer.nextToken()); 
  13.             output.collect(word, one); 
  14.         } 
  15.     } 
  16. } 

Map类继承自MapReduceBase，并且它实现了Mapper接口，此接口是一个规范类型，它有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>。

4）Reduce类中reduce方法分析

  1. public static class Reduce extends MapReduceBase implements 
  2.         Reducer<Text, IntWritable, Text, IntWritable> { 
  3.     public void reduce(Text key, Iterator<IntWritable> values, 
  4.             OutputCollector<Text, IntWritable> output, Reporter reporter) 
  5.             throws IOException { 
  6.         int sum = 0; 
  7.         while (values.hasNext()) { 
  8.             sum += values.next().get(); 
  9.         } 
  10.         output.collect(key, new IntWritable(sum)); 
  11.     } 
  12. } 

Reduce类也是继承自MapReduceBase的，需要实现Reducer接口。Reduce类以map的输出作为输入，因此Reduce的输入类型是<Text，Intwritable>。而Reduce的输出是单词和它的数目，因此，它的输出类型是<Text,IntWritable>。Reduce类也要实现reduce方法，在此方法中，reduce函数将输入的key值作为输出的key值，然后将获得多个value值加起来，作为输出的值。

3.3 新的WordCount分析

1）源代码程序

  1. package org.apache.hadoop.examples; 
  2. import java.io.IOException; 
  3. import java.util.StringTokenizer; 
  4. import org.apache.hadoop.conf.Configuration; 
  5. import org.apache.hadoop.fs.Path; 
  6. import org.apache.hadoop.io.IntWritable; 
  7. import org.apache.hadoop.io.Text; 
  8. import org.apache.hadoop.mapreduce.Job; 
  9. import org.apache.hadoop.mapreduce.Mapper; 
  10. import org.apache.hadoop.mapreduce.Reducer; 
  11. import org.apache.hadoop.mapreduce.lib.input.FileInputFormat; 
  12. import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat; 
  13. import org.apache.hadoop.util.GenericOptionsParser; 
  14. public class WordCount { 
  15. public static class TokenizerMapper 
  16. extends Mapper<Object, Text, Text, IntWritable>{ 
  17. private final static IntWritable one = new IntWritable(1); 
  18. private Text word = new Text(); 
  19. public void map(Object key, Text value, Context context) 
  20. throws IOException, InterruptedException { 
  21. StringTokenizer itr = new StringTokenizer(value.toString()); 
  22. while (itr.hasMoreTokens()) { 
  23. word.set(itr.nextToken()); 
  24. context.write(word, one); 
  25. } 
  26. } 
  27. } 
  28. public static class IntSumReducer 
  29. extends Reducer<Text,IntWritable,Text,IntWritable> { 
  30. private IntWritable result = new IntWritable(); 
  31. public void reduce(Text key, Iterable<IntWritable> values,Context context) 
  32.  throws IOException, InterruptedException { 
  33. int sum = 0; 
  34. for (IntWritable val : values) { 
  35. 　sum += val.get(); 
  36. } 
  37. result.set(sum); 
  38. context.write(key, result); 
  39. } 
  40. } 
  41. public static void main(String[] args) throws Exception { 
  42. Configuration conf = new Configuration(); 
  43. String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs(); 
  44. if (otherArgs.length != 2) { 
  45. System.err.println("Usage: wordcount <in> <out>"); 
  46. System.exit(2); 
  47. } 
  48. Job job = new Job(conf, "word count"); 
  49. job.setJarByClass(WordCount.class); 
  50. job.setMapperClass(TokenizerMapper.class); 
  51. job.setCombinerClass(IntSumReducer.class); 
  52. job.setReducerClass(IntSumReducer.class); 
  53. job.setOutputKeyClass(Text.class); 
  54. job.setOutputValueClass(IntWritable.class); 
  55. FileInputFormat.addInputPath(job, new Path(otherArgs[0])); 
  56. FileOutputFormat.setOutputPath(job, new Path(otherArgs[1])); 
  57. System.exit(job.waitForCompletion(true) ? 0 : 1); 
  58. } 
  59. } 

　 1）Map过程

public static class TokenizerMapper

extends 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);

}

}

Map过程需要继承org.apache.hadoop.mapreduce包中Mapper类，并重写其map方法。通过在map方法中添加两句把key值和value值输出到控制台的代码，可以发现map方法中value值存储的是文本文件中的一行（以回车符为行结束标记），而key值为该行的首字母相对于文本文件的首地址的偏移量。然后StringTokenizer类将每一行拆分成为一个个的单词，并将<word,1>作为map方法的结果输出，其余的工作都交有MapReduce框架处理。

2）Reduce过程

public static class IntSumReducer

extends 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任务

  1. public static void main(String[] args) throws Exception { 
  2. Configuration conf = new Configuration(); 
  3. String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs(); 
  4. if (otherArgs.length != 2) { 
  5. System.err.println("Usage: wordcount <in> <out>"); 
  6. System.exit(2); 
  7. } 
  8. Job job = new Job(conf, "word count"); 
  9. job.setJarByClass(WordCount.class); 
  10. job.setMapperClass(TokenizerMapper.class); 
  11. job.setCombinerClass(IntSumReducer.class); 
  12. job.setReducerClass(IntSumReducer.class); 
  13. job.setOutputKeyClass(Text.class); 
  14. job.setOutputValueClass(IntWritable.class); 
  15. FileInputFormat.addInputPath(job, new Path(otherArgs[0])); 
  16. FileOutputFormat.setOutputPath(job, new Path(otherArgs[1])); 
  17. System.exit(job.waitForCompletion(true) ? 0 : 1); 
  18. } 

在MapReduce中，由Job对象负责管理和运行一个计算任务，并通过Job的一些方法对任务的参数进行相关的设置。此处设置了使用TokenizerMapper完成Map过程中的处理和使用IntSumReducer完成Combine和Reduce过程中的处理。还设置了Map过程和Reduce过程的输出类型：key的类型为Text，value的类型为IntWritable。任务的输出和输入路径则由命令行参数指定，并由FileInputFormat和FileOutputFormat分别设定。完成相应任务的参数设定后，即可调用job.waitForCompletion()方法执行任务。

4、WordCount处理过程

本节将对WordCount进行更详细的讲解。详细执行步骤如下：

1）将文件拆分成splits，由于测试用的文件较小，所以每个文件为一个split，并将文件按行分割形成<key,value>对，如图4-1所示。这一步由MapReduce框架自动完成，其中偏移量（即key值）包括了回车所占的字符数（Windows和Linux环境会不同）。

image

图4-1 分割过程

2）将分割好的<key,value>对交给用户定义的map方法进行处理，生成新的<key,value>对，如图4-2所示。

image

图4-2 执行map方法

3）得到map方法输出的<key,value>对后，Mapper会将它们按照key值进行排序，并执行Combine过程，将key至相同value值累加，得到Mapper的最终输出结果。如图4-3所示。

image

图4-3 Map端排序及Combine过程

4）Reducer先对从Mapper接收的数据进行排序，再交由用户自定义的reduce方法进行处理，得到新的<key,value>对，并作为WordCount的输出结果，如图4-4所示。

image

图4-4 Reduce端排序及输出结果

5、MapReduce新旧改变

Hadoop最新版本的MapReduce Release 0.20.0的API包括了一个全新的Mapreduce JAVA API，有时候也称为上下文对象。

新的API类型上不兼容以前的API，所以，以前的应用程序需要重写才能使新的API发挥其作用 。

新的API和旧的API之间有下面几个明显的区别。

新的API倾向于使用抽象类，而不是接口，因为这更容易扩展。例如，你可以添加一个方法(用默认的实现)到一个抽象类而不需修改类之前的实现方法。在新的API中，Mapper和Reducer是抽象类。

新的API是在org.apache.hadoop.mapreduce包(和子包)中的。之前版本的API则是放在org.apache.hadoop.mapred中的。

新的API广泛使用context object(上下文对象)，并允许用户代码与MapReduce系统进行通信。例如，MapContext基本上充当着JobConf的OutputCollector和Reporter的角色。

新的API同时支持"推"和"拉"式的迭代。在这两个新老API中，键/值记录对被推mapper中，但除此之外，新的API允许把记录从map()方法中拉出，这也适用于reducer。"拉"式的一个有用的例子是分批处理记录，而不是一个接一个。

新的API统一了配置。旧的API有一个特殊的JobConf对象用于作业配置，这是一个对于Hadoop通常的Configuration对象的扩展。在新的API中，这种区别没有了，所以作业配置通过Configuration来完成。作业控制的执行由Job类来负责，而不是JobClient，它在新的API中已经荡然无存。

免责声明： 除非特别声明，文章均为投稿或网络转载，仅代表作者观点，与大数据中国网无关。其原创性以及文中陈述文字和内容未经本站证实，对本文以及其中全部或者部分内容、文字的真实性、完整性、及时性本站不作任何保证或承诺，请读者仅作参考，并请自行核实相关内容。如果本文内容有侵犯你的权益，请发送信息至ab12-120@163.com，我们会及时删除
12
收藏 分享 邀请
上一篇：Hadoop集群6：JDK和SSH无密码配置 下一篇：Hadoop集群系列8：Eclipse开发环境设置

最新评论

最新新闻

热门新闻

热门图片

关闭

站长推荐上一条 /1 下一条

大数据中国微信

QQ   

版权所有： Discuz! © 2001-2013 大数据.

GMT+8, 2024-11-25 05:54 , Processed in 0.084130 second(s), 23 queries .

返回顶部