MapReduce:Fair Scheduler前传

        Fair Scheduler是由Facebook贡献给Hadoop社区的一种task调度策略。Facebook推出它的目的是在生产环境中替换毫无特点的MapReduce默认Scheduler。

        Fair Scheduler的诞生源于加州大学Berkeley分校、Facebook与Yahoo的研究人员于2009年4月底发表的论文《Job Scheduling for Multi-User MapReduce Clusters》。他们在文章中不仅介绍了开发Fair Scheduler的动机，更重要的是，针对一些job效率的问题，他们开发了相应的解决方案，并做出详细的评测。在具体介绍Fair Scheduler之前，我感觉还是有必要对他们提出的两种技术做简单介绍，希望吸引更多的人也能关注MapReduce Scheduler的设计。对于具体细节有兴趣的朋友请参考他们的论文。

        Map task的执行效率对job运行时间有很大影响。这取决于map task的输入数据源是位于Task Tracker本地还是远端(处于本TT同一rack内的其它TT，或是另外rack的TT)。Map task的输入源离它越近就执行的越快。为了有效控制map task数据源对执行效率的影响，job tracker定义了map task输入数据的几种缓存级别(cache level) ：1表示数据就在TT本地(node-local)，2表示数据在TT的同一rack的其它TT内(rack-local)，大于2表示数据在其它rack内(off-rack)。当某个TT的心跳数据到达JT时，task调度程序为这个TT分派cache level为1的map task，这是最好的结果。像HDFS这样的集群环境最宝贵的也许就是网络流量，所以延迟调度的初衷就是让map task需要的数据尽可能在本地，避免网络资源的消耗。

        但MapReduce默认的FIFO调度策略不给力。当它为TT选择不到cache level为1或2的map task时，就会选择off-rack的map task来执行，这是一种效率很低的分派策略。一般我们不会为集群配置网络拓扑信息，但Facebook的研究性集群就有配置。那个集群有六百多个节点，如果调度程序分派很多这种跨网络的数据源请求，对集群job执行效率有很大影响。

        为了解决这个问题，研究小组提出的解决方案是：延迟调度(Delay Scheduling)。假设TT每隔3秒向JT发出心跳包，理论上3秒之内每个TT都会从JT走一圈；调度程序如果只分派node-local级别的map task，那么极有可能有一段时间每个TT都在执行这种node-local的map task，或许有些job的task可能暂时因不符合cache level而没机会执行，但3秒内还是有机会分派node-local的map task的。 如果某个job没有node-local级别的map task，那么就跳过此job继续在所有job中寻找，直到为TT找到一个node-local级别的map task。可能出现的异常情况有：1. 当前的TT还真没有存储任何task的输入数据，那么忽略此次心跳。2. 有些存储有task输入数据的节点因为某些原因，心跳包发送延迟。为了不能饿着那些很长时间没有分派map task的job，当它等待node-local执行的时间超过一定范围，就容许调度程序可以启动下一cache level的map task，如果等待时间再长的话，还可以启动任何的map task。比方说某个job的task在五秒内都没有启动一个node-local的map task，那么在下一个TT心跳到来时，就可以为此job分派一个rack-local的map task；如果十秒内也没有启动node-local的task，也可以容忍job分派off-rack的map task。因一个心跳周期，每个TT都会去请求JT，所以后面的情况理论上出现的概率不大。

        第一种方案是针对map task的优化，下一个方案是解决reduce task执行过程中遇到的问题。正常情况下job中启动一些map task后，就可以分派reduce task。假设一个job很大，每个reduce task都得等待100个map task执行完成，它才能做reduce的工作，那么reduce task就长期占据着reduce slot，从而影响这个slot为其它task服务。

        Reduce task的执行分为两步：复制(copy)和计算(compute)。复制阶段属于网络IO密集型，它需要跨网络从其它TT上拉取map的输出结果。计算阶段是CPU密集型工作，执行reduce方法，需要大量的CPU处理。我们在定义slot数量时，一般是按机器处理器的数量为依据，这不是什么标准，只是我们认为map 和reduce阶段都要处理大量的数据，希望每个处理器都能发挥最大的作用。这样reduce在执行之前，是浪费着大量的TT CPU处理能力，这些处理能力可以交付其它的task来运行。

        程序小组针对这个问题开发的解决方案是：复制和计算分离(Copy-Compute Splitting)。这种方案可有两种解读：一是把reduce task分拆成两种基本task，一种是copy task，另外一种compute task。当copy task将map输出数据抓取完成时通知compute task做reduce处理，它们之间可能需要共享内存。 二是增加reduce slot的数量。Reduce slot可有两种设置，一种是计算型slot(computing slot)，一种是普通reduce slot。计算型slot的数量还是依据CPU数量来定，而普通reduce slot则不受限制(可以受网络流量或网卡的限制)。尽管这些方案实现起来有些麻烦，但对slot的有效利用有很多的好处。

        上面的方案解决了reduce task执行分两步的问题。当然map task也有同样的问题存在，但解决方案如延迟调度策略，尽量让数据源在执行map task的TT本地，减少了网络流量，而专注于CPU工作。从这看来，两个方案的核心思想是一致的。
  
        这里我只是介绍性的描述了解决这两个问题的方案，细节部分请参考附件中的论文。一切祝好，下一篇我会细说Fair Scheduler，它更适合于你的集群。

        附件下载：Job Scheduling for Multi-User MapReduce Clusters

评论

1 楼 swordice 2011-06-22  

很不错的文章，最近正好做MapReduce Scheduler这一块，希望能跟您交流一下