hadoop的mapreduce常见算法案例有几种

Posted 2023-03-28

基本MapReduce模式

计数与求和
问题陈述:
有许多文档，每个文档都有一些字段组成。需要计算出每个字段在所有文档中的出现次数或者这些字段的其他什么统计值。例如，给定一个log文件，其中的每条记录都包含一个响应时间，需要计算出平均响应时间。
解决方案:
让我们先从简单的例子入手。在下面的代码片段里，Mapper每遇到指定词就把频次记1，Reducer一个个遍历这些词的集合然后把他们的频次加和。

1 class Mapper
2 method Map(docid id, doc d)
3 for all term t in doc d do
4 Emit(term t, count 1)
5
6 class Reducer
7 method Reduce(term t, counts [c1, c2,...])
8 sum = 0
9 for all count c in [c1, c2,...] do
10 sum = sum + c
11 Emit(term t, count sum)

这种方法的缺点显而易见，Mapper提交了太多无意义的计数。它完全可以通过先对每个文档中的词进行计数从而减少传递给Reducer的数据量:

1 class Mapper
2 method Map(docid id, doc d)
3 H = new AssociativeArray
4 for all term t in doc d do
5 Ht = Ht + 1
6 for all term t in H do
7 Emit(term t, count Ht)

如果要累计计数的的不只是单个文档中的内容，还包括了一个Mapper节点处理的所有文档，那就要用到Combiner了:

1 class Mapper
2 method Map(docid id, doc d)
3 for all term t in doc d do
4 Emit(term t, count 1)
5
6 class Combiner
7 method Combine(term t, [c1, c2,...])
8 sum = 0
9 for all count c in [c1, c2,...] do
10 sum = sum + c
11 Emit(term t, count sum)
12
13 class Reducer
14 method Reduce(term t, counts [c1, c2,...])
15 sum = 0
16 for all count c in [c1, c2,...] do
17 sum = sum + c
18 Emit(term t, count sum)

应用：Log 分析, 数据查询

整理归类

问题陈述:
有一系列条目，每个条目都有几个属性，要把具有同一属性值的条目都保存在一个文件里，或者把条目按照属性值分组。 最典型的应用是倒排索引。
解决方案：
解决方案很简单。 在 Mapper 中以每个条目的所需属性值作为 key，其本身作为值传递给 Reducer。 Reducer 取得按照属性值分组的条目，然后可以处理或者保存。如果是在构建倒排索引，那么 每个条目相当于一个词而属性值就是词所在的文档ID。
应用：倒排索引， ETL
过滤 (文本查找)，解析和校验
问题陈述:
假设有很多条记录，需要从其中找出满足某个条件的所有记录，或者将每条记录传换成另外一种形式（转换操作相对于各条记录独立，即对一条记录的操作与其他记录无关）。像文本解析、特定值抽取、格式转换等都属于后一种用例。
解决方案:
非常简单，在Mapper 里逐条进行操作，输出需要的值或转换后的形式。
应用：日志分析，数据查询，ETL，数据校验

分布式任务执行

问题陈述:
大型计算可以分解为多个部分分别进行然后合并各个计算的结果以获得最终结果。
解决方案: 将数据切分成多份作为每个 Mapper 的输入，每个Mapper处理一份数据，执行同样的运算，产生结果，Reducer把多个Mapper的结果组合成一个。
案例研究： 数字通信系统模拟
像 WiMAX 这样的数字通信模拟软件通过系统模型来传输大量的随机数据，然后计算传输中的错误几率。 每个 Mapper 处理样本 1/N 的数据，计算出这部分数据的错误率，然后在 Reducer 里计算平均错误率。
应用：工程模拟，数字分析，性能测试
排序
问题陈述:
有许多条记录，需要按照某种规则将所有记录排序或是按照顺序来处理记录。
解决方案: 简单排序很好办 – Mappers 将待排序的属性值为键，整条记录为值输出。 不过实际应用中的排序要更加巧妙一点， 这就是它之所以被称为MapReduce 核心的原因（“核心”是说排序？因为证明Hadoop计算能力的实验是大数据排序？还是说Hadoop的处理过程中对key排序的环节？）。在实践中，常用组合键来实现二次排序和分组。
MapReduce 最初只能够对键排序， 但是也有技术利用可以利用Hadoop 的特性来实现按值排序。想了解的话可以看这篇博客。
按照BigTable的概念，使用 MapReduce来对最初数据而非中间数据排序，也即保持数据的有序状态更有好处，必须注意这一点。换句话说，在数据插入时排序一次要比在每次查询数据的时候排序更高效。
应用：ETL，数据分析

