Hadoop MapReduce

MapReduce包含哪些组件？

运行机制？>>>>数据流的分片是等长的吗？

运行框架？

离线计算框架？

局限点在哪？

每个版本的优化在哪一些方面？

一、基础知识：

　　1、MapReduce是一种可用于数据处理的编程模型。Hadoop可以运行各种语言版本的MapReduce程序。MapReduce模型简单，但是不易于编写有用的程序。

　　2、MapReduce程序本质上是并行运行的，大规模的数据分析任务分发给任何一个拥有足够多机器的数据中心。

　　3、使用多台电脑运行，整个大环境的其他因数就会相互影响，主要归结为协调性和可靠性两个方面。并行运行实际会有很多问题，可以充分利用Hadoop提供的并行处理优势，将查询表示为MapReduce作业，将集群中分布式计算中的复杂性交由Hadoop中的MapReduce框架来处理。

　　4、MapReduce任务分为Map阶段和Reduce阶段，每个阶段都以键值对作为输入和输出，类型根据实际进行选择。

　　5、Hadoop本身提供了一套可优化网络序列化传输的基本类型，这些基本类型都在org.apache.hadoop.io中，如：LongWritable 类型（Java的 Long 类型），Text 类型（Java的 String 类型），IntWritable类型（Java的 Integer 类型）

　　6、Map函数由Mapper类来表示，后者声明一个抽象的 map( )方法（业务逻辑）。每一个<k,v>调用一次！

//首先要定义四个泛型的类型
//keyin:  LongWritable    valuein: Text
//keyout: Text            valueout:IntWritable
 
public class WordCountMapper extends Mapper<LongWritable, Text, Text, IntWritable>{
	//map方法的生命周期：  框架每传一行数据就被调用一次
	//key :  这一行的起始点在文件中的偏移量
	//value: 这一行的内容
	@Override
	protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
		//拿到一行数据转换为string
		String line = value.toString();
		//将这一行切分出各个单词
		String[] words = line.split(" ");
		//遍历数组，输出<单词，1>
		for(String word:words){
			context.write(new Text(word), new IntWritable(1));
		}
	}
}

　　7、Reduce函数的输入类型必须匹配map函数的输出类型。Reduce函数由 Reducer 类来表示，后者声明一个抽象的 reduce( )方法（业务逻辑）。每一组<k,v>调用一次！

//生命周期：框架每传递进来一个<k,v> 组，reduce方法被调用一次
	@Override
	protected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
		//定义一个计数器
		int count = 0;
		//遍历这一组kv的所有v，累加到count中
		for(IntWritable value:values){
			count += value.get();
		}
		context.write(key, new IntWritable(count));
	}
}

　　8、MapReduce程序需要：一个 map 函数，一个 reduce 函数，一些用来运行作业的代码

9、运行作业的代码（Driver）：提交的是一个描述了各种必要信息的 job 对象

　　9.1、Job对象指定作业执行规范，通过Job对象控制整个作业的运行。在Hadoop集群运行作业，需要把代码打包成一个JAR文件。不需要指定JAR文件名称，在Job对象的job.setJarByClass( )方法中传递一个类，如( MaxTemperature.Class)即可。Hadoop利用这个类来查找包含此类的JAR文件，进而找到相关的JAR文件。

　　9.2、设置输入文件的路径（多次调用实现多路径）：FileInputFormat.addInputPath( )；设置reduce函数输出文件的写入目录（执行前不存在且唯一指定）：FileOutputFormat.setOutputPath( )

　　9.3、指定map类型：job.setMapperClass( )；指定reduce类型：job.setReducerClass( )；

　　9.4、Job.setPartitionClass设置Partition；Job.setOutputKeyComparatorClass进行设置，然后定义排序规则；job.setCombinerClass( )指定combiner类

　　9.5、控制reduce函数的输出类型（与Reducer类产生的相匹配）：job.setOutputKeyClass( Text.class )、job.setOutputValueClass( IntWritable.class )。默认情况下，mapper产生出和reducer相同的类型，所以没有设置map函数的输出类型。如果不同，需要通过：job.MapsetOutputKeyClass( )、job.MapsetOutputValueClass( )设置map函数的输出类型。

　　9.6、在Job.setInputFormat设置输入类型，默认 TextInputFormat（文本输入格式），有TextInputFormat，DBInputFormat，SequenceFileFormat等输入类型

