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

编辑：　发布时间： 2015-9-9 　　文章来源：CSDN博客

大数据营销

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

一，好，首先来看下边两个图，看下 map reduce 在 Hadoop 中的执行流程，以及mapreduce 内部的执行流程：

以分析气象数据为例子：

二，解析： map reduce 的执行步骤：

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 ，推测式执行：即如果 job tracker 发现有拖后腿的任务，会再启动一个相同的备份任务，然后那个先执行完就会 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 ，硬件故障，即 job tracker 和 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次（可以设置），将不会再被执行，同时作业也宣布失

五，最后来看一个 word Count 的例子：

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