非基本 MapReduce 模式

迭代消息传递 (图处理)

问题陈述：
假设一个实体网络，实体之间存在着关系。 需要按照与它比邻的其他实体的属性计算出一个状态。这个状态可以表现为它和其它节点之间的距离， 存在特定属性的邻接点的迹象， 邻域密度特征等等。
解决方案：
网络存储为系列节点的结合，每个节点包含有其所有邻接点ID的列表。按照这个概念，MapReduce 迭代进行，每次迭代中每个节点都发消息给它的邻接点。邻接点根据接收到的信息更新自己的状态。当满足了某些条件的时候迭代停止，如达到了最大迭代次数（网络半径）或两次连续的迭代几乎没有状态改变。从技术上来看，Mapper 以每个邻接点的ID为键发出信息，所有的信息都会按照接受节点分组，reducer 就能够重算各节点的状态然后更新那些状态改变了的节点。下面展示了这个算法：

1 class Mapper
2 method Map(id n, object N)
3 Emit(id n, object N)
4 for all id m in N.OutgoingRelations do
5 Emit(id m, message getMessage(N))
6
7 class Reducer
8 method Reduce(id m, [s1, s2,...])
9 M = null
10 messages = []
11 for all s in [s1, s2,...] do
12 if IsObject(s) then
13 M = s
14 else // s is a message
15 messages.add(s)
16 M.State = calculateState(messages)
17 Emit(id m, item M)

一个节点的状态可以迅速的沿着网络传全网，那些被感染了的节点又去感染它们的邻居，整个过程就像下面的图示一样：

案例研究： 沿分类树的有效性传递
问题陈述：
这个问题来自于真实的电子商务应用。将各种货物分类，这些类别可以组成一个树形结构，比较大的分类（像男人、女人、儿童）可以再分出小分类（像男裤或女装），直到不能再分为止（像男式蓝色牛仔裤）。这些不能再分的基层类别可以是有效（这个类别包含有货品）或者已无效的（没有属于这个分类的货品）。如果一个分类至少含有一个有效的子分类那么认为这个分类也是有效的。我们需要在已知一些基层分类有效的情况下找出分类树上所有有效的分类。
解决方案：
这个问题可以用上一节提到的框架来解决。我们咋下面定义了名为 getMessage和 calculateState 的方法：

1 class N
2 State in True = 2, False = 1, null = 0,
3 initialized 1 or 2 for end-of-line categories, 0 otherwise
4 method getMessage(object N)
5 return N.State
6 method calculateState(state s, data [d1, d2,...])
7 return max( [d1, d2,...] )

案例研究：广度优先搜索
问题陈述：需要计算出一个图结构中某一个节点到其它所有节点的距离。
解决方案： Source源节点给所有邻接点发出值为0的信号，邻接点把收到的信号再转发给自己的邻接点，每转发一次就对信号值加1：

1 class N
2 State is distance,
3 initialized 0 for source node, INFINITY for all other nodes
4 method getMessage(N)
5 return N.State + 1
6 method calculateState(state s, data [d1, d2,...])
7 min( [d1, d2,...] )

案例研究：网页排名和 Mapper 端数据聚合
这个算法由Google提出，使用权威的PageRank算法，通过连接到一个网页的其他网页来计算网页的相关性。真实算法是相当复杂的，但是核心思想是权重可以传播，也即通过一个节点的各联接节点的权重的均值来计算节点自身的权重。

1 class N
2 State is PageRank
3 method getMessage(object N)
4 return N.State / N.OutgoingRelations.size()
5 method calculateState(state s, data [d1, d2,...])
6 return ( sum([d1, d2,...]) )

