Spark Job的提交与task本地化分析(源码阅读八)

简介:   我们又都知道,Spark中任务的处理也要考虑数据的本地性(locality),Spark目前支持PROCESS_LOCAL(本地进程)、NODE_LOCAL(本地节点)、NODE_PREF、RACK_LOCAL(本地机架)、ANY(任何)几种。

  我们又都知道,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,那么下来调用LocalBackendreviveOffers方法,向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)的资源时,如果固执的期盼得到最佳的资源,很可能被已经占用最佳资源但是运行时间很长的任务耽误,所以这些代码实现了当没有最佳本地化时,选择稍差点的资源。

 

参考文献:《深入理解Spark:核心思想与源码分析》

目录
相关文章
|
4月前
|
机器学习/深度学习 分布式计算 算法
Spark快速大数据分析PDF下载读书分享推荐
《Spark快速大数据分析》适合初学者,聚焦Spark实用技巧,同时深入核心概念。作者团队来自Databricks,书中详述Spark 3.0新特性,结合机器学习展示大数据分析。Spark是大数据分析的首选工具,本书助你驾驭这一利器。[PDF下载链接][1]。 ![Spark Book Cover][2] [1]: https://zhangfeidezhu.com/?p=345 [2]: https://i-blog.csdnimg.cn/direct/6b851489ad1944548602766ea9d62136.png#pic_center
161 1
Spark快速大数据分析PDF下载读书分享推荐
|
9天前
|
SQL 分布式计算 Serverless
EMR Serverless Spark:一站式全托管湖仓分析利器
本文根据2024云栖大会阿里云 EMR 团队负责人李钰(绝顶) 演讲实录整理而成
66 2
|
19天前
|
设计模式 数据采集 分布式计算
企业spark案例 —出租车轨迹分析
企业spark案例 —出租车轨迹分析
51 0
|
4月前
|
弹性计算 分布式计算 Serverless
全托管一站式大规模数据处理和分析Serverless平台 | EMR Serverless Spark 评测
【7月更文挑战第6天】全托管一站式大规模数据处理和分析Serverless平台 | EMR Serverless Spark 评测
23703 42
|
5月前
|
分布式计算 定位技术 Scala
使用spark基于出租车GPS数据实现车辆数量统计以及北京每个城区的车辆位置点数分析
使用spark基于出租车GPS数据实现车辆数量统计以及北京每个城区的车辆位置点数分析
105 0
|
分布式计算 Spark
|
存储 分布式计算 调度
Spark源码分析之六:Task调度(二)
        话说在《Spark源码分析之五:Task调度(一)》一文中,我们对Task调度分析到了DriverEndpoint的makeOffers()方法。这个方法针对接收到的ReviveOffers事件进行处理。
1139 0
|
分布式计算 资源调度 调度
Spark源码分析之五:Task调度(一)
        在前四篇博文中,我们分析了Job提交运行总流程的第一阶段Stage划分与提交,它又被细化为三个分阶段:         1、Job的调度模型与运行反馈;         2、Stage划分;         3、Stage提交:对应TaskSet的生成。
1267 0
|
4天前
|
分布式计算 大数据 Apache
ClickHouse与大数据生态集成:Spark & Flink 实战
【10月更文挑战第26天】在当今这个数据爆炸的时代,能够高效地处理和分析海量数据成为了企业和组织提升竞争力的关键。作为一款高性能的列式数据库系统,ClickHouse 在大数据分析领域展现出了卓越的能力。然而,为了充分利用ClickHouse的优势,将其与现有的大数据处理框架(如Apache Spark和Apache Flink)进行集成变得尤为重要。本文将从我个人的角度出发,探讨如何通过这些技术的结合,实现对大规模数据的实时处理和分析。
24 2
ClickHouse与大数据生态集成:Spark & Flink 实战

相关实验场景

更多