spark 调度优化

在做spark-streaming的时候最近遇到个特别的问题：

每个batch的任务调度执行的时候，某些excutor上调度的任务特别多，其他的excutor上只调度一个

甚至200个task只会调度到2个excutor上：

第二个图上看到下面所有的Locality Level 都为: NODE_LOCAL

第一个图上的调度很多的task也是NODE_LOCAL

所以可能原因：是spark调度策略的问题

spark 是计算追着数据走， storm 是数据追着计算走， 所以如果数据量比较小，要求延迟比较小， 就适合storm， 但是如果数据量比较大， 这个时候如果传输数据， 就会碰到很大的带宽占用和性能下降， 这个时候就比较适合让计算去找数据，

但是在计算找数据的过程中， 是怎么让计算找到数据呢，就是靠 spark 的计算本地性来决定

spark.locality系列参数，可以调节Spark等待task进行数据本地化的时间。spark.locality.wait(默认3s）、spark.locality.wait.node、spark.locality.wait.process、spark.locality.wait.rack，默认值是spark.locality.wait的值；

本地性从优到差排， PROCESS_LOCAL >NODE_LOCAL >NO_PREF >RACK_LOCAL；

就是总是尝试以最高的 locality level 去启动task， 如果对应需要是用到的 executor 正在使用中（跑别的task），满足不了， 就等一会（等待时间是有spark.locality.wait.process或spark.locality.wait.node或spark.locality.wait.rack来控制的）， 看看过一会这个忙线的host 或者 executor是不是解脱了， 如果已经空闲了，我就可以把 task 放在它最期望的 host 或者 executor 上去运行了， 这里赌的就是一般来说，task 执行耗时相对于网络传输/文件IO 要小得多，调度器多等待1 2秒可能就可以以更好的本地性执行 task，避免了更耗时的网络传输或文件IO

举个例子， 假如 一个 task 要处理的数据，在上一个 stage 中缓存下来了， 这个 task 期望的 就是以 PROCESS_LOCAL 来运行， 这个时候缓存数据的executor 不巧正在执行 其他的task， 那么我就等一会， 等多长时间呢， spark.locality.wait.process这么长时间， 如果时间超了， executor 还是没有空闲下来， 那么我没有办法， 我就以NODE_LOCAL 来运行 task， 这个时候我想到 同一台机器上其他 executor 上跨jvm 去拉取数据， 如果同一台机器上有其他空闲的 executor 可以满足， 就这么干， 如果没有， 等待 spark.locality.wait.node 时间， 还没有就以更低的 Locality Level 去执行这个 task

三个分别是局部的等待时间是可以根据实际情况调整：node、process、rack可以设置时间越来越短，rack可以设置为1s

这种调度原则通常是没有问题，在做实时任务的某些时候可能就会有问题；

由于kafka数据过滤后每个分区都比较小，默认的3s可能大部分都可以处理完，所以造成一直在延迟等待调度到NODE_LOCAL的节点上跑任务，经设小spark.locality.wait值问题得到解决

前面我们演示了如何使用集群跑一个Spark任务，比如WordCount程序。那么一个Spark任务在集群中到底是如何调度的呢？本节就来详细讨论一下Spark任务在集群中的调度过程。

Spark任务在集群中的调度过程如下图所示：

关于Spark任务在执行过程中的几点说明：

更多关于RDD、Partition、依赖关系、Stage、DAG的内容，我们将在后续章节进行深入的讨论。本节先介绍到这里。

spark默认调度模式：

Spark中的调度模式主要有两种：FIFO和FAIR。默认情况下Spark的调度模式是FIFO（先进先出），谁先提交谁先执行，后面的任务需要等待前面的任务执行。

而FAIR（公平调度）模式支持在调度池中为任务进行分组，不同的调度池权重不同，任务可以按照权重来决定执行顺序。使用哪种调度器由参数spark。scheduler。mode来设置，可选的参数有FAIR和FIFO，默认是FIFO。

含义

Spark默认采取FIFO策略运行多个Jobs，它提供一个队列来保存已经提交的Jobs，如果队头的Job不需要占用所有的集群资源，那么后续的 Job可以立即运行，但是如果队头的Job需要占用所有的集群资源，且运行时间很长，那么即使后续的Job很小，也要等待前面的Job执行完后才可以执 行，这样就造成了大量的延迟。

---