ma-大数据mapreduce思想和数据切割

  

 

一、MapReduce理论基础

 

1.1 理论基础

 

每个MapReduce job都是Hadoop客户端想要执行的一个工作单元，它一般由输入数据、MapReduce程序和配置信息组成，
而Hadoop会把每个job分隔成两类任务(task)：map任务和reduce任务。在Hadoop集群中有两类节点来执行两类job进程的执行

 

 

1.2 MapReduce和大数据问题

 

海量数据处理的核心思想无非是将一个较大的问题进行“分割包围、逐个歼灭”。
然而其难点和关键点在于如何将一个大的问题分分割成多个可以分别在不同的CPU上或不同的主机上进行处理的独立小问题，
而且这些独立进行处理的小问题所产生的中间结果又该如何合并成最终结果并予以输出。因此，看似简单的化整为零的处理思想却不得不面临如下的难题：
(1) 如何将大问题分割为小任务？进一步地，如何将大问题分解为可以并行处理的小任务？
(2) 如何将分解好的小任务派送给分布式系统中的某主机且是较为适合解决此问题的主机上的worker完成处理？
(3) 如何保证某worker获取所需的数据？
(4) 如何协调不同worker之间进行同步？
(5) 如何将某worker的部分结果共享给其它需要此结果的worker？
(6) 如何在出现软件或硬件故障时仍然能保证上述工作的顺利进行？

在传统的并行或分布式编程模型中，程序员不得不显式地解决上述的部分甚至是全部问题，而在共享内存编程中，
程序员需要显式地协调对共享数据结构的如互斥锁的访问、显式地通过栅(barrier)等设备解决进程同步问题、并得时刻警惕着程序中可能出现的死锁或竞争条件。
虽然有些编程语言也或多或少地规避了让程序员面对上述问题，但却也避免不了将资源分配给各worker的问题。MapReduce的优势之一便是有效地向程序员隐藏了这些问题。

 

 1.3 函数式编译语言

 

MapReduce是一种类似于Lisp或ML的函数式编程语言。函数式编程的核心特性之一是基于高阶函数，即能够接受其它函数作为参数的函数完成编程。
MapReduce有两个常见地内置高阶函数map和fold。


MapReduce有两个常见地内置高阶函数map和reduce，其map就类似于上述过程中的map操作，reduce对应于上述过程中的fold操作。
只不过，MapReduce的执行框架能自行协调map与reduce并将其应用于在商业服务器硬件平台上并行处理海量数据。

更为精确地说，MapReduce有三个相互关联却各不相同的概念。
首先，MapReduce是一个如上所述的函数式编程语言。
其次，MapReduce也是一个运行框架，它能够协调运行基于MapReduce思想开发的程序。
最后，MapReduce还可以被看作编程模型和执行框架的软件实现，如Google的专有实现和另一个开源实现Hadoop等。

简单来讲，用户开发的MapReduce程序其实就是包含了类似上述的f函数和g函数的程序，只不过，在MapReduce中，它们分别被称作mapper和reducer。

 

1.4 mapper和reducer

键值对儿(Key-value pair)是MapReduce的基础数据结构。Key和Value可以是基础类型数据，如整数、浮点数、字符串或未经加工的字节数据，也可以是任意形式的复杂数据类型。
程序员可以自行定义所需的数据类型，也可借助于Protocol Buffer、Thrift或Avro提供的便捷方式完成此类工作。

MapReduce算法设计的工作之一就是在给定数据集上定义“键-值”数据结构，比如在搜索引擎搜集、存储网页类工作中，key可以使用URL来表示，而value则是网页的内容。
而在有些算法中，Key也可以是没有任何实际意义的数据，其在数据处理过程中可被安全忽略。在MapReduce中，程序员需要基于如下方式定义mapper和reducer：
	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操作之间还隐含着一个应用于中间键值对的“分组”操作，同一个键的键值对需要被归类至同一组中并发送至同一个reducer，而传送给每个reducer的分组中的键值对是基于键进行排序后的列表。reducer生成的结果将会保存至分布式文件系统，并存储为一个或多个以r(即reducer号码)结尾的文件，但mapper生成的中间键值对数据则不会被保存。


MapReduce在大数据处理时，会根据要处理的数据文件及用户编写的map函数首先将数据分割为多个部分(split)，而后为每一个split启动一个map任务(map task，即map进程)，这些map任务由MapReduce运行环境调度着分散运行于集群中的一个或多个节点上；每个mapper执行结束后，都可能会输出许多的键值对，称作中间键值对，这些中间键值对临时性地存放在某位置，直到所有的mapper都执行结束；而后MapReduce把这些中间键值对重新进行分割为一个或多个分组，分组的标准是键相同的所有键值对都要排序后归入同一个组中，同一个组可以包含一个或多个键及其对应的数据，MapReduce运行环境会为每一个分组启动一个reduce任务 (reduce task)，这些reduce任务由MapReduce运行环境调度着运行于集群中的一个或多个节点上。

 

 上述一大堆话总结成如下结果：