　　9.7、job.waitForCompetion( )方法提交作业并等待作业执行完成。

package cn.test.bigdata.mr.wcdemo;
 
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.CombineTextInputFormat;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
 
public class WordcountDriver {
 
	
	public static void main(String[] args) throws Exception {
 
		
		Configuration conf = new Configuration();
		
		//设置的没有用!  ??????
//		conf.set("HADOOP_USER_NAME", "hadoop");
//		conf.set("dfs.permissions.enabled", "false");
		
		//不提交在yarn上面，只在本地跑
		conf.set("mapreduce.framework.name", "local");
		//本地模式运行mr程序时，输入输出的数据可以在本地，也可以在hdfs上
		//到底在哪里，酒宴以下两行配置，用的是哪行,默认是本地的
		conf.set("fs.defaultFS", "file:///");
		/*conf.set("fs.defaultFS", "hdfs://192.168.175.128:9000/");
		conf.set("mapreduce.framework.name", "yarn");
		conf.set("yarn.resoucemanager.hostname", "192.168.178.128");*/
		Job job = Job.getInstance(conf);
		
		/*job.setJar("/home/hadoop/wc.jar");*/
		//指定本程序的jar包所在的本地路径
		job.setJarByClass(WordcountDriver.class);
		
		//指定本业务job要使用的mapper/Reducer业务类
		job.setMapperClass(WordcountMapper.class);
		job.setReducerClass(WordcountReduce.class);
		
		//指定需要使用combiner，以及用哪个类作为combiner的逻辑
		/*job.setCombinerClass(WordcountCombiner.class);*/
		job.setCombinerClass(WordcountReduce.class);
		
		//如果不设置InputFormat，它默认用的是TextInputformat.class
		/*job.setInputFormatClass(CombineTextInputFormat.class);
		CombineTextInputFormat.setMaxInputSplitSize(job, 4194304);
		CombineTextInputFormat.setMinInputSplitSize(job, 2097152);*/
		
		//指定mapper输出数据的kv类型
		job.setMapOutputKeyClass(Text.class);
		job.setMapOutputValueClass(IntWritable.class);
		
		//指定最终输出的数据的kv类型
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(IntWritable.class);
		
		//指定job的输入原始文件所在目录
		FileInputFormat.setInputPaths(job, new Path("D:\wordstest\input"));
		//指定job的输出结果所在目录
		FileOutputFormat.setOutputPath(job, new Path("D:\wordstest\output"));
		
		//将job中配置的相关参数，以及job所用的java类所在的jar包，提交给yarn去运行
		/*job.submit();*/
		boolean res = job.waitForCompletion(true);
		System.exit(res?0:1);
		
	}
	
 
 
}

二、数据流

　　1、MapReduce作业（job）是客户端执行的一个工作单元，包括：输入数据、MapReduce程序、配置信息。Hadoop将作业（job）分成若干个任务（task）执行，其中包含两类任务map任务和reduce任务

　　2、“分片”：Hadoop将MapReduce输入数据划分为等长的小数据块，并为每一个分片构建一个map任务。map任务根据map函数并行运行，使得处理过程获得更好的负载平衡。对于大多数作业，一个合理的分片大小趋于 HDFS 的一个块大小（默认128M）。注：如果输入数据的分片大于一个HDFS数据块，对于任何一个HDFS节点，基本上都不可能同时存储这两个数据块，这就需要通过网络传输将部分数据传输到map任务执行的节点，导致效率降低

　　3、map任务执行的三种可能性：一、Hadoop在存储有输入数据的节点执行map函数，可以获得最优性能。因为这个过程无需使用宝贵的集群带框资源，这就是所谓的“数据本地化优化”。二、但是如果存储该分片的HDFS数据块副本的所有节点都在运行其他的map任务，此时yarn调度需要从某一数据块所在的机架的一个节点上寻找一个空闲的map槽（slot）来运行该map分片任务。三、甚至在极小概率下会使用其他机架的节点执行该map任务，这就导致机架与机架之间的网络传输，这基本不会发生！

　　4、map任务输出的是中间结果，由reduce任务任务处理后才产生最后的结果，任务完成，map任务的输出就会被删除，所以map任务的输出结果存储在本地磁盘而不是HDFS。如果一个map任务失败，它将在另一个不同的节点上自动重新调度执行

