MapReduce详解和WordCount模拟

最早接触大数据，常萦绕耳边的一个词「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 

输出显示