20190319&20-MapReduce编程模型和YARN平台

MapReduce编程模型

MapReduce是一种可用于数据处理的编程模型。该模型比较简单，但用于编写有用的程序并不简单。Hadoop可以运行由各种语言编写的MapReduce程序。例如：Java、Python和C++语言等。最重要的是，MapReduce程序本质上是并行运行的，因此可以将大规模的数据分析任务交给任何一个拥有足够多机器的运行商。MapReduce的优势在于处理大规模数据集

从MapReduce自身的命名特点可以看出，MapReduce由两个阶段组成：Map和Reduce。用户只需编写map()和reduce()两个函数，即可完成简单的分布式程序设计

map()函数以key/value对作为输出，产生另外一系列key/value对作为中间输出写入本地磁盘。MapReduce框架会自动将这些中间数据按照key值进行聚合，且key值相同的数据被统一交给reduce()函数处理。

reduce()函数以key及对应的value列表作为输入，经合并key相同的value值后，产生另外一系列key/value对作为最终输出写入HDFS。

• 设计目的
  易于编程
  良好的扩展性
  高容错性

wordcount词频统计

[[email protected] ~]$ hadoop jar hadoop-2.7.7/share/hadoop/mapreduce/hadoop-mapreduce-examples-2.7.7.jar wordcount /0317/s.txt /out

这里有两个参数，分别是输入路径和输出路径，输入路径是要进行词频统计的文件，输出路径必须是hdfs上本身不存在的新目录

WordCount编程实例

WordMapper.java

package org.yao.mr;

import java.io.IOException;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

// 创建WordMapper类，此类继承Mapper类
// Mapper<Object, Text, Text, IntWritable>
//object表示的是map输入的key的类型。一行内容或者一行记录的偏移量，第一个数是0，代表的是从文件的第一个位置开始
//text：map输入的value的类型。表示的是一行记录或者一行文本
//text：是map阶段输出的key的类型，文本类型
//intWritable：是map阶段输出的value类型
public class WordMapper extends Mapper<Object, Text, Text, IntWritable> {
@Override
//map()函数
//key就是偏移量，value就是传入的一行文本，context是联系上下文的一个对象，用它把结果传给reduce阶段
protected void map(Object key, Text value, Mapper<Object, Text, Text, IntWritable>.Context context)
		throws IOException, InterruptedException {
	//把这一行文本（text）拿取过来转换成string类型的数据
	String lines=value.toString();
	//按照“，”进行分割数据，同时存入数组中
	//张三，16，男，。。。//context.write（new Text（arr[0],new intWritable(arr[1]))
	
	//如果是tab键就是/t
	String[] arr=lines.split(" ");
	//把<hadoop，1>发送给reduce
	for (String word:arr){
	context.write(new Text(word), new IntWritable(1));
	}
}
}

WordReducer.java

package org.yao.mr;

import java.io.IOException;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

//Reducer<Text, IntWritable, Text, IntWritable>
//前两个参数代表的是接收的map阶段输出的数据类型，数据类型需要和map输出数据类型保持一致
//后两个参数是代表的是经过reduce阶段输出的数据类型
public class WordReducer extends Reducer<Text, IntWritable, Text, IntWritable> {
@Override
//reduce（）函数
//k2这个参数就是由map阶段传入的key值，因为已经分组排序了，所以不用对它做任何的处理
//values代表的是一个集合，也可以叫做中间输出结果，比如<hadoop,{1,1,1,1}>
//context是联系上下文，主要用于把输出结果发送出去
protected void reduce(Text k2, Iterable<IntWritable> values,
		Reducer<Text, IntWritable, Text, IntWritable>.Context context) throws IOException, InterruptedException {
	//用来保存每一个key值，出现的次数
	int sum=0;
	//通过foreach遍历得到每一个值
	for (IntWritable val : values) {
		//把获取到的每一个值进行相加
		sum+=val.get();
	}
	//把每个key，计算得出的数量发送出去
	context.write(k2, new IntWritable(sum));
}
}

WordMain.java

package org.yao.mr;

import java.io.IOException;

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.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

public class WordMain {

	public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
		//获取hdfs的配置信息
		Configuration conf=new Configuration();
		//设置集群模式，如果不写就是取本地了
		conf.set("fs.defaultFS", "hdfs://192.168.16.199:9000");
		//用hadoop集群的用户进行访问
		System.setProperty("HADOOP_USER_NAME", "yao");
		//创建一个job，（类似于客户端给一个写请求或者读请求）把集群配置信息放入这个job中
		Job job=Job.getInstance(conf);
		//给这个应用程序添加jar包
		job.setJarByClass(WordMain.class);
		
