MapReduce 的实际处理过程可以分解为 Input、Map、Sort(排序)、Combine(局部合并)、Partition(分区)、Reduce(全局聚合)、Output 等阶段。
1 MapReduce的作用
化大为小、分而治之
2 Map和Reduce阶段的作用
Map阶段:
把输入变成(Key,Value)结果,用于reduce阶段的输入.(结果局部有序)
Reduce阶段:
将map阶段的结果按分组进行汇总.(结果全局有序)
MapReduce 计算架构提供的主要功能包括以下几点:
1）任务调度提交的一个计算作业(Job)将被划分为很多个计算任务(Tasks)多个map一起进行。一个数据块就要跑一个map,假如感觉reduce特别慢,可以增加reduce的数量.
任务调度功能主要负责为这些划分后的计算任务分配和调度计算结点(Map 结点或 Reduce 结点),同时负责监控这些结点的执行状态,以及 Map 节点执行的同步控制,也负责进行一些计算性能优化处理.
例如,对最慢的计算任务采用多备份执行,选最快完成者作为结果(推测执行).
2）数据/程序互定位为了减少数据通信量,一个基本原则是本地化数据处理,即一个计算结点尽可能处理其本地磁盘上分布存储的数据,这实现了代码向数据的迁移.
当无法进行这种本地化数据处理时,再寻找其他可用结点并将数据从网络上传送给该结点(数据向代码迁移),但将尽可能从数据所在的本地机架上寻找可用结点以减少通信延迟.
3）出错处理在以低端商用服务器构成的大规模 MapReduce 计算集群中,结点硬件(主机、磁盘、内存等)出错和软件有缺陷是常态。
因此,MapReduce 架构需要能检测并隔离出错结点,并调度分配新的结点接管出错结点的计算任务.
4）分布式数据存储与文件管理海量数据处理需要一个良好的分布数据存储和文件管理系统作为支撑,该系统能够把海量数据分布存储在各个结点的本地磁盘上,且保持整个数据在逻辑上成为一个完整的数据文件.
为了提供数据存储容错机制,该系统还要提供数据块的多备份存储管理能力.
5）Combiner(整合器) 和 Partitioner(划分器)Combiner : map阶段的数据合并功能.Partitioner : 可以指定多个reduce进行数据聚合工作,每个我想要统计的数据一定会到某个同样的reduce中去,如此统计数据就快了,但缺点是数据分散了,会产生多个数据结果.为了减少数据通信开销,中间结果数据进入 Reduce 结点前需要进行合并(Combine)处理,即把具有同样主键的数据合并到一起避免重复传送.
3 输入文件的split原理
在map阶段读取数据前,FileInputFormat会将输入文件分割成split,split的个数决定了map的个数,影响map个数(split个数)的主要因素有：
1) 文件的大小
当块(dfs.block.size)为128m时,如果输入文件为128m,会被划分为1个split;当块为256m,会被划分为2个split.
2) 文件的个数
FileInputFormat按照文件分割split,并且只会分割大文件,即那些大小超过HDFS块大小(dfs.block.size)的文件.
如果HDFS中dfs.block.size设置为128m,而输入的目录中文件有100个,则划分后的split个数至少为100个.
3) splitsize的大小
分片是按照splitszie的大小进行分割的,一个split的大小在没有设置的情况下,默认等于hdfs block的大小.
配置文件和代码中的设置：
# 应用程序可以通过两个参数来对splitsize进行调节
InputSplit=Math.max(minSize, Math.min(maxSize, blockSize)
# 其中：
minSize=mapred.min.split.size
maxSize=mapred.max.split.size# 我们可以在MapReduce程序的驱动部分添加如下代码：
TextInputFormat.setMinInputSplitSize(job,1024L); # 设置最小分片大小
TextInputFormat.setMaxInputSplitSize(job,1024×1024×10L); # 设置最大分片大小
# 总结如下：
当mapreduce.input.fileinputformat.split.maxsize > mapreduce.input.fileinputformat.split.minsize > dfs.blockSize的情况下，此时的splitSize 将由mapreduce.input.fileinputformat.split.minsize参数决定
当mapreduce.input.fileinputformat.split.maxsize > dfs.blockSize > mapreduce.input.fileinputformat.split.minsize的情况下，此时的splitSize 将由dfs.blockSize配置决定
当dfs.blockSize > mapreduce.input.fileinputformat.split.maxsize > mapreduce.input.fileinputformat.split.minsize的情况下，此时的splitSize将由mapreduce.input.fileinputformat.split.maxsize参数决定。
4 Partition算法
为什么要有partition这个过程存在：
一个 reduce 结点所处理的数据可能会来自多个map结点,如果所有的map节点结果数据都拉去到同一个reduce上去,极有可能造成OOM(内存溢出);
因此,map结点输出的结果需使用一定的策略进行适当的划分(partition)处理,保证相同key的数据进入同样的reduce,如此统计数据就快了而且保证了计算结果的准确性.
注意 : 有一个reduce的情况下,就没有partition过程. 有多个reduce的情况下,就会有partition过程.
为什么要把相同的key进入到同一个reduce中计算
为了保证计算结果的快速、准确.
如何保证相同的key进入到同一个reduce中：
# hashpartion算法
PID(partition id) = (key.hashcode&Integer.max_value) % (reduce的个数)==
# (key的hashcode值与Integer的最大值进行位运算) % (reduce的个数)# 假设：reduce数量是2
PID = 0
PID = 1# 假设：reduce数量是3
PID = 0
PID = 1
PID = 2# reduce 数量决定了partition 数量。
5 Map与Reduce之间的通信
在 MapReduce 整个处理过程中,不同的 Map 任务之间不会进行任何通信,不同的 Reduce 任务之间也不会发生任何信息交换.
用户不能够显式地从一个结点向另一个结点发送消息,所有的信息交换都是通过 MapReduce 框架实现的.
MapReduce 计算模型之所以得到如此广泛的应用,就是因为应用开发者不需要处理分布式和并行编程中的各种复杂问题.
如分布式存储、分布式通信、任务调度、容错处理、负载均衡、数据可靠等,这些问题都由 Hadoop MapReduce 框架负责处理,
应用开发者只需要负责完成 Map 函数与 Reduce 函数的实现.
Hadoop应用最广泛的三种数据传输方式:
HTTP 请求与响应,map与reduce之间的通信方式
RPC 在namenode和datanode之间,datanode和datanode之间进行通信时使用.
NIO+SOCKET 在HDFS的客户端向datanode存数据的时候使用(因为可以开通道进行操作).
为什么Map与Reduce之间的通信用http协议呢？？？
http是请求与响应模式,在map向applicationmaster发送状态后到接收到reduce的请求前,会有一个时间间隔;
在这段时间内不能保证map的数据有没有发生改变,为了保证reduce获取数据的准确性,用http请求与响应模式.
Map与Reduce之间的通信流程：
map向applicationmaster发送自己的状态,applicationmaster将map的状态告知reduce;
此时reduce会向map发送一次请求,询问数据状态是否正确;
如果map响应正确,则reduce即开始从map拉取数据;
如此,可以避免拉取数据的重复或不成功的发生,保证拉取数据的准确性.
