你好,游客 登录 注册 搜索
背景:
阅读新闻

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

[日期:2015-09-09] 来源: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);
}
}

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





收藏 推荐 打印 | 录入:elainebo | 阅读:
本文评论   查看全部评论 (0)
表情: 表情 姓名: 字数
点评:
       
评论声明
  • 尊重网上道德,遵守中华人民共和国的各项有关法律法规
  • 承担一切因您的行为而直接或间接导致的民事或刑事法律责任
  • 本站管理人员有权保留或删除其管辖留言中的任意内容
  • 本站有权在网站内转载或引用您的评论
  • 参与本评论即表明您已经阅读并接受上述条款