1 mapreduce job
用户客户端提交一个请求 eg 单词计数
那么这个请求就是  mapreduce job

2 mapredcue job 分为两大类:
map task/map 任务
reduce task/reduce 任务

 

3 集群中执行 map任务基本是由集群自己决定的
  集群中执行 reduce任务可以人为设定

 

4 mapper:  个人人为map任务更多是资源层cpu 内存 IO方面
 map任务 + split(键值对列表) + map方法代码 = mapper

 

5 reducer:
 reduce任务 + 每一个map分组数据 + reduce方法代码 = reducer

 

6 看这个说法：
MapReduce在大数据处理时，会根据要处理的数据文件及用户编写的map函数首先将数据分割为多个部分(split)，
而后为每一个split启动一个map任务(map task，即map进程)，
这些map任务由MapReduce运行环境调度着分散运行于集群中的一个或多个节点上。

即：  1个mapper = 1个map task + 1个split + map代码
      1个map task = 1个map进程

约等效于： 1个mapper = 1个map task = 1个map进程

7 map分组数据：
      键相同的所有键值对都要排序后归入同一个组中
      同一个组可以包含一个或多个键及其对应的数据
      比如 hello {1,1,1}
           world {1,1} 可能会被分到同一个组中

 8 reduce最后的输出结果 part-r-00000中， r是reduce的缩写  00000表示第一个reducer输出的结果

    同理类推出如果有多个reducer 那么难结果会出现  part-r-00001 part-r-00002

 

 1.4 Hadoop运行框架 数据源切分 数据分组

 

1.4.1 客户端提交作业后的流程：

每一个需要由MapReduce框架运行的MapReduce程序称为一个MapReduce作业(MapReduce Job)，一般由mapper代码、reducer代码以及其配置参数(如从哪儿读入数据，以及输出数据的保存位置)组成。
准备好的作业由客户端提交，JobTracker接收，然后由运行框架负责完成后续的其它任务。其运行过程分为两个阶段，

map阶段和reduce阶段，每个阶段都根据作业本身(job)的属性,集群中可用资源，以及用户的配置去启动一定数量

的任务(task也即进程)来负责具体数据的处理, mapreduce集群中，

负责接收客户端提交作业的主机称为master节点，

此节点上具体负责接收作业的进程称为JobTracker,

负责运行map任务或reduce任务的节点称为slave节点，

其运行的作业处理进程为TaskTracker,

默认情况下，一个slave节点可同时运行两个map task(任务)和两个reduce task(任务)

如下是map-reduce架构图，可以看到每个tasktracker节点上最多是放了4个map/reduce task

 

这些任务(task)主要包括如下几个方面。 

 (1) 调度  

每个MapReduce作业(job)都会划分为多个称作任务(task)的较小单元，而较大的作业划分的任务数量也可能会超出整个集群可运行的任务数，
此时就需要调度器程序维护一个任务队列并能够追踪正在运行态任务的相关进程，以便让队列中处于等待状态的任务派送至某转为可用状态的节点运行。
此外，调度器还要负责分属于不同作业的任务协调工作。
对于一个运行中的作业来说，只有所用的map任务都完成以后才能将中间数据分组、排序后发往reduce作业，因此，map阶段的完成时间取决于其最慢的一个作业的完成时间。
类似的，reduce阶段的最后一个任务执行结束，其最终结果才为可用。

因此，MapReduce作业完成速度则由两个阶段各自任务中的掉队者决定，最坏的情况下，这可能会导致作业长时间得不到完成。
出于优化执行的角度，Hadoop和Google MapReduce实现了推测执行(Speculative execution)机制，即同一个任务会在不同的主机上启动多个执行副本，
运行框架从其最快执行的任务中取得返回结果。不过，推测执行并不能消除其它的滞后场景，比如中间键值对数据的分发速度等。

 总结一句话：

       一个mr job因为作业较大，会被mr框架切分成很多mr task,而集群中每个节点并行执行的mr task最多就是4个(slots) 因此需要mr框架调度器来统一协调,如果某个map调用失败，会在保存这个数据副本的节点上在执行同样的map任务，
    如果某个map或者reduce任务过慢，作为jobtracker来讲 很难确定到底哪个服务器性能如何  或者预计是这样但是到真实机器中运行就是这么慢
    只要有足够资源可用 ，jobtracker会尝试在多个节点上将这个过慢的任务在这个节点的副本上启动多个
    比如任务0在副本1节点上启动，同时在副本2节点上启动，同时在副本3节点上启动，谁先执行完 就用谁的结果

 