		//如果我们的map的输出和reduce的输出的类型不一致的时候，需要进行下面的配置
		//job.setMapOutputKeyClass(Text.class);
		//job.setMapOutputValueClass(IntWritable.class);
		
		//设置自定义的Mapper类
		job.setMapperClass(WordMapper.class);
	    //设置自定义的Reducer类
		job.setReducerClass(WordReducer.class);
		
		//combineClass运行在reduceClass之前，相当于一个小型的reduceClass，key值相同的value合并成集合，然后再交给reduce处理，这样可以减少map和reduce时间传输的数量，减少reduce的压力，对运行时间可能有一定的优势。
		//job.setCombinerClass(WordReducer.class);
		
		//reduce的数目
		//job.setNumReduceTasks(4);
		
		//设置（reduce端）文件的输出的key的类型
		job.setOutputKeyClass(Text.class);
		//设置文件的输出的value的类型////如果我们的map的输出和reduce的输出的类型不一致的时候，下面就要写新的类型比如doubleIntWritable
		job.setOutputValueClass(IntWritable.class);
		
		
		//添加文件的输出格式
		FileInputFormat.addInputPath(job, new Path(args[0]));
		//给出最后结果的输出路径，这个路径必须是一个不存在的路径
		FileOutputFormat.setOutputPath(job, new Path(args[1]));
		
		//运行job
		job.waitForCompletion(true);
		

	}

}

现在直接在windows的eclipse下运行MapReduce时会报错

Failed to locate the winutils binary in the hadoop binary path
java.io.IOException: Could not locate executable null\bin\winutils.exe in the Hadoop binaries.

• 解决办法：
  方法一：
  需要将代码打成jar包，上传到hdfs后再运行

java打jar包：



将jar包上传到linux系统上后执行

hadoop jar dou.jar /dou.txt /dou
hadoop jar kong.jar /word.txt /kong

结果：


方法二：
在https://github.com/SweetInk/hadoop-common-2.7.1-bin中
1.下载winutils.exe和libwinutils.lib将它们拷贝到$HADOOP_HOME/bin目录下
2.下载hadoop.dll，将其拷贝到C:\Windows\system32目录下

shuffle

1. 概述
   MapReduce 中，map 阶段处理的数据如何传递给 reduce 阶段，是 MapReduce 框架中最关键的一个流程，这个流程就叫Shuffle。
   shuffle就是洗牌的过程，它会将数据进行分组排序。
   具体来说它就是将 maptask 输出的处理结果数据，分发给 reducetask，并在分发的过程中，对数据按 key 进行了分组和排序。
2. 主要流程
   

• map端
  map函数开始产生输出时，利用缓冲的方式写到内存，并出于效率的考虑进行预排序。
  每个map任务都有一个环形内存缓冲区，用于存储任务的输出，默认情况是100MB，一旦缓冲内容达到阈值（默认是80%），一个后台线程开始把内容写到磁盘中。在写磁盘过程中，map输出继续被写到缓冲区，但如果在此期间缓冲区被填满，map会阻塞直到写磁盘过程完成。
  在写磁盘之前，线程首先根据数据最终要传送到reducer把数据划分成相应的分区，在每个分区中，后台线程按键进行内排序，如果有一个combiner，它会在排序后的输出上运行
  一旦内存缓冲区达到溢出写的阈值，就会新建一个溢出写文件，因此在map任务写完其最后一个输出记录之后，会有几个溢出写文件。在任务完成之前，溢出写文件被合并成一个已分区且已排序的输出文件

• reduce端
  reducer通过HTTP方式得到输出文件的分区，这就是reduce任务的复制阶段(copy phase)。reduce任务有少量复制线程，因此能够并行取得map输出。默认值是5个线程
  与map端一样，reduce端将数据复制过来的时候先被复制到内存中，当内存中的数据量到达一定阈值，就启动内存到磁盘的merge，也就是溢写文件，然后会在磁盘中生成了众多的溢写文件，直到没有map端的数据输出为止，然后进行大合并，将多个溢写文件合并成一个文件，最后，直接把数据输入reduce函数，然后输出。

---

YARN平台

1. 产生背景
   

