动态资源分配（Dynamic Resource Allocation）

转自：http://zhangxiong0301.iteye.com/blog/2192641

从spark1.2开始，可以根据application的负载动态地增加和减少分配给application的资源。也就是说，你的application在不需要资源的时候会把资源退还给集群，而在需要的时候重新申请获得资源。这在spark集群上有多个application时候很有用。当分配给某个application的资源处于空闲状态，这些资源会退还到集群的资源池从而被其他application使用。spark中的动态资源分配的粒度是executor，通过spark.dynamicAllocation.enabled=true即可开启。

目前这个功能是关闭的，仅在yarn中有效。未来发行版本会在standalone和memos coarse-grained模式应用。虽然memos目前在fine-grained模式下有类似的动态资源共享，但是开启动态资源分配可以减少memos在粗粒度的调度延迟。

配置

所有配置都在spark.dynamicAllocation.*的命名空间下。为了使用动态资源分配特性，application必须设置spark.dynamicAllocation.enabled=true，以及通过spark.dynamicAllocation.minExecutors 和spark.dynamicAllocation.maxExecutors分别设置executor数量的上下界。

另外，spark应用必须使用独立的shuffle service。这个shuffle service的目的是保存executor产生的文件并提供给后续任务使用，从而使得executor可以安全的移除。要启用独立的shuffle sevice，需要设置spark.shuffle.service.enabled=true。在yarn中，实现该shuffle service的类是 org.apache.spark.yarn.network.YarnShuffleService，该服务会运行在所有的nodemanager。具体启动shuffle service 的步骤如下：

1.编译spark，同时指定YARN profile。

2.找到spark--yarn-shuffle.jar。这是应该是shuffle sevice。如果在编译时候没有指定–tgz，则这个jar包在 $SPARK_HOME/network/yarn/target/scala-目录下；否则就在打包后的发行版的lib目录。

3.拷贝上述jar包到yarn集群的所有nodemanager。

4.在每个nodemanager的yarn-site.xml配置文件里： yarn.nodemanager.aux-services配置项添加spark_shuffle ；配置项yarn.nodemanager.aux-services.spark_shuffle.class添加org.apache.spark.network.yarn.YarnShuffleService。另外设置所有和spark.shuffle.service.* 相关的配置项。

5.重启所有nodemanager。

资源分配策略

从上层看，spark应该在不需要的时候减少executor，在需要的时候动态增加executor。虽然没有确切的方式去预测即将被去掉的executor会马上被重新用到，或者即将被添加的executor马上会空闲，我们需要一些启发式算法来动态增减executor。

请求策略

开启动态分配策略后，application会在task因没有足够资源被挂起的时候去动态申请资源，这种情况意味着该application现有的executor无法满足所有task并行运行。spark一轮一轮的申请资源，当有task挂起或等待spark.dynamicAllocation.schedulerBacklogTimeout时间的时候，会开始动态资源分配；之后会每隔spark.dynamicAllocation.sustainedSchedulerBacklogTimeout时间申请一次，直到申请到足够的资源。每次申请的资源量是指数增长的，即1,2,4,8等。

之所以采用指数增长，出于两方面考虑。其一，开始申请的少是考虑到可能application会马上得到满足；其次要成倍增加，是为了防止application需要很多资源，儿该方式可以在很少次数的申请之后得到满足。

删除executor策略

很简单，当application的executor空闲时间超过spark.dynamicAllocation.executorIdleTimeout后，就会将其删除掉。

华丽的移除executor

在动态资源分配之前，当application完成或者executor运行出错次数超过限定值时，executor就会安全退出，此时executor的所有状态将不再需要。在动态资源分配情况下，当executor被移除时，application还在执行。如果application要访问被移除executor的状态，则需要重新计算其状态。因此，spark需要一种机制确保移除executor之前保存executor的状态。

这个需求对shuffle是很重要的。在shuffle过程中，spark executor首先保存它的map结果文件到本地磁盘上，然后以server的身份供其他executor来获取文件。对于某些executor执行比其他executor慢很多的情况，动态资源分配会移除空闲的executor，当后续任务需要被移除executor的结果文件时，就要重新计算结果文件了。

保存已移除executor的结果文件的方式就是使用独立的shuffle server。这个server是一个在所有nodemanager都会长期运行的进程。当开启独立shuffle server时，executor将会从该server直接读取文件而不是相互之间获取文件。这样executor产生的结果文件就会比executor具有更长的生命周期。

另外，除了executor的shuffle文件，executor还会在磁盘或内存缓存数据。当executor被移除后，这些缓存的数据将不再可用。目前还没有解决方案。