Spark源码学习1.1——DAGScheduler.scala

本文以Spark1.1.0版本为基础。

经过前一段时间的学习,基本上能够对Spark的工作流程有一个了解,但是具体的细节还是需要阅读源码,而且后续的科研过程中也肯定要修改源码的,所以最近开始Spark的源码的学习。首先以重要文件为基础分别分析,然后再整体的分析。

(一)DAGScheduler.scala文件的主要功能

DAGScheduler是划分Job为stage的调度,它是在作业所需要的数据已经被分为RDD之后执行的。DAGScheduler将Job划分为DAG图,以stage为图的结点,明确了各个stage之间的依赖关系,然后依次提交stage和Job。

(二)DAGScheduler.scala中的类和方法

1、initializeEventProcessActor()

该方法初始化eventProcessActor,也就是说,它可以阻塞线程直到supervisor启动,而在这整个过程中eventProcessActor都不会空闲。

2、executorHeartbeatReceived(execId,taskMetrics,blockManagerId)

该方法有三个参数,是用来接收心跳信息的。心跳信息是各个节点向master提供当前状态的,通知master当前节点是正在运行任务的,而运行任务的单位正是executor。

关于executor的方法还有:executorLost()处理executor执行失败,executorAdd()处理executor添加和tasksetFailed()设置任务失败。

3、getcacheLocs(RDD[_])

该方法查询本地缓存,检查本地存在哪些RDD,RDD的本地性是关系到stage分配的重要因素,在后文的函数中会实现。

关于Cache的方法还有:clearcacheLocs()清除缓存。

4、getShuffleMapStage(ShuffleDependence[_,_,_],JobId)

该方法是stage建立的关键,它明确stage的建立方法是以shuffle为边界的,如果出现wide依赖关系,那么就是shuffle的边界,就可以建立新的stage了,建立stage的方法为newOrUsedStage。

5、newStage(RDD[_],numTask,shuffleDependence,JobId,callsite)和newOrUsedStage(RDD[_],numTask,shuffleDependence,JobId,callsite)

这两个方法以同样的参数建立Stage,不同的是前者是建立无依赖关系的stage,而后者是由父系stage建立的有一定依赖关系的stage。

6、getParentStage(RDD[_],JobId)

该方法根据参数中的RDD数组(也就是一个stage中的所有RDD),查询当前RDD所属的Stage。

7、registerShuffleDependencies(shuffleDependencies[_,_,_],JobId)

该方法扫描依赖关系,利用getAncestorShuffleDependencies返回的RDD关系栈将依赖关系加入到Stage中。

8、getAncestorShuffleDependencies(RDD[_])

该方法建立RDD之间的依赖关系。

9、getMissingParentStages(stage)

该方法在getShuffleMapStage的基础上确定Stage的父系Stage,利用的也是RDD之间的依赖关系。

10、updataJobIdStageIdMaps(JobId,stage)

该方法用来更新Stage中的StageId和JobId。

11、cleanupStateForJobAndIndependentStages(Job)

该方法清除一个活动的Job的所有状态,以及其所属的不再需要的stage的状态。

12、submit[T,U](RDD[T],func,partitions,callsite,allowlocal,resulthandler,properties)

向任务调度器提交一个Job,并同时差生一个Jobwaiter,这个Jobwaiter可以用来保证Job执行时其他Job是阻塞的,也可以用来取消Job。

13、runJob[T,U](RDD[T],func,partitions,callsite,allowlocal,resulthandler,properties)

这个方法其实只是对Job提交之后的一个返回信息的处理,如果submitJob方法没有返回异常,就表示执行正确了,否则报错并加入日志。

14、runApproxiateJob[T,U,R](RDD[T],func,partitions,evaluator,callsite,timeout,properties)

该方法执行当前Job的下一个Job。

15、取消任务或者Stage的相关方法:doCancelAllJobs(),cancelStage(stageId)

16、重新提交失败的或者等待的Stage的方法:resubmitFailedStages(),submitWaitingStages()

17、runLocally(Job)和runLocallyWithinThread(job)

前者建立新的线程,启动本地执行;后者在线程中实际执行Job。

18、handle***()

这一些以handle开头的方法特点是进行了多次判定。handleJobSubmitted()方法对已经提交的Job进行操作,找到最后一个的Stage,检查Job执行的本地性等,然后将Job中的最后一个Stage提交,提交方法为submitStage()。submitStage()会根据最后一个Stage依次提交父系Stage,这其中需要考虑到丢失的task。handleTaskCompletion()方法在于处理Task执行后的各种状况,失败、阻塞或者失去联系,重新提交Task。详细的代码涉及诸多方法和类。

