Mapreduce

分布式运算程序编程框架 MapReduce

MapReduce概述

●源自Google的MapReduce论文，发表于2004年12月

●Hadoop Mapreduce是Google MapReduce的克隆版

●MapReduce优点：海量数据离线处理&易开发&易运行

●MapReduce缺点：实时流式数据

Mapreduce是一个分布式运算程序的编程框架，是用户开发“基于hadoop的数据分析应用”的核心框架。

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

分布式运算程序往往需要分成2个阶段：

1、第一个阶段的task（map task）并发实例各司其职，完全并行(map阶段)；

2、第二个阶段的task (reduce task) 并发实例互不相干，但是数据依赖于上一个阶段的task并发实例的输出（reduce阶段）。

Mapreduce编程模型只能包含一个map阶段和一个reduce阶段，如果用户业务逻辑复杂，那只能多个Mapreduce程序串行运行互相依赖。

另外需要MRAppMaster负责监控、调度、协调map task和reduce task。

MapReduce三个角色：map task、reduce task、MRAppMaster。

从WordCount看MapReduce的编程模型

1）文件小，直接使用shell

2）文件大，使用分布式计算框架

MapReduce编程模型之Map阶段和Reudce阶段

●将作业拆分成Map阶段和Reduce阶段

●Map阶段，Map tasks

●Reduce阶段：Reduce tasks

MapReduce编程模型之执行步骤

MapReduce处理<key,value>键值对。

1、准备map处理的输入数据

2、Mapper处理

3、Shuffle

4、Reduce处理

5、结果输出

MapReduce核心概念

Split

交由MapReduce作业来处理的数据块（是MapReduce中的最小的计算单元）

HDFS : blocksize 是HDFS中最小的存储单元 128M

默认情况：mapreduce的数据块和HDFS的block是一一对应，也可以自由设置。

InputFormat

将输入数据进行分片（Split）

TextInputFormat:处理文本格式的数据

OutputFormat

输出

MapReduce版本

MapReduce1.x

1) JobTracker：JT

作业的管理者

将作业分解成一堆任务:Task(MapTask和ReduceTask)

将任务分派给TaskTracker运行

作业的监控、容错处理（task挂了，重启task机制）

在一定时间间隔内，JT没有收到TT的心跳信息，TT可能挂了，TT上运行的任务会被指派到其他TT上执行

2)TaskTracker：TT

任务的执行者

在TT上执行Task(MapTask和ReduceTask)

与JT进行交互：执行/启动/停止作业，发送心跳信息给JT

3)MapTask

自己开发的map任务交由该Task处理

解析每条记录的数据，交给自己的map方法处理

将map输出结果写到本地磁盘（有些作业只有map没有reduce，就写到hdfs）

4）ReduceTask

将Map Task输出的数据进行读取

按照数据进行分组传给我们自己编写的reduce方法处理

输出结果到HDFS

MapReduce2.x

没有了JobTracker和TaskTracker，出现了Resoure manager和Node manager。

其实就是引入了YARN。

Combiner

●本地的reducer

●减少Map Tasks输出的数据量及数据网络传输量

在Map端做了一个本地端的reduce操作。

使用场景：求和、次数。而不适用于求平均数的概念。

Partitioner

●Partitioner决定了MapTask输出的数据交由哪个ReduceTask处理

●默认实现：分发的key的hash值对Reduce Task个数取模

有几个partitioner就有几个reduce输出。

JobHistory

●记录已运行完的MapReduce信息到指定的HDFS目录下

●默认是不开启的

1、修改mapred-site.xml配置文件

2、修改yarn-site.xml开启聚合功能

3、启动：./mr-jobhistory-deaom.sh start historyserver

WordCount 编码

Map task和Reduce task都由Hadoop封装好了，自己只需要写map()方法和reduce()方法即可。

需求：在一堆给定的文本文件中统计输出每一个单词出现的总次数

Mapper：

Mapper类：

Mapper<KEYIN,VALUEIN,KEYOUT,VALUEOUT>

Mapper<LongWritable ,Text,Text,IntWritable>

● KEYIN : 默认情况下，是mr框架所读到的一行文本的起始偏移量，Long，在Hadoop中有自己的更精简的序列化接口，所以

不直接用Long，而用LongWritable。

● VALUEIN : 默认情况下，是mr框架所读到的一行文本的内容，String，同上使用Text（hadoop.io.text）

● KEYOUT :用户自定义逻辑处理完成之后输出数据中的key，此处是是单词，String，同上使用Text（hadoop.io.text）

● VALUEOUT : 用户自定义逻辑处理完成之后输出数据中的value，此处是单词次数，Integer 同上使用IntWritable

public class WordcountMapper extends Mapper<LongWritable,Text,Text,IntWritable>{

/**
*map阶段的业务逻辑就写在自定义的map()方法中
*maptask会对每一行输入数据调用一次我们自定义的map()方法
*/
@Override
protected void map(LongWritable key,Text value,Context context) throws IOException,InterruptedException{
//将map task传给我们的文本内容先转换成String
String string = value.toString();
//根据空格将这一行单词分成单词
String[] words = line.split(" ");

//将单词输出为<单词，1>
for(String word:words){
//将单词作为key,将次数1作为value，以便于后续的数据分发，可以根据单词分发
//以便于相同单词会到相同的reduce task中
context.write(new Text(word),new IntWritable(1));
}
}
}

Reducer:

Reducer<KEYIN,VALUEIN,KEYOUT,VALUEOUT>
KEYIN,VALUEIN 对应mapper输出的 KEYOUT,VALUEOUT 类型对应
KEYOUT ,VALUEOUT 是自定义reduce逻辑处理结果的输出数据类型
KEYOUT是单词
VALUEOUT是总次数

public class WordcountReducer extends Reducer<Text,IntWritable,Text,IntWritable>{

/**
*入参key,是一组相同单词kv对的key
*<hello,1><hello,1><hello,1><hello,1><hello,1><hello,1>
*<banana,1><banana,1><banana,1><banana,1><banana,1>
*入参key，传的是组里面的第一个单词
*/
@Override
protected void reduce(Text key,Iterable<IntWritable> values,Context context) throws IOException,InterruptedException{
int count=0;
for(IntWritable value : values){
count +=value.get();
}
context.write(key,new IntWritable(count));
}
}

每个Reduce的context.write写入hdfs文件。


/**
*相当于yarn集群的客户端
*需要再次封装mr程序的相关运行参数，指定jar包
*最后提交给jar
*/
public class WordcountDriver{
public static void main(String[] args) throws Exception{
Configuration conf = new Configuration();
conf.set("mapredue.framework.name","yarn");
conf.set("yarn.resourcemanager.hostname","mini1");
Job job = Job.getInstance();

//指定本程序的jar包所在的本地路径
job.setJarByClass(WordcoutDriver.class);

//告诉本业务job要使用的mapper/reducer业务类
job.setMapperClass(WordcountMapper.class);
job.serReducerClass(WordcountReducer.class);

//指定mapper输出数据的kv类型
job.setMapOutputKeyClass(Text.class);
job.setMapOutputValueClass(IntWritable.class);

//指定最终输出的数据的kv类型（有些运算无需reducer）
job.seOutputKeyClass(Text.class);
job.seOutputValueClass(IntWritable.class);

//指定job的输入原始文件所在目录
FileinputFormat.setInputPaths(job, new Path(args[0]));
//指定job的输出结果所在目录
FileOutputFormat.setOutputPaths(job, new Path(args[1]));

//将job中配置的相关参数，以及job所用的java类所在的jar包，提交给yarn去运行
//job.submit();
boolean res = job.waitForCompletion(true);
System.exit(res?0:1);
}
}

java -cp wordcount.jar cn.itcast.bigdata.mr.wcdemo.WordcountDriver /wordcount/input /wordcount/output
或者
hadoop jar wordcount.jar  cn.itcast.bigdata.mr.wcdemo.WordcountDriver /wordcount/input /wordcount/output

mapreduce默认是一个。

输出文件是按照单词的字母顺序排序的。

总结

（1）用户编写的程序分成三个部分：Mapper、Reducer、Driver(提交运行mr程序的客户端)

（2）Mapper的输入数据是KV对的形式（KV的类型可自定义）

（3）Mapper的输出数据是KV对的形式（KV的类型可自定义）

（4）Mapper中的业务逻辑写在map()方法中

（5）map()方法（maptask进程）对每一个<K,V>调用一次

（6）Reducer的输入数据类型对应Mapper的输出数据类型，也是KV

（7）Reducer的业务逻辑写在reduce()方法中

（8）Reducetask进程对每一组相同k的<k,v>组调用一次reduce()方法

（9）用户自定义的Mapper和Reducer都要继承各自的父类

（10）整个程序需要一个Driver来进行提交，提交的是一个描述了各种必要信息的job对象

WordCount如何运行

MapTask并行度决定机制

maptask的并行度决定map阶段的任务处理并发度，进而影响这个job的处理速度。

一个job的map阶段并行度的规划的基本逻辑为：

将待处理数据执行逻辑切片（即按照一个特定切片大小，将待处理数据划分成逻辑上的多个split），然后每一个split分配给一个maptask并行实例处理。

这段逻辑及形成的切片规划描述文件，由InputFormat实现类的getSplits()方法完成。

任务切片是一个逻辑划分，它与block没有关系。

一个切片对应一个maptask实例。

FileInputFormat中默认的切片机制：

a)简单地按照文件的内容长度进行切片；

b)切片大小，默认等于block大小;

c)切片时不考虑数据集整体，而是逐个针对每一个文件单独切片

FileInputFormat中切片大小的参数配置：

在FileInputFormat中，计算切片大小的逻辑：

Math.max(minSize,Math.min(maxSize,blockSize)) ；

切片主要有几个值来确定：

minSize: 默认1，配置参数：mapreduce.input.fileinputformat.split.minsize

maxSize:默认Long.MAXValue ，配置参数：mapreduce.input.fileinputformat.split.maxsize

因此，面膜人情况下，切片大小=blocksize

maxsize(切片最大值)：参数如果调得比blocksize小，则会让切片变小，而且就等于配置的这个参数的值；

minsize(切片最小值)：参数调的比blocksize大，则可以让切片变得比blocksize还大

客户端提交Job的流程

ReduceTask并行度决定机制

reducetask的并行度同样影响整个job的执行并发度和执行效率，但与maptask的并发数由切片数决定不同，reducetask

数量的决定是可以直接手动设置：

//默认值是1，手动设置为4

job.setNumReduceTask(4);

如果数据分布不均匀，就有可能在reduce阶段产生数据倾斜。

注意：reducetask数量并不是任意设置，还需要考虑业务逻辑需求，有些情况下，需要计算全局汇总结果，就只能有1个

reducetask。

尽量不要运行太多reducetask。