spark on yarn上的dynamic executor

　　首先交代一下背景: 先前公司采用的spark版本是1.6.1, spark1.6之后有一个新的特性：Dynamic Allocation，也就是动态的executor，会根据task的情况自动增加或者

减少executor的个数，这样就能够在一定程度上避免资源的浪费。但很遗憾，先前公司使用spark的同事并没有使用这一特性。后来spark2.1.0出来之后，经过测试，发现

性能要比spark1.x好一些，在这种背景下，决定对spark进行升级，并应用动态的executor，防止资源的滥用。

　　这里着重说一下spark on yarn的动态executor的配置。其实官方文档写的很详细。这里写下来就当是自己对这一块的总结。

　　步骤：

　　1. 修改所有的nodemanger节点的yarn-site.xml, 如下：

　　<property>

 　　  　　 <name>yarn.nodemanager.aux-services</name>

   　　　　 <value>mapreduce_shuffle,spark_shuffle</value>

 　　</property>

 　　<property>

   　　　　  <name>yarn.nodemanager.aux-services.spark_shuffle.class</name>

   　　　　  <value>org.apache.spark.network.yarn.YarnShuffleService</value>

  　　</property>

 

      2. 将spark-2.1.0-yarn-shuffle.jar 拷贝到 所有nodemanger节点的classpath下，比如这里我拷贝到了/data1/app/hadoop-2.7.1/share/hadoop/yarn/lib 下

 

 　  3. 重启所有的nodemanger节点

 

　　以上就配置完毕，接下来说一下使用方式：

　　为了使用动态executor，在提交程序的时候，需要带上下面的参数(当然也可以配置在spark-default.conf中), 建议在提交程序的时候配置该参数。

　　配置示例：

　　--conf spark.dynamicAllocation.enabled=true

　　--conf spark.shuffle.service.enabled=true

　　--conf spark.dynamicAllocation.initialExecutors=10

　　--conf spark.dynamicAllocation.maxExecutors=50

　　--conf spark.dynamicAllocation.minExecutors=3

　　--conf spark.dynamicAllocation.executorIdleTimeout=30s

　　--conf spark.dynamicAllocation.schedulerBacklogTimeout=3s

　　--conf spark.dynamicAllocation.cachedExecutorIdleTimeout=30s

　　在以前使用spark-submit或者spark-shell的基础上额外加上以上配置就可以了。

 

　　下面对这些配置做一个简单的说明, 也可以参考官方文档，这里主要说一下下面的三个参数，因为其它的都很好理解:

　　1. spark.dynamicAllocation.executorIdleTimeout:  executor的空闲时间，如果经过该参数设置的时间之后，该executor上还没有分配任务，那么就会认为该executor是空闲的，该executor就可以移除了;

　　2. spark.dynamicAllocation.schedulerBacklogTimeout:  如果有task积压等待，那么经过该参数设置的时间之后，将会为task申请新的executor

　　3. spark.dynamicAllocation.cachedExecutorIdleTimeout：默认情况下如果executor中缓存了数据，即使executor上没有任务在跑，该executor也不会被移除。因为默认该参数是infinity, 也就是无穷大。

　　　所以可以合理的设置改参数，防止资源浪费。比如我示例中设置的是30s，那就说明即使该节点缓存了数据，但是它却没有任务在跑，那么过了时间之后该executor也会被移除。

 

 

　　以上就是spark on yarn的动态资源配置以及一些说明。

 

　　再说一个注意事项：我们在生产环境中有大量的使用spark sql的广播，本来好好的，在上了动态executor之后频繁报错，后来发现只要有uncache语法，很大几率就会报错。经过排查日志，发现都是

spark sql中的 uncache 语法会引发报错，后来全部去掉uncache语句之后就正常了。估计这是spark动态executor的bug，不知到在以后的版本中是否有修复。关于spark sql的广播使用方式，后面再整理。