(2) 数据和代码的协同工作(data/code co-location)

术语“数据分布”可能会带来误导，因为MapReduce尽力保证的机制是将要执行的代码送至数据所在的节点执行，因为代码的数据量通常要远小于要处理的数据本身。
当然，MapReduce并不能消除数据传送，比如在某任务要处理的数据所在的节点已经启动很多任务时，此任务将不得不在其它可用节点运行。
此时，考虑到同一个机架内的服务器有着较充裕的网络带宽，一个较优选择是从数据节点同一个机架内挑选一个节点来执行此任务。

 这点就涉及到 mapreduce job对数据源切分的问题： 明确一点，源文件上传到hdfs后被切分存储时，

切分并非完全均匀，以如下为例，说明下 数据要移动的情况：

不考虑副本下， 源数据上传到hdfs后被分成10片，分别放在三个节点上，第一个节点存放6份， 后两节点分别存放2份
集群共5个节点。

执行单词计数任务，mr框架启动5个进程， 节点1启动1个进程，节点2,3各自启动一个进程，此时节点1中的6份数据  ，
会在将这6份数据分成3份，一份放在节点4中，一份放在节点5中，然后在节点4,5上分别启动map进程，
map-reduce框架上，并非每个数据都是在本地处理的，如果不是本地处理，那么数据复制是
不可避免的， 没有任何理想的局面，数据分割也必然不是理想的平均切分。

如下图：

 

 

(3) 同步(Synchronization)

异步环境下的一组并发进程因直接制约而互相发送消息而进行互相合作、互相等待，使得各进程按一定的速度执行的过程称为进程间同步，
其可分为进程同步(或者线程同步)和数据同步。就编程方法来说，保持进程间同步的主要方法有内存屏障(Memory barrier)，互斥锁（Mutex），信号量(Semaphore)和锁(Lock)，
管程(Monitor)，消息(Message)，管道(Pipe)等。
MapReduce是通过在map阶段的进程与reduce阶段的进程之间实施隔离来完成进程同步的，
即map阶段的所有任务都完成后对其产生的中间键值对根据键完成分组、排序后通过网络发往各reducer方可开始reduce阶段的任务，
因此这个过程也称为“shuffle and sort”。

 总结一句话：
 map进程结束后，把数据归并排序后(shuffle and sort)网络发送给reduce后 mr框架才启动reduce进程。

 

(4) 错误和故障处理(Error and fault handling)

MapReduce运行框架本身就是设计用来容易发生故障的商用服务器上了，因此，其必须有着良好的容错能力。
在任何类别的硬件故障发生时，MapReduce运行框架均可自行将运行在相关节点的任务在一个新挑选出的节点上重新启动。
同样，在任何程序发生故障时，运行框架也要能够捕获异常、记录异常并自动完成从异常中恢复。
另外，在一个较大规模的集群中，其它任何超出程序员理解能力的故障发生时，MapReduce运行框架也要能够安全挺过。

 

如下是 map-reduce处理数据的流程图：

 

 如下是客户端提交一个任务后的流程图：

 

 

1.4.2 partitioner和combiner：

 
除了前述的内容中的组成部分，MapReduce还有着另外两个组件：partiontioner和combiner。

Partitioner负责分割中间键值对数据的键空间(即前面所谓的“分组”)，并将中间分割后的中间键值对发往对应的reducer，
也即partitioner负责完成为一个中间键值对指派一个reducer。
最简单的partitioner实现是将键的hash码对reducer进行取余计算，并将其发往余数对应编号的reducer(这就说明了一个reduce里面可以有key1:组1  key3:组3的结果，因为他们的hash一致)
这可以尽力保证每个reducer得到的键值对数目大体上是相同的。不过，由于partitioner仅考虑键而不考虑“值”，
因此，发往每个reducer的键值对在键数目上的近似未必意味着数据量的近似。

Combiner是MapReduce的一种优化机制，它的主要功能是在“shuffle and sort”之前先在本地将中间键值对进行聚合，以减少在网络上发送的中间键值对数据量。
因此可以把combiner视作在“shuffle and sort”阶段之前对mapper的输出结果所进行聚合操作的“mini-reducer”。在实现中，各combiner之间的操作是隔离的，
因此，它不会涉及到其它mapper的数据结果。需要注意的是，就算是某combiner可以有机会处理某键相关的所有中间数据，也不能将其视作reducer的替代品，
因为combiner输出的键值对类型必须要与mapper输出的键值对类型相同。无论如何，combiner的恰当应用将有机会有效提高作业的性能。

 下图是partition：

 

 下图是partition + combiner(小reducer只能在同一个map内实现数据聚合)