Hadoop(四)——编程核心MapReduce（上）

上篇讲述了Hadoop的核心内容之一HDFS，是Hadoop分布式的平台基础，而这讲的MapReduce则是充分利用Hdfs分布式，提高运行效率的算法模型 ，Map(映射)和Reduce(归约)两个主要阶段都以<key,value>键值对作为输入和输出，我们需要做的就是对这些<key，value>做我们想要的处理。看似简单实则麻烦，因为这里太灵活多变。 一，好，首先来看下边两个图，看下mapreduce在Hadoop中的执行流程，以及mapreduce内部的执行流程：


以分析气象数据为例子：



二，解析：mapreduce的执行步骤：
map任务处理: 1.读取输入文件内容,解析成键值对(key/value).对输入文件的每一行,解析成 键值对(key/value).每一个键值对调用一次map函数 2.写自己的逻辑,对输入的键值对(key/value)处理,转换成新的键值对 (key/value)输出. 3.对输出的键值对(key/value)进行分区.(partition) 4.对不同分区的数据,按照key进行排序,分组.相同的key/value放到 一个集合中.(shuffle) 5.分组后的数据进行规约.(combiner,可选择的)，也就是可以在mapper中处理一部 分reduce的工作，将reduce的工作进行减压
reduce任务处理: 1.对多个map任务的输出,按照不同的分区,通过网络copy到不同的reduce节点. 2.对多个map任务的输出进行合并,排序.写reduce函数自己的逻辑,对输入的 key/value处理,转换成新的key/value输出. 3.把reduce的输出保存到文件中(写入到hdfs中). 三，任务执行优化：
1，推测式执行：即如果jobtracker发现有拖后腿的任务，会再启动一个相同的备份任务，然后那个先执行完就会kill掉另一个。因此在监控网页上经常能看到正常执行完的作业有被kill的任务。 2，推测式执行缺省打开，但如果是代码问题，并不能解决问题，而且会使集群更慢，通过在mapred-site.xml配置文件中设置mapred.map.tasks.speculative.execution和mapred.reduce.tasks.speculative.execution可为map任务或reduce任务开启或关闭推测式执行 3，重用JVM，可以省去启动新的JVM消耗的时间，在mapred-site.xml配置文件中设置mapred.job.reuse.jvm.num.tasks设置单个JVM上运行的最大任务数（ 1， >1或-1表示没有限制） 4，忽略模式， 任务在读取数据失败2次后， 会把数据位置告诉jobtracker， 后者重新启动该任务并且在遇到所记录的坏数据时直接跳过（ 缺省关闭， 用SkipBadRecord方法打开）
四，错误机制处理故障：
1，硬件故障，即jobtracker和tasktracker故障：A,Jobtracker是单点， 若发生故障目前hadoop还无法处理， 唯有选择最牢靠的硬件作为jobtrackerB,Jobtracker通过心跳（ 周期1分钟） 信号了解tasktracker是否发生故障或负载过于严重C,Jobtracker将从任务节点列表中移除发生故障的tasktracker D,如果故障节点在执行map任务并且尚未完成， jobtracker会要求其它节点重新执行此map任务F,如果故障节点在执行reduce任务并且尚未完成，jobtracker会要求其它节点继续执行尚未完成的reduce任务
2,任务失败：由于代码或者进程崩溃引起任务失败： A,Jvm自动退出，向tasktracker父进程发送方错误信息，错误信息也会写入到日志 B,Tasktracker监听程序会发现进程退出，或进程很久没有更新信息送回，将任务标记为失败 C,标记失败任务后，任务计数器减去1以便接受新任务，并通过心跳信号告诉jobtracker任务失败的信息 D,Jobtrack获悉任务失败后，将把该任务重新放入调度队列，重新分配出去再执行 E,如果一个任务失败超过4次（可以设置），将不会再被执行，同时作业也宣布失
五，最后来看一个wordCount的例子：

package job;

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;

/**
 * hadoop的第一个mapreduce例子，wordCount,计算单词的个数
 * 
 * @author Administrator
 * 
 */
public class WordCount {
	
	/*
	 *  继承mapper接口，设置map的输入类型为<Object,Text>,输出类型为<Text,IntWritable>
	 */
	public static class TokenizerMapper extends Mapper<Object, Text, Text, IntWritable> {
		//one表示单词出现了一次
		private final static IntWritable one=new IntWritable(1);
		//word用来存储切下来的单词
		private Text word=new Text();
		
		//map进行将内容分割，以<单词，1>的形式write出来
		public void map(Object key, Text value,Context context) throws IOException,InterruptedException{
			//进行单词的切分
			StringTokenizer itr=new StringTokenizer(value.toString());
			
			while (itr.hasMoreElements()) {
				word.set(itr.nextToken());//切下的单词放到word中
				context.write(word, one);
			}
		}
	}
	
	/**
	 * reducer函数的编写
	 * @author Administrator
	 *
	 */
	public static class IntSumReducer extends Reducer<Text, IntWritable, Text, IntWritable>{
		
		//result记录单词的频数
		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);
		}
	}
	
	
	public static void main(String[] args) throws Exception{
		Configuration configuration=new Configuration();
		String[] otherArgs=new GenericOptionsParser(configuration,args).getRemainingArgs();
		if(otherArgs.length!=2){
			System.err.println("Usage:wordcount <in> <out>");
			System.exit(2);
		}
		
		//配置作业名
		Job job=new Job(configuration, "word count");
		job.setJarByClass(WordCount.class);
		job.setMapperClass(TokenizerMapper.class);
		job.setCombinerClass(IntSumReducer.class);
		job.setReducerClass(IntSumReducer.class);
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(IntWritable.class);
		FileInputFormat.addInputPath(job, new Path(otherArgs[0]));
		FileOutputFormat.setOutputPath(job, new Path(otherArgs[1]));
		System.exit(job.waitForCompletion(true)?0:1);
	}
}

MapReduce，多理解流程执行，属性对应的API，然后就是锻炼自己的建模的思维，算法的相关锻炼等……