Spark 性能相关参数配置详解－任务调度篇

随着Spark的逐步成熟完善, 愈来愈多的可配置参数被添加到Spark中来, 本文试图通过论述这其中部份参数的工作原理和配置思路, 和大家1起探讨1下如何根据实际场合对Spark进行配置优化。

由于篇幅较长，所以在这里分篇组织，如果要看最新完全的网页版内容，可以戳这里：http://spark-config.readthedocs.org/，主要是便于更新内容

schedule调度相干

调度相干的参数设置，大多数内容都很直白，其实不必过量的额外解释，不过基于这些参数的经常使用性（大概会是你针对自己的集群第1步就会配置的参数），这里多少就其内部机制做1些解释。

spark.cores.max

1个集群最重要的参数之1，固然就是CPU计算资源的数量。spark.cores.max 这个参数决定了在Standalone和Mesos模式下，1个Spark利用程序所能申请的CPU Core的数量。如果你没有并发跑多个Spark利用程序的需求，那末可以不需要设置这个参数，默许会使用spark.deploy.defaultCores的值（而spark.deploy.defaultCores的值默许为Int.Max，也就是不限制的意思）从而利用程序可使用所有当前可以取得的CPU资源。

针对这个参数需要注意的是，这个参数对Yarn模式不起作用，YARN模式下，资源由Yarn统1调度管理，1个利用启动时所申请的CPU资源的数量由另外两个直接配置Executor的数量和每一个Executor中core数量的参数决定。（历史缘由造成，不同运行模式下的1些启动参数个人认为还有待进1步整合）

另外，在Standalone模式等后台分配CPU资源时，目前的实现中，在spark.cores.max允许的范围内，基本上是优先从每一个Worker中申请所能得到的最大数量的CPU core给每一个Executor，因此如果人工限制了所申请的Max Core的数量小于Standalone和Mesos模式所管理的CPU数量，可能产生利用只运行在集群中部份节点上的情况（由于部份节点所能提供的最大CPU资源数量已满足利用的要求），而不是平均散布在集群中。通常这不会是太大的问题，但是如果触及数据本地性的场合，有可能就会带来1定的必须进行远程数据读取的情况产生。理论上，这个问题可以通过两种途径解决：1是Standalone和Mesos的资源管理模块自动根据节点资源情况，均匀分配和启动Executor，2是和Yarn模式1样，允许用户指定和限制每一个Executor的Core的数量。 社区中有1个PR试图走第2种途径来解决类似的问题，不过截至我写下这篇文档为止（2014.8），还没有被Merge。

spark.task.cpus

这个参数在字面上的意思就是分配给每一个任务的CPU的数量，默许为1。实际上，这个参数其实不能真的控制每一个任务实际运行时所使用的CPU的数量，比如你可以通过在任务内部创建额外的工作线程来使用更多的CPU（最少目前为止，将来任务的履行环境是不是能通过LXC等技术来控制还不好说）。它所发挥的作用，只是在作业调度时，每分配出1个任务时，对已使用的CPU资源进行计数。也就是说只是理论上用来统计资源的使用情况，便于安排调度。因此，如果你期望通过修改这个参数来加快任务的运行，那还是赶快换个思路吧。这个参数的意义，个人觉得还是在你真的在任务内部自己通过任何手段，占用了更多的CPU资源时，让调度行动更加准确的1个辅助手段。

spark.scheduler.mode

这个参数决定了单个Spark利用内部调度的时候使用FIFO模式还是Fair模式。是的，你没有看错，这个参数只管理1个Spark利用内部的多个没有依赖关系的Job作业的调度策略。

如果你需要的是多个Spark利用之间的调度策略，那末在Standalone模式下，这取决于每一个利用所申请和取得的CPU资源的数量（暂时没有取得资源的利用就Pending在那里了），基本上就是FIFO情势的，谁先申请和取得资源，谁就占用资源直到完成。而在Yarn模式下，则多个Spark利用间的调度策略由Yarn自己的策略配置文件所决定。

那末这个内部的调度逻辑有甚么用呢？如果你的Spark利用是通过服务的情势，为多个用户提交作业的话，那末可以通过配置Fair模式相干参数来调剂不同用户作业的调度和资源分配优先级。

