|
map: (k1,v1)-->[(k2,v20)] reduce: (k2,[v2])-->[(k3,v3)] 其中[...]意味着其可以是一个列表。这些传递给MapReduce进行处理的数据存储于分布式文件上，mapper操作将应用于每一个传递过来的键-值对并生成一定数量的中间键值对(intermediate key-value)，而后reduce操作将应用于这些中间键值对并输出最终的键值对。然而，mapper操作和reducer操作之间还隐含着一个应用于中间键值对的“分组”操作，同一个键的键值对需要被归类至同一组中并发送至同一个reducer，而传送给每个reducer的分组中的键值对是基于键进行排序后的列表。reducer生成的结果将会保存至分布式文件系统，并存储为一个或多个以r(即reducer号码)结尾的文件，但mapper生成的中间键值对数据则不会被保存。 在Hadoop中，mapper和reducer是分别由MAP和REDUCE方法实现的对象。每个map任务(接收一个称作input split的键值对列表)都被初始化一个mapper对象，并会由执行框架为每个输入的键值对调用一次其map方法。程序员可以配置启动的map任务个数，但其真正启动的数目则由执行框架根据数据的物理分布最终给定。类似地，每个reduce任务由REDUCE方法初始化为一个reduce对象，并会由执行框架为其接受的每个中间键值对调用一次REDUCE方法，所不同的是，程序员可以明确限定启动的reduce任务的个数。 mapper和reducer可以直接在各自接收的数据上执行所需要的操作，然而，当使用到外部资源时，多个mapper或reducer之间可能会产生资源竞争，这势必导致其性能下降，因此，程序员必须关注其所用资源的竞争条件并加入适当处理。其次，mapper输出的中间键值对与接受的键值对可以是不同的数据类型，类似地，reducer输出的键值对与其接收的中间键值对也可以是不同的数据类型，这可能会给编程过程及程序运行中的故障排除带来困难，但这也正是MapReduce强大功能的体现之一。 除了常规的两阶段MapReduce处理流外，其还有一些变化形式。比如将mapper输出的结果直接保存至磁盘中(每个mapper对应一个文件)的没有reducer的MapReduce作业，不过仅有reducer而没有mapper的作业是不允许的。不过，就算用不着reducer处理具体的操作，利用reducer将mapper的输出结果进行重新分组和排序后进行输出也能以另一种形式提供的完整MapReduce模式。 MapReduce作业一般是通过HDFS读取和保存数据，但它也可以使用其它满足MapReduce应用的数据源或数据存储，比如Google的MapReduce实现中使用了Bigtable来完成数据的读入或输出。BigTable属于非关系的数据库，它是一个稀疏的、分布式的、持久化存储的多维度排序Map，其设计目的是可靠的处理PB级别的数据，并且能够部署到上千台机器上。在Hadoop中有一个类似的实现HBase可用于为MapReduce提供数据源和数据存储。 1.5 Hadoop运行框架 MapReduce程序也称作为MapReduce作业，一般由mapper代码、reducer代码以及其配置参数(如从哪儿读入数据，以及输出数据的保存位置)组成。准备好的作业可通过JobTracker(作业提交节点)进行提交，然后由运行框架负责完成后续的其它任务。这些任务主要包括如下几个方面。 (1) 调度 每个MapReduce作业都会划分为多个称作任务(task)的较小单元，而较大的作业划分的任务数量也可能会超出整个集群可运行的任务数，此时就需要调度器程序维护一个任务队列并能够追踪正在运行态任务的相关进程，以便让队列中处于等待状态的任务派送至某转为可用状态的节点运行。此外，调度器还要负责分属于不同作业的任务协调工作。 对于一个运行中的作业来说，只有所用的map任务都完成以后才能将中间数据分组、排序后发往reduce作业，因此，map阶段的完成时间取决于其最慢的一个作业的完成时间。类似的，reduce阶段的最后一个任务执行结束，其最终结果才为可用。因此，MapReduce作业完成速度则由两个阶段各自任务中的掉队者决定，最坏的情况下，这可能会导致作业长时间得不到完成。出于优化执行的角度，Hadoop和Google MapReduce实现了推测执行(Speculative execution)机制，即同一个任务会在不同的主机上启动多个执行副本，运行框架从其最快执行的任务中取得返回结果。不过，推测执行并不能消除其它的滞后场景，比如中间键值对数据的分发速度等。 (2) 数据和代码的协同工作(data/code co-location) 术语“数据分布”可能会带来误导，因为MapReduce尽力保证的机制是将要执行的代码送至数据所在的节点执行，因为代码的数据量通常要远小于要处理的数据本身。当然，MapReduce并不能消除数据传送，比如在某任务要处理的数据所在的节点已经启动很多任务时，此任务将不得不在其它可用节点运行。此时，考虑到同一个机架内的服务器有着较充裕的网络带宽，一个较优选择是从数据节点同一个机架内挑选一个节点来执行此任务。 (3) 同步(Synchronization) 异步环境下的一组并发进程因直接制约而互相发送消息而进行互相合作、互相等待，使得各进程按一定的速度执行的过程称为进程间同步，其可分为进程同步(或者线程同步)和数据同步。就编程方法来说，保持进程间同步的主要方法有内存屏障(Memory barrier)，互斥锁（Mutex），信号量(Semaphore)和锁(Lock)，管程(Monitor)，消息(Message)，管道(Pipe)等。MapReduce是通过在map阶段的进程与reduce阶段的进程之间实施隔离来完成进程同步的，即map阶段的所有任务都完成后对其产生的中间键值对根据键完成分组、排序后通过网络发往各reducer方可开始reduce阶段的任务，因此这个过程也称为“shuffle and sort”。 (4) 错误和故障处理(Error and fault handling) MapReduce运行框架本身就是设计用来容易发生故障的商用服务器上了，因此，其必须有着良好的容错能力。在任何类别的硬件故障发生时，MapReduce运行框架均可自行将运行在相关节点的任务在一个新挑选出的节点上重新启动。同样，在任何程序发生故障时，运行框架也要能够捕获异常、记录异常并自动完成从异常中恢复。另外，在一个较大规模的集群中，其它任何超出程序员理解能力的故障发生时，MapReduce运行框架也要能够安全挺过。 1.6 partitioner和combiner 除了前述的内容中的组成部分，MapReduce还有着另外两个组件：partiontioner和combiner。 Partitioner负责分割中间键值对数据的键空间(即前面所谓的“分组”)，并将中间分割后的中间键值对发往对应的reducer，也即partitioner负责完成为一个中间键值对指派一个reducer。最简单的partitioner实现是将键的hash码对reducer进行取余计算，并将其发往余数对应编号的reducer，这可以尽力保证每个reducer得到的键值对数目大体上是相同的。不过，由于partitioner仅考虑键而不考虑“值”，因此，发往每个reducer的键值对在键数目上的近似未必意味着数据量的近似。 Combiner是MapReduce的一种优化机制，它的主要功能是在“shuffle and sort”之前先在本地将中间键值对进行聚合，以减少在网络上发送的中间键值对数据量。因此可以把combiner视作在“shuffle and sort”阶段之前对mapper的输出结果所进行聚合操作的“mini-reducer”。在实现中，各combiner之间的操作是隔离的，因此，它不会涉及到其它mapper的数据结果。需要注意的是，就算是某combiner可以有机会处理某键相关的所有中间数据，也不能将其视作reducer的替代品，因为combiner输出的键值对类型必须要与mapper输出的键值对类型相同。无论如何，combiner的恰当应用将有机会有效提高作业的性能。 二、分布式文件系统(DFS) 前面的讨论中，我们已经了解了Hadoop中实现的MapReduce是一个编程模型和运行框架，它能够通过JobTracker接收客户提交的作业而后将其分割为多个任务后并行运行在多个TaskTracker上。而问题是，这些TaskTracker如何高效获取所要处理的数据？ 在传统的高性能集群中，计算节点和存储节点是各自独立的，它们之间通过高速网络完成互联，然而，在面临海量数据处理的问题时，网络必然会成为整个系统的性能瓶颈，这就需要引入超高速的网络如万兆以太网或Infiniband。然而，对大数场景来讲它们属于“奢侈品”，且昂贵的投入并不能带来网络性能的线性提升，因此性价比不高。面对这种问题，MapReduce采取了将计算节点与存储节点合二为一的集群模型，它利用分布式文件系统将数据存储于多个节点上，而后让处理过程在各数据节点本地直接进行，从而极大地降低了数据通过网络传送的需求。不过，这里仍然需要说明的是，MapReduce并非依赖于分布式文件系统，只不过运行在非分布式文件系统的MapReduce的诸多高级特性将无用武之地。 事实上，分布式文件系统并非MapReduce带来的新生事物，只不过，MapReduce站在前人的基础上将分布式文件系统进行了改造以使得它更能够适用于在MapReduce中完成海量数据处理。Google为在他们的MapReduce中实现的分布式文件系统为GFS(Google File System)，而Hadoop的实现称作HDFS(Hadoop Distributed File System)。 (责任编辑：liangzh) |