1. 背景

推测执行(Speculative Execution)是指在分布式集群环境下,因为程序BUG,负载不均衡或者资源分布不均等原因,造成同一个job的多个task运行速度不一致,有的task运行速度明显慢于其他task(比如:一个job的某个task进度只有10%,而其他所有task已经运行完毕),则这些task拖慢了作业的整体执行进度,为了避免这种情况发生,Hadoop会为该task启动备份任务,让该speculative task与原始task同时处理一份数据,哪个先运行完,则将谁的结果作为最终结果。

推测执行优化机制采用了典型的以空间换时间的优化策略,它同时启动多个相同task(备份任务)处理相同的数据块,哪个完成的早,则采用哪个task的结果,这样可防止拖后腿Task任务出现,进而提高作业计算速度,但是,这样却会占用更多的资源,在集群资源紧缺的情况下,设计合理的推测执行机制可在多用少量资源情况下,减少大作业的计算时间。

关于MRv1中的推测执行机制,我已经在“Hadoop中Speculative Task调度策略”(http://dongxicheng.org/mapreduce/hadoop-speculative-task/)中进行了介绍,有兴趣的读者可先阅读这篇文章。

2. MRv2中采用的算法

MRv2限定了任意一个作业的备份任务的数目上线,该数目是以下三个数值的最大值:

  • MINIMUM_ALLOWED_SPECULATIVE_TASKS(常量10)
  • PROPORTION_TOTAL_TASKS_SPECULATABLE(常量0.01) * totalTaskNumber
  • PROPORTION_RUNNING_TASKS_SPECULATABLE (常量0. 1)* numberRunningTasks

当决定一个任务是够可以启动备份任务时,采用了下面的计算方法:

总是取speculationValue值最大的任务并为之启动备份任务,speculationValue计算方法为:

speculationValue= estimatedEndTime – estimatedReplacementEndTime

其中,estimatedEndTime是通过预测算法推测的该任务的最终完成时刻,计算方法为:

estimatedEndTime = estimatedRunTime + taskAttemptStartTime

其中,taskAttemptStartTime为该任务的启动时间,而estimatedRunTime为推测出来的任务运行时间,计算方法如下:

estimatedEndTime = (timestamp – start) / Math.max(0.0001, progress)

其中,timestamp为当前时刻,而start为任务开始运行时间,(timestamp-start)表示已经运行时间,progress为任务运行进度(0.0~1.0)。

estimatedReplacementEndTime含义为:如果此刻启动该任务,(可推测出来的)任务最终可能的完成时刻。很明显,如果estimatedReplacementEndTime大于estimatedEndTime,则没必要启动备份任务,因为即使启动了,它的完成时刻也会大于当前正在运行任务的完成时刻,只有当estimatedReplacementEndTime小于estimatedEndTime时,才有必要启动备份任务。而MRv2总是选择speculationValue值最大的任务并为之启动备份任务,且启动备份任务之前需检查是否满足以下条件:

  • 每个任务最多只能有一个备份任务
  • 已经完成的任务数目比例不小于MINIMUM_COMPLETE_PROPORTION_TO_SPECULATE(0.05,即5%),只有这样,才能有足够的历史信息估算estimatedReplacementEndTime

estimatedReplacementEndTime计算过方法为:{当前时刻}+{已经成功运行完成的任务所使用的平均运行时间}。

简单提炼一下核心思想:在某一时刻,判断一个任务是否拖后腿或者是否是值得为其运行备份任务时,则首先假设为其启动一个备份任务,那我们估算一下它的完成时间estimatedReplacementEndTime,同样,如果按照此刻该任务的计算速度,我们可以估算一下该任务最有可能的完成时间estimatedEndTime,如果estimatedEndTime与estimatedReplacementEndTime之差越大,则表明为该任务启动备份任务的价值越大。

3. 推测执行相关类

Speculator是一个服务,它由DefaultSpeculator实现,DefaultSpeculator接受并处理以下几种事件:

  • ATTEMPT_STATUS_UPDATE
  • ATTEMPT_START
  • TASK_CONTAINER_NEED_UPDATE
  • JOB_CREATE

DefaultSpeculator每隔一段事件会扫描一次所有正在运行的任务,如果一个任务可以启动备份任务,则会向Task发出一个T_ADD_SPEC_ATTEMPT事件,以启动另外一个任务实例。

DefaultSpeculator依赖于一个执行时间估算器,默认采用了LegacyTaskRuntimeEstimator,此外,MRv2还提供了另外一个实现:ExponentiallySmoothedTaskRuntimeEstimator,该实现采用了平滑算法对结果进行平滑处理。

原创文章,转载请注明: 转载自董的博客

本文链接地址: YARN/MRv2 MRAppMaster深入剖析—推测执行机制

微信公众号:hadoop-123,专注于大数据技术分享,欢迎加入!

说点什么

avatar
  Subscribe  
提醒