Hadoop 稳定性与性能提升浅析

在2012 Hadoop与大数据技术大会上，阿里系的无影带来了题为“阿里Hadoop集群架构及服务体系”的演讲，他主要介绍了阿里集群的发展现状、服务模式和挑战、Hadoop版本特性等方面的内容。本文主要尝试解析该PPT（http://hbtc2012.hadooper.cn/subject/track1liangliyin1.pdf）中提到的Hadoop版本特性。

在PPT中，谈到了阿里曾经遇到的Hadoop稳定性方面的挑战，并给出了几个例子：

阿里稳定性方面的挑战：
<1>异常作业拖垮\拖慢集群
<2>异常大作业浪费slots
<3>部分Slave机器带宽被打满
<4>NameNode被单个作业拖慢
<5>近期案例
– 用户作业的Task中创建了大量线程
– 用户提交的作业配置了777777700个Reduce Task
– 用户作业的Task在slave机器上创建了1000w+临时文件

针对遇到的这些问题，他介绍了具体的一些方案，下面逐一进行解析。需要说明的是，以下改进点均来自阿里的PPT，我只是解析，且解析的内容在各个Hadoop书籍中均有介绍，比如《Hadoop权威指南》。PPT中讲到的特性，一部分已在现有Hadoop中得到支持，另外一些尚未解决，应该是阿里自己添加的。

Hadoop已经支持的功能有：

3（1）1），3（3）（4），4（2）2）4）。

Hadoop尚未支持的功能：

1（1），1（2），1（3），2（1），2（2），3（1）2），3（2）1）2），3（3）（5）（6），4（2）1），4（2）3）。

1. 调度器改造

（1）增加JobLevel概念，优先保证Level高的作业

当前Hadoop将作业分为了5个等级，分别是VERY_HIGH、HIGH、NORMAL、LOW、VERY_LOW，用户提交作业时可通过参数“mapred.job.priority”设置作业优先级。如果你的Hadoop集群规模足够大或者作业足够多，你会发现这几个优先级难以区分作业的重要程度，为此，一种可行的解决方案是增加作业的优先级数目，比如增加到120个（像Linux Kernal那样），其中，最重要最紧急的作业优先级区间为90~120，次紧急的是60~90……

（2）支持异构操作系统调度，如RTEL 4u7,5u6

由于Hadoop是采用Java语言编写的，因此，Hadoop与平台无关的。但此处为什么提到支持异构操作系统调度呢？原因可能是，有些应用程序需要特殊操作系统上的API，为了能够让任务成功运行，必须将之调度到安装了对应操作系统的节点上。为了实现这个功能，需要让每个TaskTracker启动时（或者通过心跳，不过这样太冗余了）将自己的操作系统类型汇报给JobTracker，以便调度时用。

（3）服务端控制Reduce启动时机（slow start）

我们知道，对于一个MapReduce作业，它的Reduce Task依赖于Map Task的中间输出结果，每个Reduce Task要从每个Map Task中读取部分数据。选择合适的Reduce Task启动时机对于优化集群资源利用率十分必要，如果Reduce Task启动过早，会造成reduce slot资源利用率过低，如果启动过晚，会使得Map Task和Reduce Task串行执行，增加了作业运行时间。为此，通常待Map Task完成数目超过一定比例（用户可通过参数mapred.reduce.slowstart.completed.maps配置，默认是0.05，表示map task完成数目达到5%后，才可以调度reduce task）后，才开始让Reduce Task调度运行。然而，在实际运行应用中，由于不同作业的Map Task运行时间长短不一，对于一些Map Task运行时间较长的作业，即使只剩下一个Map Task时启动Reduce Task也会造成Reduce slot利用率低下，为了优化这种情况，可以让最后一个Map Task剩余时间达到一定值后，才开始调度Reduce Task。我们可通过已经运行完成的Map Task运行时间估算正在运行的Map Task的剩余运行时间。让服务端控制Reduce启动时机，既减少了用户配置参数，又能优化集群的资源利用率，一举两得。

2. 动态调整slots配置

（1）动态增减TaskTracker上的slots个数