• 扩展性差：难以支持MR之外的计算
• 资源利用率低：不能将硬件资源利用到最高
• 可靠性差：NameNode单点故障

2. 架构与组件
   

• ResourceManager
  ResourceManager 是基于应用程序对集群资源的需求进行调度的 Yarn 集群主控节点，负责协调和管理整个集群（所有 NodeManager）的资源，通过AppMaster响应用户提交的不同类型应用程序的解析，调度，监控等工作。ResourceManager 会为每一个 Application 启动一个 ApplicationMaster， 并且 ApplicationMaster 分散在各个 NodeManager 节点

• NodeManager
  管理每一个节点的资源，并上报给RM。
  NodeManager 是 YARN 集群当中真正资源的提供者，是真正执行应用程序的容器的提供者， 监控应用程序的资源使用情况，并通过心跳向集群资源调度器 ResourceManager 进行汇报。

• ApplicationMaster
  ApplicationMaster 对应一个应用程序，职责是向资源调度器申请执行任务的资源容器，监控整个任务的执行，跟踪整个任务的状态，处理任务失败以异常情况

• Container
  Container 是一个抽象出来的逻辑资源单位。它封装了一个节点上的 CPU，内存，磁盘，网络等信息，MapReduce 程序的所有 task 都是在这个容器里执行完成的

3. 作业运行过程
（JVM是java里的虚拟机）

• 作业提交
  Client端发出请求运行一个Job（第1步），请求资源管理器（ResourceManager），获取一个作业ID（应用ID）（第2步），作业的客户端核实作业的输出, 计算输入的split, 将作业的资源(包括Jar包, 配置文件, split信息)拷贝给HDFS（第3步），最后, 通过调用资源管理器的submitApplication()方法来提交作业（第4步）
• 作业初始化
  当RM资源管理器收到submitApplication()提交申请的请求时, RM判断那个NN比较空闲，就将该请求发给调度器(scheduler), 调度器分配第一个container, 并与对应的 NodeManager 通讯，然后资源管理器在该container内启动应用管理器（Application Master）进程， 由节点管理器（NodeManager）监控（第5a和5b步）
  MapReduce作业的application master是一个Java应用程序，它的主类是MRAppMaster。MRAppMaster对作业进行初始化：通过创建多个对象以保持对作业进度的跟踪，因为他将接受来自任务的进度和完成报告（第6步）
  MRAppMaster接受来自共享文件系统的在客户端计算的输入分片（第7步），去HDFS拿取作业的具体要求，比如运行什么程序，在什么环境等。（去HDFS拿取作业的具体要求，比如运行什么程序，在什么环境等）对每一个分片创建map任务对象以及reduce任务对象。
• 任务分配
  如果不是小作业, 那么MRAppMaster应用管理器向RM资源管理器采用轮询的方式通过 RPC 协议向 ResourceManager 申请和领取资源，MRAppMaster请求第二个container来运行所有的maptask和reducetask任务(第8步)，每个任务对应一个container，且只能在该container上运行，这些请求是通过心跳来传输它们的状态。如果当前的NN空闲，就在当前的NN上运行，如果当前的NN比较忙，就去另外的NN上运行。
• 任务执行
  MRAppMaster 申请到任务的资源后，资源管理器的调度器为NN分配container，MRAppMaster与对应的 NodeManager 通讯， MRAppMaster应用管理器通过练习节点（NodeManager）管理器来启动container(第9a步和9b步).
  任务有一个主类为YarnChild的Java应用执行. 在运行任务之前首先本地化任务需要的资源, 比如作业配置, JAR文件, 以及分布式缓存的所有文件(第10步).
  NodeManager 为任务设置好运行环境（包括环境变量、JAR 包、二进制程序等）后，将任务启动命令写到一个脚本中，并通过运行该脚本启动任务。
  最后, 运行map或reduce任务(第11步).
• 进度和状态更新
  YARN中的任务将其进度和状态(包括counter)返回给AppMaster应用管理器, 后者通过每3秒的脐带接口有整个作业的视图。各个任务通过某个 RPC 协议向 MRAppMaster 汇报自己的状态和进度，以让 MRAppMaster随时掌握各个任务的运行状态，从而可以在任务败的时候重新启动任务。
• 作业完成
  除了向应用管理器请求作业进度外, 客户端每5分钟都会通过调用waitForCompletion()来检查作业是否完成。应用程序运行完成后，MRAppMaster 向 ResourceManager 注销并关闭自己。