最早接触大数据,常萦绕耳边的一个词「MapReduce」。它到底是什么,能做什么,原理又是什么?且听下文讲解。
是什么
MapReduce 即是一个编程模型,又是一个计算框架,它充分采用了分治的思想,将数据处理过程拆分成两步:Map 和 Reduce。用户只需要编写 map() 和 reduce() 函数,就能使问题的计算实现分布式,并在Hadoop上执行。
数据处理
MapReduce 操作数据的最小单位是一个键值对。map 端的主要输入是一对<key,value>值,经过 map 计算后输出一对<key,value>,然后将相同的 key 合并,形成<key,value 集合>,再将这个<key,value 集合>输入 reduce ,经过计算输出零个或多个<key,value>对。
两个重要的进程
JobTracker
JobTracker 在集群中负责任务调度和集群资源监控这两个功能。TaskTracker 通过周期性的心跳向 JobTracker 汇报当前的健康状况和状态,心跳中包括自身计算资源的信息、被占用的计算资源的信息和正在运行中的任务的状态信息。JobTracker 会根据各个 TaskTracker 周期性发送过来的心跳信息综合考虑TaskTracker 的资源余量、作业优先级、作业提交时间等因素,为 TaskTracker 分配合适的任务。
JobTracker 提供了一个基于 web 的管理界面,可以通过 JobTracker:50030 端口访问。
TaskTracker
TaskTracker 主要负责汇报心跳和执行 JobTracker 命令这两个功能。命令主要包括5种:启动命令、提交命令、杀死任务、杀死作业和重新初始化。
几个概念
作业(Job) 和 任务(Task)
MapReduce 作业是用户提交的最小单位,任务是 MapReduce 计算的最小单位。 简单讲,用户提交的是一个MapReduce作业,一个 MapReduce 作业可以被拆分成两种——Map 任务和 Reduce 任务。
槽(slot)
槽是Hadoop计算资源的表示模型,Hadoop 将各个节点上的多维度资源(CPU、内存等)抽象成一维度的槽。一个TaskTracker 能够启动的任务数量是由 TaskTracker 配置的任务槽决定的。
MapReduce 过程
一个MapReduce作业通常经过 input、map、combine、reduce、output 五个阶段。combine 阶段不一定发生,map输出的中间结果分发到 reduce 的过程被称为 shuffle。shuffle 阶段还会发生 copy 和 sort。
两幅重要的流程图
- map任务流程图
- reduce 任务流程图
几个重要的阶段说明
map 函数处理后的中间结果会写到本地磁盘上,在刷写磁盘的过程中,还做了 partition 和 sort 操作。
map 函数输出时,并不是简单地刷写磁盘,为了保证 I/O 效率,采取了先写到内存的环形缓冲区,并做一次预排序。请结合map任务流程图理解。
partition
在分区阶段,通过对 key 取模,生成<partition,key,value>三元组,分区阶段进行了一次内排序。
MemoryBuffer
内存缓冲区,保存 map 的结果和 partition 处理后的结果,默认大小为100M,溢写阈值为80M。
spill(溢写)
内存缓冲区达到阈值时,溢写线程锁住这80M的缓冲区,开始将数据写到本地磁盘中,然后释放内存。
每次溢写都会生成一个数据文件,溢出的数据写到磁盘前会对数据进行 sort 以及合并(combine)。
combine
combine 对map 函数的输出结果进行早期聚合以减少传输的数据量,其作用其实和reduce 函数一样。combine 的过程发生在 spill(溢写) 阶段。
combine 能够提升程序性能,但并不是所有常见都适合使用 combine ,例如:求中值。
sort
MapReduce 计算框架主要用到了两种排序:快速排序和归并排序。在 Map 任务和 Reduce 任务的过程中,一共发生了三次排序操作:
- partition 过程中按照键值进行的内排序。
- map 任务完成之前,合并溢写文件产生输出文件时进行的一次 sort 操作。
- shuffle 过程的 sort 操作。
wordcount 实验模拟
map 端编程代码(map_a.py):
import sys
import rep =re.compile(r'\w+')
for line in sys.stdin:world_list =line.strip().split()for word in world_list:if len(word)<2:continuew_list =p.findall(word)if len(w_list)>0:w =w_list[0].lower()print "%s\t%d"%(w,1)
reduce 端编程代码(red_b.py)
import sys
wt =0
cur_word =None
for line in sys.stdin:word,cnt =line.strip().split('\t')if cur_word ==None:cur_word =wordif cur_word !=word:print "%s\t%d"%(cur_word,wt)wt =0cur_word =wordwt =wt+int(cnt)
print "%s\t%d"%(cur_word,wt)
模拟命令
cat The_man_of_property.txt |python ./project/map_a.py | sort -k 1 |python ./project/red_b.py