现在Hadoop中使用slot表示各个TaskTracker上的资源，一个slot代表固定的资源组合，比如<2G内存,1CPU>，管理用可通过参数“mapred.tasktracker.map.tasks.maximum”和mapred.tasktracker.reduce.tasks.maximum配置每个TaskTracker上的Map slot和Reduce slot个数，为JobTracker中的调度器负责将各个TaskTracker上的map slot和reduce slot分配给各个作业。这两个参数一旦配置后，不能够动态修改，如果想修改必须重启对应的TaskTracker服务。由于不同作业的任务需要的资源量不同，因此，尽管节点资源相同，跑的任务数目相同，但机器的资源利用率可能是不同的。为此，可修改TaskTracker相关代码，让其可根据资源使用情况动态增减slot数目。

（2）根据集群map和reduce使用比例调整

由于map slot和reduce slot不能在map task和reduce task之间共享（map slot只能供map task使用，reduce slot只能供reduce task使用），因此，当某一种类型任务对资源需求量而另外一种小时，会造成后一种资源浪费，为此，可根据集群map和reduce需求比例动态调整map slot和reduce slot数目。

3. 消除异常Job的影响

（1）内存监控

1） 单个Task内存限制

Task内存控制是由各个TaskTracker负责的，TaskTracker会启动一个额外的线程监控每个Task进程树所消耗的总内存，一旦超过了用户设置的内存量，则将其杀掉。管理员可通过参数“mapred.cluster.map.memory.mb”和“mapred.cluster.reduce.memory.mb”设置每个map slot和reduce slot代表的内存量，一旦设置了这两个参数，则会启动Task内存监控线程。用户提交作业时，可通过参数“mapred.job.map.memory.mb”和“mapred.job.reduce.memory.mb”设置每个map task和reduce task要消耗的内存量。

当然，对于一些大内存作业，可能要占用多个slot，这要求调度器支持大内存作业调度，当前仅有Capacity Scheduler支持该功能，其他调度器，如Fair Scheduler和FIFO，仅支持“一个task一个slot”。

2） 计算节点内存上限控制

JobTracker是否为一个TaskTracker分配任务，并不取决于该TaskTracker是否有空闲的slot。即使存在空闲的slot，TaskTracker也可根据需要选择是否接收（askForNewTask=false则不接收，该变量在心跳中），比如，当前该TaskTracker内存利用率过高，它可通过心跳告诉JobTracker不要为之分配新的任务。这一点，当前Hadoop并未实现，需要各公司研发人员添加，O(∩_∩)O~。

（2）磁盘IO监控

1）Shuffle磁盘的读取限制

2）Shuffle速度限制

前面两个feature，当前Hadoop尚未实现，个人觉得，可通过两种方法实现，一种是利用cgroups进行资源隔离，另一种不太严格的方法是调节Reduce Task的并发数据拷贝线程数目，即调整参数“mapred.reduce.parallel.copies”，默认是5，或者根据需要调整每个TaskTracker上允许同时通过Jetty Server读取数据的Reduce Task数目。

3）Job的日志量限制

管理员可在TaskTracker上通过参数“mapreduce.cluster.map.userlog.retain-size”和“mapreduce.cluster.reduceuserlog.retain-size”设置每个Map Task或者Reduce Task保存的日志量，如果任务输出日志超过该限制，则只保存后面的那一部分。

（3）Job的Task数限制

如果一个Job包含大量的Task，可能会将JobTracker内存耗光，为此，可在JobTracker端设置Map Task和Reduce Task上限，将“大作业”扼杀在摇篮中，这一点，当前Hadoop并未实现。

（4）Job的Counter数限制

如果作业包含大量的counter，会严重降低该作业的性能，为此，用户可通过参数“mapreduce.job.counters.limit”设置作业Counter数目上限，默认是120，另外，作业的Counter Group数目上限是50，这个不能配置。

（5）Job本地文件系统数据读写量监控

（6）Job创建HDFS文件数目的监控

以上两个只是监控，可将相应的数值表示成Hadoop系统Counter，这样可在日志和界面上看到这些监控值。

4. Master节点的单点性能压力

（1）NameNode改进

略过

（2）JobTracker改进

