MapReduce编程模型介绍
MapReduce采用"分而治之"的思想,把大规模数据集的操作分发(map)给一个主节点(JobTracker)管理下的各个分节点(TaskTracker)共同完成,然后通过整合(reduce)各个节点的中间结果,得到最终结果.简单地说,MapReduce就是"任务的分解与结果的汇总"。
在Hadoop中,用于执行MapRecude任务的机器角色有两个: 一个是执行JobTracker任务的机器(只有一个);另一个是执行TaskTracker任务的机器(有多个). JobTracker负责调度控制管理TackTracker的任务执行,失败重试等.
在hadoop2.x的版本中,处理任务的节点已经不再叫JobTraker和TaskTracker ,而是有YARN来管理.
MapRecude处理的数据集或任务必须具备这样的特点:
待处理的数据集可以分解成许多小的数据集,并且每一个小的数据集都可以完全并行的进行处理而相互之间不会产生影响
WordCount程序源码
WordCountMapper类
package com.shell.hadoop.wordcount;
import java.io.IOException;
import java.util.StringTokenizer;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
public class WordCountMapper extends Mapper<LongWritable, Text, Text, IntWritable>{
private final static IntWritable one = new IntWritable(1);
private Text word = new Text();
@Override
protected void map(LongWritable key, Text value, Context context)
throws IOException, InterruptedException {
String line = value.toString();
StringTokenizer stringTokenizer = new StringTokenizer(line, " ");
while(stringTokenizer.hasMoreElements()) {
word.set(stringTokenizer.nextToken());
context.write(word, one);
}
}
}
WordCountReducer类
package com.shell.hadoop.wordcount;
import java.io.IOException;
import java.util.Iterator;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;
public class WordCountReducer extends Reducer<Text, IntWritable, Text, IntWritable>{
@Override
protected void reduce(Text key, Iterable<IntWritable> values,
Reducer<Text, IntWritable, Text, IntWritable>.Context context)
throws IOException, InterruptedException {
int sum = 0;
Iterator<IntWritable> iterator = values.iterator();
while(iterator.hasNext()) {
sum = sum + iterator.next().get();
}
context.write(key, new IntWritable(sum));
}
}
主程序WordCount
package com.shell.hadoop.wordcount;
import java.io.IOException;
import org.apache.hadoop.conf.Configured;
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.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;
/**
* @author shell
*/
public class WordCount extends Configured implements Tool {
public static void main(String[] args) throws Exception {
int exitCode = ToolRunner.run(new WordCount(), args);
System.exit(exitCode);
}
@Override
public int run(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
if (args.length != 2) {
System.err.printf("Usage: %s [generic options] <input> <output>\n", getClass().getSimpleName());
ToolRunner.printGenericCommandUsage(System.err);
return -1;
}
/**
* 作业配置Job Configuration
*/
Job job = Job.getInstance();
job.setJarByClass(WordCount.class);
job.setJobName("WordCount");
FileInputFormat.addInputPath(job, new Path(args[0])); // 指定输入路径,是hdfs文件系统中的路径
FileOutputFormat.setOutputPath(job, new Path(args[1])); // 指定结果输出路径, hdfs文件系统中的路径
job.setOutputKeyClass(Text.class); // 指定输出map中的key类型
job.setOutputValueClass(IntWritable.class); // 指定输出map中的value类型
job.setOutputFormatClass(TextOutputFormat.class); // 指定最后输出结果的存储类型
job.setMapperClass(WordCountMapper.class); // 指定执行map操作的类
job.setReducerClass(WordCountReducer.class); // 指定执行reduce操作的类
int retureValue = job.waitForCompletion(true) ? 0 : 1; // 提交任务
System.out.println("job.isSuccessful " + job.isSuccessful());
return retureValue;
}
}
程序作用: WordCount程序是根据用户的输入统计每个输入中各个单词出现次数,并输出结果。
运行：
下面主要讲解WordCount中MapReduce执行过程分析
在WordCount的main方法中调用ToolRunner.run(Tool, args)方法,这个方法会主动调用Tool接口的run(args)方法,在这里也就是WordCount中的run()方法。
在WordCount的run()方法中,首先,我们通过Job.getInstance()实例化了一个工作job,并通过setXXX()方法给这个job配置了一些东西; 最后通过job.waitForCompletion()(会根据job的状态来确定是否要提交任务执行)来提交job。
job根据输入路径,从input目录下获取到wordcount.txt和wordcount2.txt两个文件作为输入内容,然后对这两个文件进行split操作将两个文件根据指定的分隔符(这里是换行)分成若干个InputSplit块(每个InputSplit块会记录分块针对文件的起始位置和分块长度,并不记录实际的内容,InputSplit的获取通过job设置的InputFormat的getSplits方法获得),然后传递给Mapper去进行map操作,map将输入的InputSplit分块传递给job设置的InputFormat类(这里是FileInputFormat)的createRecordReader()方法来生成特定的RecordReader对象,由RecordReader调用getCurrentKey()和getCurrentValue()来生成可供map处理的<key, value>。也就是说job设置的InputFormat类决定了map处理的<key, value>键值对的生成方式。
如果hadoop提供的InputFormat类不能满足我们的应用场景，那么我们可以实现自己的InputFormat，怎么实现呢？通过继承InputFormat类和InputSplit类重写其中的特定方法来实现自己的InputSplit分块，而如果hadoop提供的RecordReader类不能满足我们的需求，我们也可以实现自己的RecordReader类，实现getCurrentKey()和getCurrentValue()方法即可。
