spark 调度优化

1.问题

在做spark-streaming的时候最近遇到个特别的问题：
每个batch的任务调度执行的时候，某些excutor上调度的任务特别多，其他的excutor上只调度一个

image.png

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

image.png

2.分析

第二个图上看到下面所有的Locality Level 都为: NODE_LOCAL
第一个图上的调度很多的task也是NODE_LOCAL
所以可能原因：是spark调度策略的问题

3.spark task延迟调度的策略

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

image.png

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

3.1调度本地化判断

image.png

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；

3.2 spark 调度的总体原则（最大程度的满足 task 的本地性）：

就是总是尝试以最高的 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

4.结论

这种调度原则通常是没有问题，在做实时任务的某些时候可能就会有问题；
由于kafka数据过滤后每个分区都比较小，默认的3s可能大部分都可以处理完，所以造成一直在延迟等待调度到NODE_LOCAL的节点上跑任务，经设小spark.locality.wait值问题得到解决