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

标签：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还无法处理， 唯有选择最牢靠的硬件作为

jobtracker

B,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);
	}
}

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

标签：hadoop   mapreduce   

原文地址：http://blog.csdn.net/liujiahan629629/article/details/48301427