要指出的是上面用一个数值来作为评分实际上是一种简化，在实际情况下，我们需要在Mapper端来进行聚合计算得出这个值。下面的代码片段展示了这个改变后的逻辑 （针对于 PageRank 算法）：

1 class Mapper
2 method Initialize
3 H = new AssociativeArray
4 method Map(id n, object N)
5 p = N.PageRank / N.OutgoingRelations.size()
6 Emit(id n, object N)
7 for all id m in N.OutgoingRelations do
8 Hm = Hm + p
9 method Close
10 for all id n in H do
11 Emit(id n, value Hn)
12
13 class Reducer
14 method Reduce(id m, [s1, s2,...])
15 M = null
16 p = 0
17 for all s in [s1, s2,...] do
18 if IsObject(s) then
19 M = s
20 else
21 p = p + s
22 M.PageRank = p
23 Emit(id m, item M)

应用：图分析，网页索引

值去重 （对唯一项计数）
问题陈述: 记录包含值域F和值域 G，要分别统计相同G值的记录中不同的F值的数目 (相当于按照 G分组).
这个问题可以推而广之应用于分面搜索（某些电子商务网站称之为Narrow Search）
Record 1: F=1, G=a, b
Record 2: F=2, G=a, d, e
Record 3: F=1, G=b
Record 4: F=3, G=a, b

Result:
a -> 3 // F=1, F=2, F=3
b -> 2 // F=1, F=3
d -> 1 // F=2
e -> 1 // F=2

解决方案 I:
第一种方法是分两个阶段来解决这个问题。第一阶段在Mapper中使用F和G组成一个复合值对，然后在Reducer中输出每个值对，目的是为了保证F值的唯一性。在第二阶段，再将值对按照G值来分组计算每组中的条目数。
第一阶段：

1 class Mapper
2 method Map(null, record [value f, categories [g1, g2,...]])
3 for all category g in [g1, g2,...]
4 Emit(record [g, f], count 1)
5
6 class Reducer
7 method Reduce(record [g, f], counts [n1, n2, ...])
8 Emit(record [g, f], null )

第二阶段：

1 class Mapper
2 method Map(record [f, g], null)
3 Emit(value g, count 1)
4
5 class Reducer
6 method Reduce(value g, counts [n1, n2,...])
7 Emit(value g, sum( [n1, n2,...] ) )

解决方案 II:
第二种方法只需要一次MapReduce 即可实现，但扩展性不强。算法很简单-Mapper 输出值和分类，在Reducer里为每个值对应的分类去重然后给每个所属的分类计数加1，最后再在Reducer结束后将所有计数加和。这种方法适用于只有有限个分类，而且拥有相同F值的记录不是很多的情况。例如网络日志处理和用户分类，用户的总数很多，但是每个用户的事件是有限的，以此分类得到的类别也是有限的。值得一提的是在这种模式下可以在数据传输到Reducer之前使用Combiner来去除分类的重复值。