spark.locality.wait

spark.locality.wait和spark.locality.wait.process，spark.locality.wait.node, spark.locality.wait.rack这几个参数影响了任务分配时的本地性策略的相干细节。

Spark中任务的处理需要斟酌所触及的数据的本地性的场合，基本就两种，1是数据的来源是HadoopRDD; 2是RDD的数据来源来自于RDD Cache（即由CacheManager从BlockManager中读取，或Streaming数据源RDD）。其它情况下，如果不触及shuffle操作的RDD，不构成划分Stage和Task的基准，不存在判断Locality本地性的问题，而如果是ShuffleRDD，其本地性始终为No Prefer，因此其实也无所谓Locality。

在理想的情况下，任务固然是分配在可以从本地读取数据的节点上时（同1个JVM内部或同1台物理机器内部）的运行时性能最好。但是每一个任务的履行速度没法准确估计，所以很难在事前取得全局最优的履行策略，当Spark利用得到1个计算资源的时候，如果没有可以满足最好本地性需求的任务可以运行时，是退而求其次，运行1个本地性条件稍差1点的任务呢，还是继续等待下1个可用的计算资源已期望它能更好的匹配任务的本地性呢？

这几个参数1起决定了Spark任务调度在得到分配任务时，选择暂时不分配任务，而是等待取得满足进程内部/节点内部/机架内部这样的不同层次的本地性资源的最长等待时间。默许都是3000毫秒。

基本上，如果你的任务数量较大和单个任务运行时间比较长的情况下，单个任务是不是在数据本地运行，代价区分可能比较显著，如果数据本地性不理想，那末调大这些参数对性能优化可能会有1定的好处。反之如果等待的代价超过带来的收益，那就不要斟酌了。

特别值得注意的是：在处理利用刚启动后提交的第1批任务时，由于当作业调度模块开始工作时，处理任务的Executors可能还没有完全注册终了，因此1部份的任务会被放置到No Prefer的队列中，这部份任务的优先级仅次于数据本地性满足Process级别的任务，从而被优先分配到非本地节点履行，如果的确没有Executors在对应的节点上运行，或的确是No Prefer的任务（如shuffleRDD），这样做确切是比较优化的选择，但是这里的实际情况只是这部份Executors还没来得及注册上而已。这类情况下，即便加大本节中这几个参数的数值也没有帮助。针对这个情况，有1些已完成的和正在进行中的PR通过例如动态调剂No Prefer队列，监控节点注册比例等等方式试图来给出更加智能的解决方案。不过，你也能够根据本身集群的启动情况，通过在创建SparkContext以后，主动Sleep几秒的方式来简单的解决这个问题。

spark.speculation

spark.speculation和spark.speculation.interval,spark.speculation.quantile, spark.speculation.multiplier等参数调剂Speculation行动的具体细节，Speculation是在任务调度的时候，如果没有合适当前本地性要求的任务可供运行，将跑得慢的任务在空闲计算资源上再度调度的行动，这些参数调剂这些行动的频率和判断指标，默许是不使用Speculation的。

通常来讲很难正确的判断是不是需要Speculation，能真正发挥Speculation用途的场合，常常是某些节点由于运行环境缘由，比如CPU资源由于某种缘由被占用，磁盘破坏致使IO缓慢造成任务履行速度异常的情况，固然条件是你的分区任务不存在仅能被履行1次，或不能同时履行多个拷贝等情况。Speculation任务参照的指标通常是其它任务的履行时间，而实际的任务可能由于分区数据尺寸不均匀，本来就会有时间差异，加上1定的调度和IO的随机性，所以如果1致性指标定得过严，Speculation可能其实不能真的发现问题，反而增加了没必要要的任务开消，定得过宽，大概又基本相当于没用。

个人觉得，如果你的集群范围比较大，运行环境复杂，的确可能常常产生履行异常，加上数据分区尺寸差异不大，为了程序运行时间的稳定性，那末可以斟酌仔细调剂这些参数。否则还是斟酌如何排除造成任务履行速度异常的因数比较靠铺1些。

固然，我没有实际在很大范围的集群上运行过Spark，所以如果看法有些偏颇，还请有实际经验的XD指正。