19、getPreferredLocsInternal(RDD,partition,visited)

根据Cache和输入RDD的位置和依赖关系递归查询最适合的位置。

时间: 2024-10-10 09:10:39

Spark源码学习1.1——DAGScheduler.scala的相关文章

Spark源码学习1.6——Executor.scala

Executor.scala 一.Executor类 首先判断本地性,获取slaves的host name(不是IP或者host: port),匹配运行环境为集群或者本地.如果不是本地执行,需要启动一个handler来监控所有的executor进程,避免阻塞.然后,初始化Spark执行环境.向SparkEnv注册executor资源,即registerSource方法.第三步,装载类,序列化类到内存中.第四,启动worker的线程池.第五,收集所有的task任务.接下就可以分配资源给task了,

Spark源码学习1.3——TaskSetManager.scala

TaskSetManager.scala TaskSet是指一系列被提交的task,一般是代表特定的stage中丢失的partition.TaskSetManager通过一个TaskSchedulerImpl实体来对TaskSet进行管理. 该方法定义了以下参数: EXECUTOR_TASK_BLACKLIST_TIMEOUT:executor加入黑名单的时间.如果executor意外终止导致任务执行失败,那么会暂时将这个executor加入黑名单,不再分配任务给它. SPECULATION_Q

Spark源码学习1.4——MapOutputTracker.scala

相关类:MapOutputTrackerMessage,GetMapOutputStatuses extends MapPutputTrackerMessage,StopMapOutputTracker extends MapOutputTrackerMessage,MapOutputTrackerMasterActor,MapOutputTrackerMaster. 首先重写MapOutputTrackerMasterActor的receiveWithLogging:判断MapOutputTr

Spark源码学习1.5——BlockManager.scala

一.BlockResult类 该类用来表示返回的匹配的block及其相关的参数.共有三个参数: data:Iterator [Any]. readMethod: DataReadMethod.Value. bytes: Long. 实例化InputMetrics类. 二.BlockManager类 关系的参数有executorId,blockManagerMaster,mapOutPutTracker等.关系的类有ShuffleBlockManager类,DiskBlockManager类,Co

Spark源码学习和总结1

一. Spark源码编译 解析: wget http://archive.apache.org/dist/spark/spark-1.6.0/spark-1.6.0-bin-hadoop2.6.tgz gtar -zxvf spark-1.6.0-bin-hadoop2.6.tgz cd spark-1.6.0-bin-hadoop2.6 ./sbt/sbt gen-idea 说明:漫长的等待后,上述命令生成SBT工程,我们就可以使用IDEA以SBT Project的方式打开. 二. RDD实现

Spark源码学习2

转自:http://www.cnblogs.com/hseagle/p/3673123.html 在源码阅读时,需要重点把握以下两大主线. 静态view 即 RDD, transformation and action 动态view 即 life of a job, 每一个job又分为多个stage,每一个stage中可以包含多个rdd及其transformation,这些stage又是如何映射成为task被distributed到cluster中 一.概要 本文以wordCount为例,详细说

spark源码学习-withScope

withScope是最近的发现版中新增加的一个模块,它是用来做DAG可视化的(DAG visualization on SparkUI) 以前的sparkUI中只有stage的执行情况,也就是说我们不可以看到上个RDD到下个RDD的具体信息.于是为了在 sparkUI中能展示更多的信息.所以把所有创建的RDD的方法都包裹起来,同时用RDDOperationScope 记录 RDD 的操作历史和关联,就能达成目标.下面就是一张WordCount的DAG visualization on Spark

Spark源码学习和总结2

2. SchedulerBackend SchedulerBackend是一个trait,主要的功能是向当前等待分配计算资源的Task分配计算资源Executor,并且在分配的Executor上启动Task,完成计算的调度过程.上述任务调度是通过方法def reviveOffers(): Unit实现的.需要说明的是CoarseGrainedSchedulerBackend是SchedulerBackend的一个具体实现,而YARN.Standalone和Mesos都是基于SchedulerBa

大数据Spark蘑菇云前传第15课:Scala类型参数编程实战及Spark源码鉴赏(学习笔记)

前传第15课:Scala类型参数编程实战及Spark源码鉴赏 本課課程: Spark源码中的Scala类型系統的使用 Scala类型系統编程操作实战 Spark源码中的Scala类型系統的使用 classOf[RDD[_]] 這個也是类型系統 這里的意思是說 B 這種類型必需至少是 A 這樣類型 Ordering Scala类型系統编程操作实战 作為類型系統最大的就可以對類型進行限制,在Scala 中的類型系統,他本身也作為對象.e.g. 我們可以建立 Person 這個類,現在可以建立一個什麼