1） Scheduler调度算法重写，从O(n2)降低到O(1)

在Hadoop中，当一个TaskTracker通过心跳向JobTracker请求资源时，Jobracker会调用调度器的assignTasks()函数为该TaskTracker分配任务，调度器需要扫描系统中所有作业和任务以便选择一个或多个最合适的Task，如果作业和任务数目非常多，这必然是非常耗时的。为了降低时间复杂度，通常在调度器中启动一个更新线程，这个线程实时更新各个作业和任务的状态，并将下次可以调度的任务放到一个数据结构中，这样，当一个TaskTracker请求任务时，可直接从该数据结构中取。另外，可通过Set或者Map这种树形结构维护作业和任务信息，以便降低时间复杂度。

2） 一次心跳分配多个Task

对于多租户调度器，包括Fair Scheduler和Capacity Scheduler，旧的Hadoop版本（具体指哪个版本，我没调查）仅支持一个心跳分配一个任务，在比较新的Hadoop版本中，则支持一个心跳分配多个任务，这样可大大提高任务分配速度。在Capacity Scheduler中，管理员可通过参数“mapred.capacity-scheduler.maximum-tasks-per-heartbeat ”设置一次心跳最多分配的任务数，默认是Short.MAX_VALUE；在Fair Scheduler中，管理员可通过参数“mapred.fairscheduler.assignmultiple”设置是否允许一次心跳分配多个任务，默认是true（TaskTracker能容纳几个任务就分配几个），当然，管理员可通过设置参数“mapred.fairscheduler.assignmultiple.maps”和“mapred.fairscheduler.assignmultiple.reduces”设置一次心跳最多分配的Map/Reduce Task数目，默认是-1，表示不限制。

3）Job History改造成异步写

Hadoop会为一个作业运行过程中的重要阶段作日志，比如提交作业时，启动作业时，任务开始运行时等，这些日志有两个作用，一个是JobTracker出现问题重启时恢复之前运行的作业（JobTracker HA），二是方便用户查询历史作业运行信息。然而，作业日志是同步做的，伪代码如下：

Job.submitJob();

JobHistory.submitJob(job); //执行完这条语句，才能继续往下执行

同步作日志有一个问题是，当写日志时，由于磁盘、文件系统等原因，卡住了，则作业无法往下执行，甚至可能失败。一种解决方法是将写日志改成异步的，即：

Job.submitJob();

JobHistory.addJobEvent(new SubmitJobEvent(job)); //将写日志封装成事件放到JobHistory的一个队列中，由一个consumer线程从这个队列中取事件进行处理。只要将事件添加到队列中就可以往下执行。

异步写日志已在下一代MpaReduce中采用。

4）Out-of-bound heartbeat提高调度的效率

我们知道，TaskTracker与JobTracker之间的通信方式采用了pull模型，即JobTracker永远不会主动与TaskTracker通信，而是TaskTracker周期性地主动通过RPC心跳与JobTracker通信，汇报各个任务的运行状态和自己的资源使用情况， JobTracker则通过心跳应答为TaskTracker分配任务。

然而，Hadoop的心跳间隔最低是3s（从Hadoop 0.21.0开始，为了照顾小集群用户，将该默认值改为500ms），且随着集群规模递增，默认情况下，每增加100个节点，会增加1s（可以配置），比如你的集群规模是500，则心跳间隔是8s，8s啊，亲！如果一个TaskTracker运行完一个Task并释放了slot，则它最长需要等待8s才能获取下一个新任务！为了解决该问题，Hadoop采用了一种基于带外心跳（Out-of-bound heartbeat）的优化方法：当一个Task运行结束或者Task运行失败时，可立刻触发一个心跳（带外心跳），与JobTracker通信以领取新的任务。管理员可以在各个TaskTracker上通过将参数“mapreduce.tasktracker.outofband.heartbeat”置为true以启动带外心跳这一功能。

推荐阅读：

1、统一资源管理与调度平台（系统）介绍

http://dongxicheng.org/mapreduce-nextgen/mesos_vs_yarn/

2、Hadoop中Speculative Task调度策略

http://dongxicheng.org/mapreduce/hadoop-speculative-task/