Hadoop中的Speculative Task调度策略

Posted 2020-06-12

一：背景

Speculative Task，又叫推测式任务，是指在分布式集群环境下，因为程序bug，负载不均衡或者资源分布不均，造成同一个job多个task运行速度不不一致，有的task运行速度明显要慢于其他task(比如：一个job的某个task进度只有10%，而其他所有task已经运行完毕)，则这些task拖慢了了作业的整体执行进度，为了避免这种情况发生，Hadoop会为该task启动Speculative task，让该Speculative task与原始task同时运行，哪个先运行完，则使用它的结果。

Speculative Task思路是以空间换时间的，同时启动多个相同Task，哪个完成的早，则采用哪个Task的结果，这样明显可以提高任务计算速度，但是，这样却会占用更多的资源，在集群资源紧缺的情况下，合理的控制Speculative Task，可在多用少量资源情况下，减少大作业的计算时间。

 

本文主要介绍了Hadoop各个版本中Speculative Task设计思路，并指出了各自的不足及改进之处。

二：Hadoop 0.20.2和1.0.0中的算法

【算法流程】

如果一个task满足以下条件，则会为该task启动Speculative Task：

(1)：该task还未启动任何Speculative Task(TaskInProgress仅对应一个running的task)。

(2)：在60s内落后(同一个作业所有task的)平均进度20%。

某个Task可能对应多个正在运行的attempt task，任何一个运行结束，则会kill(不是fail)其他task。

【相关代码】

 

[java] view plain copy

 

1. boolean hasSpeculativeTask(long currentTime, double averageProgress) {  
2. // these constants should be examined  
3. // in more depth eventually...  
4. //  
5.   if (!skipping && activeTasks.size() <= MAX_TASK_EXECS &&  
6.      (averageProgress - progress >= SPECULATIVE_GAP) &&  
7.          (currentTime - startTime >= SPECULATIVE_LAG)  
8.             && completes == 0 && !isOnlyCommitPending()) {  
9.     return true;  
10.   }  
11.   return false;  
12. }  

【存在问题】

 

以上算法可能造成以下几个问题：

(1)：作业的某个task被卡在87.7%处MapReduce-94。
(2)：当作业将要运行结束时，总不会启动Speculative Task。

(3)：各个参数不可配置(SPECULATIVE_GAP,SPECULATIVE_LAG)，不够灵活。

 

三：Hadoop0.21.0中的算法

为了对Hadoop-0.20.2中的算法进行改进，Hadoop-0.21.0进行了如下优化：

1：添加了三个可配置选项：

 

[java] view plain copy

 

1. mapreduce.job.speculative.slownodethreshold，默认是1，用于判断某个TaskTracker是否适合启动某个task的speculative task  
2. mapreduce.job.speculative.slowtaskthreshold，默认是1，用于判断某个task是否可以启动speculative task  
3. mapreduce.job.speculative.speculativecap， 默认是0.1，用于限定某个job最多同时启动的spculative task的数目  

2：限定条件

 

如果一个TaskTracker/job/task满足以下条件，则会在该TaskTracker上为task启动 一个Speculative task：

(1)：Job已经启动的Speculative task数目小于SpeculativeCap。

(2)：该TaskTracker上该Job的所有task平均进度不小于SlowNodeThreshold，具体判断方法为：

 

[java] view plain copy

 

1. tasktracker. mean-job.progessRate >job.std*job. SlowNodeThreshold  

其中，TaskTracker.Mean为该Job在该TaskTracker正在运行/已经运行完成的task的平均进度，job.progressRate为该作业的平均计算速度，job.std为该作业所有task计算速度的标准方差。

 

(3)：按照Task剩余时间，对Task进行排序

Task剩余时间定义为：(1-progress)/progressRate，其中process为task当前进度，progressRate为task的平均计算速度：progressRate=progress/deltaTime,其中deltaTime为该task启动以来所经历的时间。

(4)：选择一个剩余时间最多，且mean-progressRate>std*SlowTaskThreshold的task，并为该task启动Speculative task，其中mean为所有task平均计算速度，std为所有task计算速度的标准方差。

 

3.存在问题

(1)：MapReduce-2062

当一个作业只有一个task时，所有task的计算速度的标准方差为0，因而，总会为这样的作业启动Speculative task。

(2)：MapReduce-3895

在Hadoop中，reduce task进度(对应上面的progress变量)计算很不合理，采用的方法是，将reduce task分为三个子过程：Shuffle(copy)，sort和reduce，各个阶段占进度的1/3，比如，一个task的Shuffle阶段刚结束，它的进度应该是33.3%。对于大部分作业，reduce task的进度变化并不是均匀的，一个task在某一时刻进度为33.3%，下一秒进度可能变为66.6%，这主要是因为sort阶段所做的工作非常少，时间很短，也正是由于以上原因，reduce task很容易被误判速度过慢，进而启动Speculative task。一种可行的解决方案是将sort阶段利用某个数学模型平滑处理。

 

四：终极解决方案

拖后腿task出现的原因是系统负载不均衡和资源分配不均。尤其是在异构Hadoop集群中，拖后腿task会经常出现，而且最好不要打开Speculative task功能，否则会出现大量的Speculative task，造成严重的资源浪费，因为当前所有的Speculative task解决方案军事假设进群是同构的。

为什么会造成这种问题？根本原因是这种基于Speculative task来解决拖后腿task是有问题的。拖后腿task最终应该通过调度器解决：每个TaskTracker通过heartbeat会报自己机器上资源分布情况和当前资源使用情况，比如CPU个数，CPU利用率，内存剩余量，IO负载等，然后由调度器根据当前资源使用情况，动态对任务进行调度(参考：https://issues.apache.org/jira/browse/MAPREDUCE-1380)，进而最大限度避免产生拖后腿task。

关于拖后腿task，还有一个需要解决的问题是，防止为某个task启动Speculative task后，该task又变成拖后腿task，比如：某个node上所有task均往一个磁盘上写数据，且该磁盘负载很重，而此时又将一个Speculative task调度到该节点上(也往该磁盘上写数据)，此时，该Speculative task变得很缓慢，因而有人提出了Hadoop disk scheduler(参考：https://issues.apache.org/jira/browse/MAPREDUCE-2636)。

五：关于Speculative Task的Hadoop代码

Hadoop中，关于推测式任务的相关代码均在文件JobInProgress.java和TaskInProgress.java中，JobInProgress.java主要函数定义如下：

 

[java] view plain copy

 

1. protected synchronized TaskInProgress findSpeculativeTask(  
2.   Collection<TaskInProgress> list, String taskTrackerName,  
3.     String taskTrackerHost, TaskType taskType) {….}  

TaskInProgress.java中主要函数定义如下：

 

 

[java] view plain copy

 

1. boolean canBeSpeculated(long currentTime){…}  

文章来自：董的博客