　　5、单个reduce任务的输入通常来自“所有的”mapper中对应的partition（分区）的输出，reduce的输出通常存储在HDFS中，以实现可靠存储。HDFS默认多复制块是3，第一个复本存储在本地节点上，其他复本出于可靠性考虑，存储在其他机架的节点上！将reduce任务的输出写入HDFS需要占用网络带宽。

　　6、reduce任务是独立指定的，用户可以通过job.setNumReduceTasks(4)为指定的作业选择reduce任务的数量，默认值是1，手动设置为4。

　　7、map任务和reduce任务之间的数据流称为shuffle（混洗），调整shuffle参数对作业总执行时间的影响非常大！如果没有reduce任务，数据处理完全并行，唯一的非本节点数据传输是map任务的输出结果写入HDFS

　　8、Hadoop允许用户针对于map任务的输出指定一个combiner函数（combiner组件的父类就是Reducer，combiner继承Reducer，重写reduce方法）。在一定情况下（不能影响最终的业务逻辑），使用combiner函数的输出作为reduce任务的输入。combiner函数尽量减小map任务和reduce任务之间的数据传输量，因为集群的带宽是有限的，集群上的可用带宽会限制MapReduce作业的数量！

　　9、Hadoop Streaming支持任何可以从标准输入读/写到标准输出中的编程语言。

三、MapReduce并行处理的基本过程

1、Mapreduce是一个分布式运算程序的编程框架

2、Mapreduce核心功能是将用户编写的业务逻辑代码和自带默认组件整合成一个完整的分布式运算程序，并发运行在一个hadoop集群上

3、MR流程示意图

图-1 MR流程示意图

4、 Map任务的并行度：一个job的map阶段并行度由客户端在提交job时决定。FileInputFormat实现类的getSplits()方法决定待处理数据执行逻辑切片的split的大小，而每个split分配一个 Map Task。

　　FileInputFormat中默认的切片机制：

简单地按照文件的内容长度进行切片
切片大小，默认等于block大小（128M）默认情况下，切片大小=blocksize
切片时不考虑数据集整体，而是逐个针对每一个文件单独切片

5、ReduceTask并行度的：通过 job.setNumReduceTasks(4);手动设置，默认为1，手动设置为4。如果数据分布不均匀，就有可能在reduce阶段产生数据倾斜

6、对大多数job来说，最好Reduce Task的个数最多和集群中的 reduce slots持平，或者比集群的 reduce slots 小

四、MapReduce程序运行模式

一、本地运行模式

mapreduce程序是被提交给LocalJobRunner在本地以单进程的形式运行
处理的数据及输出结果可以在本地文件系统，也可以在hdfs上
写一个程序，不要带集群的配置文件（本质是mr程序的conf中是否有mapreduce.framework.name=local以及yarn.resourcemanager.hostname参数）
本地模式非常便于进行业务逻辑的debug，只要在eclipse中打断点即可

注：如果在windows下想运行本地模式来测试程序逻辑，需要在windows中配置环境变量：
　　％HADOOP_HOME％ = d:/hadoop-2.6.1
　　%PATH% = ％HADOOP_HOME％in
　　并且要将d:/hadoop-2.6.1的lib和bin目录替换成windows平台编译的版本

二、集群运行模式

将mapreduce程序提交给yarn集群resourcemanager，分发到很多的节点上并发执行
处理的数据和输出结果应该位于hdfs文件系统(也需要启动yarn)
提交集群的实现步骤：需要在输入路径有文件

将程序打成JAR包，然后在集群的任意一个节点上用hadoop命令启动： $ hadoop jar wordcount.jar cn.test.bigdata.mrsimple.WordCountDriver inputpath outputpath
直接在linux的eclipse中运行main方法（项目中要带参数：mapreduce.framework.name=yarn以及yarn的两个基本配置）
如果要在windows的eclipse中提交job给集群，则要修改YarnRunner类

五、MapReduce的shuffle机制

一、shuffle是MR处理流程中的一个过程，整体来看，分为3个操作：Input->Map->Partition->Sort->Combine->Reduce->Output

分区partition
Sort根据key排序
Combiner进行局部value的合并

二、详细操作

