Spark Job的提交与task本地化分析（源码阅读）
Spark中任务的处理也要考虑数据的本地性(locality)，Spark目前支持PROCESS_LOCAL（本地进程）、NODE_LOCAL（本地节点）、NODE_PREF、RACK_LOCAL（本地机架）、ANY（任何）几种。其他都很好理解，NODE_LOCAL会在spark日志中执行拉取数据所执行的task时，打印出来,因为Spark是移动计算，而不是移动数据的嘛。
那么什么是NODE_PREF？
当Driver应用程序刚刚启动，Driver分配获得的Executor很可能还没有初始化，所以有一部分任务的本地化级别被设置为NO_PREF.如果是ShuffleRDD,其本地性始终为NO_PREF。这两种本地化级别是NO_PREF的情况，在任务分配时会被优先分配到非本地节点执行，达到一定的优化效果。
那么下来我们从job的任务提交开始玩起~
getMissingParentStages方法用来找到Stage的所有不可用的父Stage(宽依赖和窄依赖问题).从代码可以到这里是个递归的调用，submitWaitingStages实际上循环waitingStages中的stage并调用submitStaghe:
那么下来开始提交task，提交task的入口是submitMissingTasks，此函数在Stage没有可用的父stage时，被调用处理当前Stage未提交的任务。
1、那么在没有父stage时，会首先调用paendingPartitions.clear 用于清空pendingTasks.由于当前Stage的任务刚开始提交，所以需要清空，便于记录需要计算的任务。
2、将当前Stage加入运行中的Stage集合,是用HashSet进行构造的。
3、找出位计算的partition，如果Stage是map任务，那么outputLocs中partition对应的List为Nil,说明此partition还未计算。如果Stage不是map任务，那么需要获取stage的finalJob,调用finished方法判断每个partition的任务是否完成。
4、然后通过stage.makeNewStageAttemp，使用StageInfo.fromStage方法创建当前Stage的_latestInfo:
5、如果是Stage Map任务，那么序列化Stage的RDD及ShuffleDependency,如果Stage不是map任务，那么序列化Stage的RDD及resultOfJob的处理函数。最终这些序列化得到的字节数组需要用sc.broadcast进行广播。
6、最后，创建所有Task、当前stage的id、jobId等信息创建TaskSet,并调用taskScheduler的submitTasks，批量提交Stage及其所有Task.
有可能同时有多个任务提交，所以就有了调度策略FIFO，那么下来调用LocalBackend的reviveOffers方法，向local-Actor发送ReviveOffers消息。localActor对ReviveOffers消息的匹配执行reviveOffers方法。调用TaskSchedulerImpl的resourceOffers方法分配资源，最后调用Executor的launchTask方法运行任务。
同时你会发现，这里有段代码，shuffleOffers = Random.shuffle(offers)，是为了计算资源的分配与计算，对所有WorkerOffer随机洗牌，避免将任务总是分配给同样的WorkerOffer。
好了，知道了整个流程，下来我们来看一下本地化问题：
myLocalityLevles：当前TaskSetManager允许使用的本地化级别。那么这里的computeValidLocalityLevels方法是用于计算有效的本地化缓存级别。如果存在Executor中的有待执行的任务，且PROCESS_LOCAL本地化的等待时间不为0，且存在Executor已被激活，那么允许的本地化级别里包括PROCESS_LOCAL.
这里又发现新大陆，获取各个本地化级别的等待时间。
spark.locality.wait 本地化级别的默认等待时间
spark.locality.wait.process 本地进程的等待时间
spark.locality.wait.node 本地节点的等待时间
spark.locality.wait.rack 本地机架的等待时间
这些参数呢，在任务的运行很长且数量很多的情况下，适当调高这些参数可以显著提高性能，然而当这些参数值都已经超过任务的运行时长时，则需要调小这些参数。任何任务都希望被分配到可以从本地读取数据的节点上以得到最大的性能提升，然而每个任务的运行时长时不可预计的。当一个任务在分配时，如果没有满足最佳本地化(PROCESS_LOCAL)的资源时，如果固执的期盼得到最佳的资源，很可能被已经占用最佳资源但是运行时间很长的任务耽误，所以这些代码实现了当没有最佳本地化时，选择稍差点的资源。
最新文章
- 如何获取url中的参数并传递给iframe中的报表
- SQL Server 2008 R2——TRUNCATE TABLE 无法截断表 该表正由 FOREIGN KEY 约束引用
- JAVA Lambda Expressions streams
- flask-cors 实现跨域请求
- 《深度探索C++对象模型》学习笔记
- nginx 要改进的地方基础
- 表格table常见的边框设置和初步的立体效果
- nodejs的简单爬虫
- JAVA多线程实现和应用总结
- 【sklearn入门】通过sklearn实现k-means并可视化聚类结果
- 3.7 unittest之断言
- mininet的学习之二-----miniedit可视化
- 同步pod时区与node主机保持一致
- C# HtmlElement的GetAttribute("class") return ""
- Aborted connection 1055898 to db: 'xxx' user: 'yyy' host: 'xxx.xxx.xxx.xxx' (Got timeout reading communication packets)
- BIOS和CMOS【转载】
- spring-boot @Async 的使用、自定义Executor的配置方法
- 【BZOJ1046】上升序列（动态规划，贪心）
- grpc 实现微服务生态笔记
- 关于MFC中的OnPaint和OnDraw