1 class Mapper
2 method Map(null, record [value f, categories [g1, g2,...] )
3 for all category g in [g1, g2,...]
4 Emit(value f, category g)
5
6 class Reducer
7 method Initialize
8 H = new AssociativeArray : category -> count
9 method Reduce(value f, categories [g1, g2,...])
10 [g1', g2',..] = ExcludeDuplicates( [g1, g2,..] )
11 for all category g in [g1', g2',...]
12 Hg = Hg + 1
13 method Close
14 for all category g in H do
15 Emit(category g, count Hg)

应用：日志分析，用户计数
互相关
问题陈述：有多个各由若干项构成的组，计算项两两共同出现于一个组中的次数。假如项数是N，那么应该计算N*N。
这种情况常见于文本分析（条目是单词而元组是句子），市场分析（购买了此物的客户还可能购买什么）。如果N*N小到可以容纳于一台机器的内存，实现起来就比较简单了。
配对法
第一种方法是在Mapper中给所有条目配对，然后在Reducer中将同一条目对的计数加和。但这种做法也有缺点：
使用 combiners 带来的的好处有限，因为很可能所有项对都是唯一的
不能有效利用内存

1 class Mapper
2 method Map(null, items [i1, i2,...] )
3 for all item i in [i1, i2,...]
4 for all item j in [i1, i2,...]
5 Emit(pair [i j], count 1)
6
7 class Reducer
8 method Reduce(pair [i j], counts [c1, c2,...])
9 s = sum([c1, c2,...])
10 Emit(pair[i j], count s)

Stripes Approach（条方法？不知道这个名字怎么理解）
第二种方法是将数据按照pair中的第一项来分组，并维护一个关联数组，数组中存储的是所有关联项的计数。The second approach is to group data by the first item in pair and maintain an associative array (“stripe”) where counters for all adjacent items are accumulated. Reducer receives all stripes for leading item i, merges them, and emits the same result as in the Pairs approach.
中间结果的键数量相对较少，因此减少了排序消耗。
可以有效利用 combiners。
可在内存中执行，不过如果没有正确执行的话也会带来问题。
实现起来比较复杂。
一般来说， “stripes” 比 “pairs” 更快

1 class Mapper
2 method Map(null, items [i1, i2,...] )
3 for all item i in [i1, i2,...]
4 H = new AssociativeArray : item -> counter
5 for all item j in [i1, i2,...]
6 Hj = Hj + 1
7 Emit(item i, stripe H)
8
9 class Reducer
10 method Reduce(item i, stripes [H1, H2,...])
11 H = new AssociativeArray : item -> counter
12 H = merge-sum( [H1, H2,...] )
13 for all item j in H.keys()
14 Emit(pair [i j], Hj)

应用：文本分析，市场分析
参考资料：Lin J. Dyer C. Hirst G. Data Intensive Processing MapReduce
用MapReduce 表达关系模式
在这部分我们会讨论一下怎么使用MapReduce来进行主要的关系操作。
筛选（Selection）

1 class Mapper
2 method Map(rowkey key, tuple t)
3 if t satisfies the predicate
4 Emit(tuple t, null)

投影（Projection）
投影只比筛选稍微复杂一点，在这种情况下我们可以用Reducer来消除可能的重复值。

1 class Mapper
2 method Map(rowkey key, tuple t)
3 tuple g = project(t) // extract required fields to tuple g
4 Emit(tuple g, null)
5
6 class Reducer 参考技术A

可以只用一行代码来运行MapReduce作业：JobClient.runJon(conf)，Job作业运行时参与的四个实体：

     1.JobClient 写代码，配置作业，提交作业。

     2.JobTracker:初始化作业，分配作业，协调作业运行。这是一个java程序，主类是JobTracker。

     3.TaskTracker：运行作业划分后的任务，即分配数据分配上执行Map或Reduce任务。

     4.HDFS：保存作业数据、配置信息等，保存作业结果。

Map/Reduce 作业总体执行流程：

     代码编写 ----> 作业配置  ---->  作业提交 ----> Map任务分配和执行 ----> 处理中间结果 ---->  Reduce任务分配与执行 ---->  输出结果

而对于每个作业的执行，又包含：

     输入准备 ----> 任务执行 ----> 输出结果

作业提交JobClient：

     JobClient的runJob方法产生一个Jobclient实例并调用其submitJob方法，然后runJob开始循环吗，并在循环中调用getTaskCompetionEvents方法，获得TaskCompletionEvent实例，每秒轮询作业进度（后面有介绍进度和状态更新），把进度写到控制台，作业完成后显示作业计数器，若失败，则把错误记录到控制台。

     submitJob方法作业提交的过程：

     1.向JobTracker请求一个新的JobId。

     2.检查作业相关路径，如果路径不正确就会返回错误。

     3.计算作业输入分片及其划分信息。

     4.将作业运行需要的资源（jar文件、配置文件等）复制到Shared HDFS，并

复制多个副本（参数控制，默认值为10）供tasktracker访问，也会将计算的分片复制到HDFS。

     5.调用JobTracker对象的submitJob()方法来真正提交作业，告诉JobTracker作业准备执行。

作业的初始化JobTracker：

     JobTracker收到submitJob方法调用后，会把调用放入到一个内部队列，由作业调度器（Job scheduler）进行调度并对其初始化。Job初始化即创建一个作业对象。

     当作业被调度后，JobTracker会创建一个代表这个作业的JobInProgress对象，并将任务和记录信息封装在这个对象中，以便跟踪任务状态和进程。

     初始化过程就是JobInProgress对象的initTasks方法进行初始化的。

     初始化步骤：

          1.从HDFS中读取作业对应的job.split信息，为后面的初始化做好准备。

          2.创建并初始化map和reduce任务。根据数据分片信息中的个数确定map task的个数，然后为每个map task生成一个TaskInProgress对象来处理数据分片，先将其放入nonRunningMapCache,以便JobTracker分配任务的时候使用。接下来根据JobConf中的mapred.reduce.tasks属性利用setNumReduceTasks()方法设置reduce task的数量，然后同map task创建方式。

          3.最后就是创建两个初始化task，进行map和reduce的初始化。

任务的分配JobTracker：

    消息传递HeartBeat： tasktracker运行一个简单循环定期发送心跳（heartbeat）给JobTracker。由心跳告知JobTracker自己是否存活，同时作为消息通道传递其它信息（请求新task）。作为心跳的一部分，tasktracker会指明自己是否已准备好运行新的任务，如果是，jobtracker会分配它一个任务。

    分配任务所属于的作业：在Jobtracker分配任务前需先确定任务所在的作业。后面会介绍到各种作业调度算法，默认是一个FIFO的作业调度。

    分配Map和Reduce任务：tasktracker有固定数量的任务槽,一个tasktracker可以同时运行多个Map和Reduce任务，但其准确的数量由tasktracker的核的数量和内存大小决定。默认调度器会先填满Map任务槽，再填Reduce任务槽。jobtracker会选择距离离分片文件最近的tasktracker，最理想情况下，任务是数据本地化（data-local）的,当然也可以是机架本地化（rack-local），如果不是本地化的，那么他们就需要从其他机架上检索数据。Reduce任务分配很简单，jobtracker会简单的从待运行的reduce任务列表中选取下一个来执行，不用考虑数据本地化。

任务的执行TaskTracker：

     TaskTracker收到新任务后，就要在本地运行任务了，运行任务的第一步就是通过localizedJob将任务本地化所需要的注入配置、数据、程序等信息进行本地化。

     1.本地化数据：从共享文件系统将job.split 、job.jar (在分布式缓存中)复制本地，将job配置信息写入job.xml。

     2.新建本地工作目录：tasktracker会加压job.jar文件到本工作目录。

     3.调用launchTaskForJob方法发布任务（其中会新建TaskRunner实例运行任务），如果是Map任务就启用MapTaskRunner，对于Reduce就是ReduceTaskRunner。

   

  在这之后，TaskRunner会启用一个新的JVM来运行每个Map/Reduce任务，防止程序原因而导致tasktracker崩溃，但不同任务间重用JVM还是可以的，后续会讲到任务JVM重用。

     对于单个Map，任务执行的简单流程是：

     1.分配任务执行参数

     2.在Child临时文件中添加map任务信息（Child是运行Map和Reduce任务的主进程）

     3.配置log文件夹，配置map任务的通信和输出参数

     4.读取input split，生成RecordReader读取数据

     5.为Map生成MapRunnable,依次从RecordReader中接收数据，并调用Map函数进行处理。

     6.最后将map函数的输出调用collect收集到MapOutputBuffer（参数控制其大小）中。

Streaming和Pipes:

     Streaming和Pipes都运行特殊的Map和Reduce任务，目的是运行用户提供的可执行程序并与之通信。

     Streaming:使用标准输入输出Streaming与进程进行通信。

     Pipes:用来监听套接字，会发送一个端口号给C++程序，两者便可建立链接。

     

进度和状态更新：

     一个作业和它的任务都有状态（status），其中包括：运行成功失败状态、Map/Reduce进度、作业计数器值、状态消息。

     状态消息与客户端的通信：

     1.对于Map任务Progress的追踪：progress是已经处理完的输入所占的比例。

     2.对于Reduce：稍复杂，reduce任务分三个阶段（每个阶段占1/3），复制、排序和Reduce处理，若reduce已执行一半的输入的话，那么任务进度便是1/3+1/3+1/6=5/6。

     3.任务计数器：任务有一组计数器，负责对任务运行各个事件进行计数。

     4.任务进度报告：如果任务报告了进度，便会设置一个标记以表明状态将被发送到tasktracker。有一个独立线程每隔三秒检查一次此标记，如果已设置，则告知tasktracker当前状态。

     5.tasktracker进度报告：tasktracker会每隔5秒（这个心跳是由集群大小决定，集群越大时间会越长）发送heartbeat到jobtracker，并且tasktracker运行的所有状态都会在调用中被发送到jobtracker。

     6.jobtracker合并各任务报告：产生一个表明所有运行作业机器所含任务状态的全局视图。

     前面提到的JobClient就是通过每秒查询JobTracker来接收最新状态，而且客户端JobClient的getJob方法可以得到一个RunningJob的实例，其包含了作业的所以状态信息。

     

作业的完成：

     当jobtracker收到作业最后一个任务已完成的通知后，便把作业状态设置成成功。JobClient查询状态时，便知道任务已成功完成，于是JobClient打印一条消息告知用户，然后从runJob方法返回。

     如果jobtracker有相应设置，也会发送一个Http作业通知给客户端，希望收到回调指令的客户端可以通过job.end.notification.url属性来进行设置。

     jobtracker情况作业的工作状态，指示tasktracker也清空作业的工作状态，如删除中间输出。

     

失败

     实际情况下，用户的代码存在软件错误进程会崩溃，机器也会产生故障，但Hadoop能很好的应对这些故障并完成作业。

     1.任务失败    

     子任务异常：如Map/Reduce任务中的用户代码抛出异常，子任务JVM进程会在退出前向父进程tasktracker发送错误报告，错误被记录用户日志。tasktracker会将此次task attempt标记为tailed，并释放这个任务槽运行另外一个任务。

     子进程JVM突然退出：可能由于JVM bug导致用户代码造成的某些特殊原因导致JVM退出，这种情况下，tasktracker会注意到进程已经退出，并将此次尝试标记为failed。

     任务挂起：一旦tasktracker注意一段时间没有收到进度更新，便会将任务标记为failed，JVM子进程将被自动杀死。任务失败间隔时间通常为10分钟，可以以作业或者集群为基础设置过期时间，参数为mapred.task.timeout。注意：如果参数值设置为0，则挂起的任务永远不会释放掉它的任务槽，随着时间的推移会降低整个集群的效率。

     任务失败尝试次数：jobtracker得知一个tasktracker失败后，它会重新调度该任务执行，当然，jobtracker会尝试避免重新调度失败过的tasktracker任务。如果一个任务尝试次数超过4次，它将不再被重试。这个值是可以设置的，对于Map任务，参数是mapred.map.max.attempts,对于reduce任务，则由mapred.reduce.max.attempts属性控制。如果次数超过限制，整个作业都会失败。当然，有时我们不希望少数几个任务失败就终止运行的整个作业，因为即使有些任务失败，作业的一些结果可能还是有用的，这种情况下，可以为作业设置在不触发作业失败情况下的允许任务失败的最大百分比，Map任务和Reduce任务可以独立控制，参数为mapred.max.map.failures.percent 和mapred.max.reduce.failures.percent。

     任务尝试中止（kill）：任务终止和任务失败不同，task attempt可以中止是因为他是一个推测副本或因为它所处的tasktracker失败，导致jobtracker将它上面的所有task attempt标记为killed。被终止的task attempt不会被计入任务运行尝试次数，因为尝试中止并不是任务的错。

     2.tasktracker失败

     tasktracker由于崩溃或者运行过慢而失败，他将停止向jobtracker发送心跳（或很少发送心跳）。jobtracker注意已停止发送心跳的tasktracker（过期时间由参数mapred.tasktracker.expiry.interval设置，单位毫秒），并将它从等待调度的tasktracker池中移除。如果是未完成的作业，jobtracker会安排次tasktracker上已经运行成功的Map任务重新运行，因为此时reduce任务已无法访问（中间输出存放在失败的tasktracker的本地文件系统上）。

     即使tasktracker没有失败，也有可能被jobtracker列入黑名单。如果tasktracker上面的失败任务数量远远高于集群的平均失败任务次数，他就会被列入黑名单，被列入黑名单的tasktracker可以通过重启从jobtracker黑名单中移除。

     3.jobtracker失败

     老版本的JobTracker失败属于单点故障，这种情况下作业注定失败。

作业调度：

     早期作业调度FIFO：按作业提交顺序先进先出。可以设置优先级，通过设置mapred.job.priority属性或者JobClient的setJobPriority()方法制定优先级（优先级别：VERY_HIGH,HIGH,NORMAL,LOW,VERY_LOW）。注意FIFO调度算法不支持抢占（preemption）,所以高优先级作业仍然会被那些已经开始的长时间运行的低优先级作业所阻塞。

     Fair Scheduler：目标是让每个用户公平地共享集群能力。当集群存在很多作业时，空闲的任务槽会以”让每个用户共享集群“的方式进行分配。默认每个用户都有自己的作业池。FairScheduler支持抢占，所以，如果一个池在特定的一段时间未得到公平地资源共享，它会终止池中得到过多的资源任务，以便把任务槽让给资源不足的池。FairScheduler是一个后续模块，使用它需要将其jar文件放在Hadoop的类路径下。可以通过参数map.red.jobtracker.taskScheduler属性配置（值为org.apache.hadoop.mapred.FairScheduler）

     Capacity Scheduler：

     集群由很多队列组成，每个队列都有一个分配能力，这一点与FairScheduler类似，只不过在每个队列内部，作业根据FIFO方式进行调度。本质上说，Capacity Scheduler允许用户或组织为每个用户模拟一个独立使用FIFO的集群。

shuffle和排序：

     MapReduce确保每个Reducer的输入都是按键排序的。系统执行排序的过程-将map输出作为输入传给reducer的过程称为shuffle。shuffle属于不断被优化和改进的代码库的一部分，从许多方面来看，shuffle是MapReduce的心脏。

     整个shuffle的流程应该是这样：

     map结果划分partition  排序sort 分割spill   合并同一划分   合并同一划分  合并结果排序 reduce处理 输出

     Map端：

     写入缓冲区：Map函数的输出，是由collector处理的，它并不是简单的将结果写到磁盘。它利用缓冲的方式写到内存，并处于效率的考虑进行预排序。每个map都有一个环形的内存缓冲区，用于任务输出，默认缓冲区大小为100MB（由参数io.sort.mb调整），一旦缓冲区内容达到阈值（默认0.8），后台进程边开始把内容写到磁盘（spill），在写磁盘过程中，map输出继续被写到缓冲区，但如果缓冲区被填满，map会阻塞知道写磁盘过程完成。写磁盘将按照轮询方式写到mapred.local.dir属性制定的作业特定子目录中。

     写出缓冲区：collect将缓冲区的内容写出时，会调用sortAndSpill函数，这个函数作用主要是创建spill文件，按照key值对数据进行排序，按照划分将数据写入文件，如果配置了combiner类，会先调用combineAndSpill函数再写文件。sortAndSpill每被调用一次，就会写一个spill文件。

     合并所有Map的spill文件：TaskTracker会在每个map任务结束后对所有map产生的spill文件进行merge，merge规则是根据分区将各个spill文件中数据同一分区中的数据合并在一起，并写入到一个已分区且排序的map输出文件中。待唯一的已分区且已排序的map输出文件写入最后一条记录后，map端的shuffle阶段就结束了。

     在写磁盘前，线程首先根据数据最终要传递到的reducer把数据划分成响应的分区（partition），在每个分区中，后台线程按键进行内排序，如果有一个combiner,它会在排序后的输出上运行。

     内存达到溢出写的阈值时，就会新建一个溢出写文件，因为map任务完成其最后一个输出记录之后，会有几个溢出写文件。在任务完成前，溢出写文件会被合并成一个已分区且已排序的输出文件。配置属性io.sort.facor控制一次最多能合并多少流，默认值是10。

     如果已经指定combiner,并且写次数至少为3（通过min.mum.spills.for.combine设置）时，则combiner就会在输出文件写到磁盘之前运行。运行combiner的意义在于使map输出更紧凑，舍得写到本地磁盘和传给reducer的数据更少。

     写磁盘时压缩：写磁盘时压缩会让写的速度更快，节约磁盘空间，并且减少传给reducer的数据量。默认情况下，输出是不压缩的，但可以通过设置mapred.compress.map.output值为true，就可以启用压缩。使用的压缩库是由mapred.map.output.compression.codec制定。

     reducer获得文件分区的工作线程：reducer通过http方式得到输出文件的分区，用于文件分区的工作线程数量由tracker.http.threads属性指定，此设置针对的是每个tasktracker，而不是每个map任务槽。默认值为40，在大型集群上此值可以根据需要而增加。

     Reduce端：

     复制阶段：reduce会定期向JobTracker获取map的输出位置，一旦拿到输出位置，reduce就会从对应的TaskTracker上复制map输出到本地（如果map输出很小，则会被复制到TaskTracker节点的内存中，否则会被让如磁盘），而不会等到所有map任务结束（当然这个也有参数控制）。

     合并阶段：从各个TaskTracker上复制的map输出文件（无论在磁盘还是内存）进行整合，并维持数据原来的顺序。

     Reduce阶段：从合并的文件中顺序拿出一条数据进行reduce函数处理，然后将结果输出到本地HDFS。

     Map的输出文件位于运行map任务的tasktracker的本地磁盘，现在，tasktracker要为分区文件运行reduce任务。每个任务完成时间可能不同，但是只要有一个任务完成，reduce任务就开始复制其输出，这就是reduce任务的复制阶段（copy phase）。reduce任务有少量复制线程，因此能够并行取得map输出。默认值是5个线程，可以通过mapred.reduce.parallel.copies属性设置。

     Reducer如何得知从哪个tasktracker获得map输出：map任务完成后会通知其父tasktracker状态已更新，tasktracker进而通知（通过heart beat）jobtracker。因此，JobTracker就知道map输出和tasktracker之间的映射关系，reducer中的一个线程定期询问jobtracker以便获知map输出位置。由于reducer有可能失败，因此tasktracker并没有在第一个reducer检索到map输出时就立即从磁盘上删除它们，相反他会等待jobtracker告示它可以删除map输出时才删除，这是作业完成后最后执行的。

     如果map输出很小，则会被直接复制到reduce tasktracker的内存缓冲区（大小由mapred.job.shuffle.input.buffer.percent控制，占堆空间的百分比），否则，map输出被复制到磁盘。一旦内存缓冲区达到阈值大小（由mapred.iob.shuffle.merge.percent）

或达到map输出阈值大小（mapred.inmem.threadhold），则合并后溢出写到磁盘中。

     随着磁盘上副本增多，后台线程会将他们合并为更大的、排好序的文件。注意：为了合并，压缩的map输出必须在内存中被解压缩。

     排序阶段：复制阶段完成后，reduce任务会进入排序阶段，更确切的说是合并阶段，这个阶段将合并map输出，维持其顺序排列。合并是循环进行的，由合并因子决定每次合并的输出文件数量。但让有可能会产生中间文件。

     reduce阶段：在最后reduce阶段，会直接把排序好的文件输入reduce函数，不会对中间文件进行再合并，最后的合并即可来自内存，也可来自磁盘。此阶段的输出会直接写到文件系统，一般为hdfs。

     细节：这里合并是并非平均合并，比如有40个文件，合并因子为10，我们并不是每趟合并10个，合并四趟。而是第一趟合并4个，后三趟合并10，在最后一趟中4个已合并的文件和余下6个未合并会直接并入reduce。