maptask收集我们的map()方法输出的kv对，放到内存缓冲区中
从内存缓冲区不断溢出本地磁盘文件，可能会溢出多个文件（默认100M）
多个溢出文件会被合并成大的溢出文件
在溢出过程中，及合并的过程中，都要调用partitoner进行分组和针对key进行排序
reducetask根据自己的分区号，去各个maptask机器上取相应的结果分区数据
reducetask会取到同一个分区的来自不同maptask的结果文件，reducetask会将这些文件再进行合并（归并排序）
合并成大文件后，shuffle的过程也就结束了，后面进入reducetask的逻辑运算过程（从文件中取出一个一个的键值对group，调用用户自定义的reduce()方法）

六、MapReduce中的序列化

1、Java的序列化是一个重量级序列化框架（Serializable）。一个对象被序列化后，会附带很多额外的信息（各种校验信息，header，继承体系等），不便于在网络中高效传输

2、Hadoop开发了一套序列化机制（Writable），精简，高效。序列化的最主要的作用就是持久化存储或者是用于网络传输

3、Writable接口是一个实现了序列化协议的序列化对象。在Hadoop中定义一个结构化对象都要实现Writable接口，使得该结构化对象可以序列化为字节流，字节流也可以反序列化为结构化对象

4、将自定义的bean放在key中传输，则还需要实现comparable接口，因为mapreduce框中的shuffle过程一定会对key进行排序,此时，自定义的bean实现的接口应该是：

　　public class FlowBean implements WritableComparable<FlowBean>

public class FlowBean implements WritableComparable<FlowBean>{
	
	long upflow;
	long downflow;
	long sumflow;
	
	//如果空参构造函数被覆盖，一定要显示定义一下，否则在反序列时会抛异常
	public FlowBean(){}
	
	public FlowBean(long upflow, long downflow) {
		super();
		this.upflow = upflow;
		this.downflow = downflow;
		this.sumflow = upflow + downflow;
	}
	
	public long getSumflow() {
		return sumflow;
	}
 
	public void setSumflow(long sumflow) {
		this.sumflow = sumflow;
	}
 
	public long getUpflow() {
		return upflow;
	}
	public void setUpflow(long upflow) {
		this.upflow = upflow;
	}
	public long getDownflow() {
		return downflow;
	}
	public void setDownflow(long downflow) {
		this.downflow = downflow;
	}
 
	//序列化，将对象的字段信息写入输出流
	@Override
	public void write(DataOutput out) throws IOException {
		
		out.writeLong(upflow);
		out.writeLong(downflow);
		out.writeLong(sumflow);
		
	}
 
	//反序列化，从输入流中读取各个字段信息
	@Override
	public void readFields(DataInput in) throws IOException {
		upflow = in.readLong();
		downflow = in.readLong();
		sumflow = in.readLong();
		
	}

Mapper和Reducer：

public class FlowCount {
 
	static class FlowCountMapper extends Mapper<LongWritable, Text, FlowBean,Text > {
 
		@Override
		protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
 
			String line = value.toString();
			String[] fields = line.split("	");
			try {
				String phonenbr = fields[0];
 
				long upflow = Long.parseLong(fields[1]);
				long dflow = Long.parseLong(fields[2]);
 
				FlowBean flowBean = new FlowBean(upflow, dflow);
 
				context.write(flowBean,new Text(phonenbr));
			} catch (Exception e) {
 
				e.printStackTrace();
			}
 
		}
 
	}
 
	static class FlowCountReducer extends Reducer<FlowBean,Text,Text, FlowBean> {
 
		@Override
		protected void reduce(FlowBean bean, Iterable<Text> phonenbr, Context context) throws IOException, InterruptedException {
 
			Text phoneNbr = phonenbr.iterator().next();
 
			context.write(phoneNbr, bean);
 
		}
 
	}
 
	public static void main(String[] args) throws Exception {
 
		Configuration conf = new Configuration();
 
		Job job = Job.getInstance(conf);
 
		job.setJarByClass(FlowCount.class);
 
		job.setMapperClass(FlowCountMapper.class);
		job.setReducerClass(FlowCountReducer.class);
 
		 job.setMapOutputKeyClass(FlowBean.class);
		 job.setMapOutputValueClass(Text.class);
 
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(FlowBean.class);
 
		// job.setInputFormatClass(TextInputFormat.class);
 
		FileInputFormat.setInputPaths(job, new Path(args[0]));
		FileOutputFormat.setOutputPath(job, new Path(args[1]));
 
		job.waitForCompletion(true);
 
	}
 
}