Hadoop MapReduce 概念和基础

secce

Hadoop MapReduce 逻辑代码
public class NewMaxTemperature {
public static final int CONST_INT_ZERO = 0;
public static final int CONST_INT_ONE = 1;
//【关键代码】Mapper类继承 Mapper类
static class NewMaxTemperatureMapper extends Mapper{
//【关键代码】覆盖父类Mapper的map(LongWritable, Text, Context)方法
@Override
protected void map(LongWritable key, Text value, Context context ) throws IOException , InterruptedException{
//[业务代码]
super.map(key, value, context);
String line = value.toString();
String year = line.substring(15, 19);
int airTemperature;
if( line.charAt(87) == '+'){
airTemperature = Integer.parseInt(line.substring(88, 92));
}else{
airTemperature =Integer.parseInt( line.substring( 87, 92));
}
String quality = line.substring(92, 93);
if(airTemperature != MISSING && quality.matches( "[01459]" )){
context.write( new Text( year), new IntWritable( airTemperature));
}
//[/业务代码]      
}
}
//【关键代码】继承Reducer
static class NewMaxTemperatureReducer extends Reducer{
//【关键代码】覆盖Reducer的reduce(Text , Iterable, Context)方法
@Override
protected void reduce (Text key, Iterable values, Context context) throws IOException , InterruptedException{
//[业务代码]
for (IntWritable value : values){
maxValue = Math.max(maxValue , value.get() );
}
context.write(key,  new IntWritable(maxValue));
//[/业务代码]
}
}
public static void main(String[] args) throws IOException, InterruptedException, ClassNotFoundException{
//【关键代码】创建一个任务，设置任务入口类
Job job= new Job();
job.setJarByClass(NewMaxTemperature.class);
FileInputFormat.addInputPath(job, new Path(args[0]));
FileInputFormat.addInputPath(job, new Path(args[0]));
//【关键代码】设置任务的map任务，和reduce任务，及入口类型
job.setMapperClass(NewMaxTemperatureMapper.class);
job.setReducerClass(NewMaxTemperatureReducer.class);
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(IntWritable.class);
//【关键代码】 等待完成
System.exit(job.waitForCompletion(true) ? CONST_INT_ZERO : CONST_INT_ONE);
}
}



MapReduce作业时客户端执行的单位：包括输入数据、MapReduce、程序和配置信息。 Hadoop把作业分为若干个小任务task， 任务包括map任务和reduce任务。
有两种节点控制作业的完成， 这种节点包括jobtracker和多个tasktracker。jobtracker是tasktracker的调度器， tasktracker是jobtracker的slaves。tasktracker在运行过程中从jobtracker获取任务，并把进度报告给jobtracker， 如果一个任务失败，jobtracker会重新调度任务到另外一个tasktracker。 Hadoop 把输入的数据划分为小数据到MapReduce，称为输入分片（input split），每个分片包括创建map任务， 由它运行自定义的map函数来分析片中的记录。分片大小一般为一个block的大小64M。

map任务把输出到本地硬盘， 而不是HDFS，应为map的输出是一个中间输出，中间输出是reduce处理的输入， 最终结果是reduce输出的。一但结果完成，map的输出是可以删除的，所以map的结果一般是不会存储在HDFS中的。

reduce任务不具备本地读取的优势-- 一个单一的reduce任务的输入往往来自于所有map的输出。 map的输出通过网络传输到reduce任务运行的节点， 并在哪里合并， 然后传递给自定义的reduce函数， 为了增加可靠性， reduce的输出通常存储在HDFS中。

如果是多个reducer， map任务会会对输出结果进行分区， 为配给reduce任务创建一个分区（partition）。 每个分区包含许多键/值对， 分区可以有用户自定义的partitioner来控制，通常是采用默认的分区工具， 它使用的算法是hashhas函数， 这种方法效率非常高。
也有可能不存在reduce任务， 直接的输出就是map任务的输出


注释： 大部分内容来自《Hadoop权威指南》