Spark Core源码分析: Spark任务执行模型

简介:

DAGScheduler

面向stage的调度层,为job生成以stage组成的DAG,提交TaskSet给TaskScheduler执行。

每一个Stage内,都是独立的tasks,他们共同执行同一个compute function,享有相同的shuffledependencies。DAG在切分stage的时候是依照出现shuffle为界限的。

private[spark]
class DAGScheduler(
    taskScheduler: TaskScheduler,
    listenerBus: LiveListenerBus,
    mapOutputTracker: MapOutputTrackerMaster,
    blockManagerMaster: BlockManagerMaster,
    env: SparkEnv)
  extends Logging {

// Actor模式收取发来的DAGSchedulerEvent,并进行processEvent
private var eventProcessActor: ActorRef = _

  private[scheduler] val nextJobId = new AtomicInteger(0)
  private[scheduler] def numTotalJobs: Int = nextJobId.get()
  private val nextStageId = new AtomicInteger(0)

  // 一系列信息维护,很清晰
  private[scheduler] val jobIdToStageIds = new HashMap[Int, HashSet[Int]]
  private[scheduler] val stageIdToJobIds = new HashMap[Int, HashSet[Int]]
  private[scheduler] val stageIdToStage = new HashMap[Int, Stage]
  private[scheduler] val shuffleToMapStage = new HashMap[Int, Stage]
  private[scheduler] val jobIdToActiveJob = new HashMap[Int, ActiveJob]
  private[scheduler] val resultStageToJob = new HashMap[Stage, ActiveJob]
  private[scheduler] val stageToInfos = new HashMap[Stage, StageInfo]

  // 不同状态stages的维护,很清晰
  // Stages we need to run whose parents aren't done
  private[scheduler] val waitingStages = new HashSet[Stage]

  // Stages we are running right now
  private[scheduler] val runningStages = new HashSet[Stage]

  // Stages that must be resubmitted due to fetch failures
  private[scheduler] val failedStages = new HashSet[Stage]

  // Missing tasks from each stage
  private[scheduler] val pendingTasks = new HashMap[Stage, HashSet[Task[_]]]

  private[scheduler] val activeJobs = new HashSet[ActiveJob]

  // Contains the locations that each RDD's partitions are cached on
  private val cacheLocs = new HashMap[Int, Array[Seq[TaskLocation]]]

在start()方法中会初始化Actor,然后接收DAGSchedulerEvent处理。Scheduler会在SparkContext里start起来。


Event处理

源码的阅读入口:可以根据processEvent(event:DAGSchedulerEvent)方法展开。

处理的事件包括这么一些:



Submit Job

JobSubmitted事件:



提交任务的事件传入参数如下

case JobSubmitted(jobId, rdd, func, partitions, allowLocal, callSite, listener, properties)

处理过程可以拆成三步看,每一步里面涉及的具体逻辑在下面补充展开


finalStage = newStage(rdd, partitions.size, None, jobId, Some(callSite))

本次newStage()操作可以对应新的result stage或者shuffle stage。返回Stage类(里面记录一些信息)。Stage类会传入Option[ShuffleDependency[_,_]]参数,内部有一个isShuffleMap变量,以标识该Stage是shuffle or result。


val job = new ActiveJob(jobId, finalStage, func, partitions, callSite, listener, properties)

ActiveJob类也是记录一些信息的类,可以当作是一个VO类


if (allowLocal && finalStage.parents.size == 0 && partitions.length == 1) {
// Compute very short actions like first() or take() 
// with no parent stages locally.
listenerBus.post(SparkListenerJobStart(
job.jobId, Array[Int](), properties))
runLocally(job)
} else {
jobIdToActiveJob(jobId) = job
activeJobs += job
resultStageToJob(finalStage) = job
listenerBus.post(
SparkListenerJobStart(
job.jobId, jobIdToStageIds(jobId).toArray, properties))
submitStage(finalStage)
}

首先判断stage没有父亲依赖,且partition为1的话,就执行本地任务。否则,submitStage。

 

submitStage的逻辑为,首先寻找本次stage的parents。如果没有missing的parent stage,那么就submitMissingTask,即提交本次stage的tasks了。如果有,会对parent stage进行递归submitStage,而且getMissingParentStages得到的结果集是按id降序排的,也就是说递归submitStage的时候会按parent stage的id顺序进行。

 

submitMissingTask处理的是stage的parent已经available的stage。主要逻辑如下:

第一步:通过stage.isShuffleMap来决定生成ShuffleMapTask还是ResultTask,生成的ShuffleMapTask数目和partition数目相等。

第二步:把生成的tasks组建成一个TaskSet,提交给TaskScheduler的submitTasks方法。


TaskScheduler

DAGScheduler以stage为单位,提tasks给TaskScheduer,实现类为TaskSchedulerImpl。

 

TaskSchedulerImpl几个内部部件:

SchedulerBackend

SchedulableBuilder

DAGScheduler

TaskSetManager

TaskResultGetter

Tasks信息(taskIdToTaskSetId,taskIdToExecutorId,activeExecutorIds)

别的信息(SchedulerMode)

 

TaskScheduler做接收task、接收分到的资源和executor、维护信息、与backend打交道、把任务分配好等事情。

 

start(),stop()的时候,backend的start(),stop()


submitTasks(TaskSet)逻辑:

为这批Task生成新的TaskSetManager,把TaskSetManager加到SchedulerBuilder里,然后向backend进行一次reviveOffer()操作。


SchedulerBuilder

SchedulableBuilder有FIFO和Fair两种实现, addTaskSetManager会把TaskSetManager加到pool里。FIFO的话只有一个pool。Fair有多个pool,Pool也分FIFO和Fair两种模式。

SchedulableBuilder的rootPool里面可以新增pool或者TaskSetManager,两者都是Scheduable的继承类,所以SchedulableBuilder用于维护rootPool这棵Scheduable 树结构。Pool是树上的非叶子节点,而TaskSetManager就是叶子节点。

在TaskScheduler初始化的时候会buildDafaultPool。


                            


TaskSetManager

TaskSetManager负责这批Tasks的启动,失败重试,感知本地化等事情。每次reourseOffer方法会寻找合适(符合条件execId, host, locality)的Task并启动它。

 

reourseOffer方法,

  def resourceOffer(
      execId: String,
      host: String,
      maxLocality: TaskLocality.TaskLocality)

寻找符合execId, host和locality的task,找到的话就启动这个Task。启动的时候,把task加到runningTask的HashSet里,然后调DAGScheduler的taskStarted方法,taskStarted方法向eventProcessorActor发出BeginEvent的DAGSchedulerEvent。


TaskResultGetter

维护一个线程池,用来反序列化和从远端获取task结果。


def enqueueSuccessfulTask(taskSetManager: TaskSetManager, tid: Long, serializedData: ByteBuffer)

把序列化的数据反序列化解析出来之后,有两种情况:直接可读的result和间接task result。


前者是DirectTaskResult[T]类:

class DirectTaskResult[T](var valueBytes: ByteBuffer, var accumUpdates: Map[Long, Any], var metrics: TaskMetrics)

后者是IndirectTaskResult[T]类:

case class IndirectTaskResult[T](blockId: BlockId) extends TaskResult[T] with Serializable

在反解析出IndirectTaskResult后,可以得到BlockId这个类,他的实现有这么些:


在TaskResultGetter里,会通过blockManager的getRemoteBytes(BlockId)方法来获得序列化的task result,对这个task result进行反解析后得到DirectTaskResult类,从而获得反序列化后的真正结果数据。

这是大致的一个过程,具体还有一些细节在之中,比如会向scheduler发送不同的event、blockManager会调用BlockManagerMaster把该Block remove掉。

 

BlockId类有这么些关键变量:

private[spark] sealed abstract class BlockId {
  /** A globally unique identifier for this Block. Can be used for ser/de. */
  def name: String

  // convenience methods
  def asRDDId = if (isRDD) Some(asInstanceOf[RDDBlockId]) else None
  def isRDD = isInstanceOf[RDDBlockId]
  def isShuffle = isInstanceOf[ShuffleBlockId]
  def isBroadcast = isInstanceOf[BroadcastBlockId]

下面看BlockManager如何通过BlockId获得数据:

调用的是BlockManager的内部方法

private def doGetRemote(blockId: BlockId, asValues: Boolean): Option[Any] = {
require(blockId != null, "BlockId is null")
// 通过BlockManagerMaster获得这个blockId的locations
    val locations = Random.shuffle(master.getLocations(blockId))
    for (loc <- locations) {
      logDebug("Getting remote block " + blockId + " from " + loc)
      // 使用BlockManagerWorker来获得block的数据
      val data = BlockManagerWorker.syncGetBlock(
        GetBlock(blockId), ConnectionManagerId(loc.host, loc.port))
      if (data != null) {
        if (asValues) {
          // 取到就返回
          return Some(dataDeserialize(blockId, data))
        } else {
          return Some(data)
        }
      }
      logDebug("The value of block " + blockId + " is null")
    }
    logDebug("Block " + blockId + " not found")
    None
  }

思路是通过BlockManagerMaster来获得block的位置信息,得到的集合打乱后,遍历位置信息,通过BlockManagerWorker去获得数据,只要得到了,就反序列化之后返回。

 

在TaskResultGetter处理的时候,成功和失败分别向Scheduler调用handleSuccessfulTask和handleFailedTask方法。

handleSuccessfulTask在DAGScheduler里,会发出CompletionEvent事件,这一步结尾工作会有很多细节处理,这里先不阅读了。

handleFailedTask的话,只要TaskSetManager不是zombie,task没有被kill,那么会继续调用backend.reviveOffers()来re-run。



全文完 :)


目录
相关文章
|
2月前
|
分布式计算 监控 Spark
Spark 任务运行时日志分析
Spark 任务运行时日志分析
24 0
|
2月前
|
设计模式 SQL 分布式计算
Spark Day06:Spark Core之Spark 内核调度和SparkSQL快速入门
Spark Day06:Spark Core之Spark 内核调度和SparkSQL快速入门
39 0
|
2月前
|
存储 分布式计算 负载均衡
【大数据技术Hadoop+Spark】MapReduce概要、思想、编程模型组件、工作原理详解(超详细)
【大数据技术Hadoop+Spark】MapReduce概要、思想、编程模型组件、工作原理详解(超详细)
36 0
|
2月前
|
分布式计算 Java 数据库连接
回答粉丝疑问:Spark为什么调优需要降低过多小任务,降低单条记录的资源开销?
回答粉丝疑问:Spark为什么调优需要降低过多小任务,降低单条记录的资源开销?
26 1
|
2月前
|
分布式计算 监控 分布式数据库
Spark Day05:Spark Core之Sougou日志分析、外部数据源和共享变量
Spark Day05:Spark Core之Sougou日志分析、外部数据源和共享变量
38 0
|
2月前
|
存储 缓存 分布式计算
【Spark】Spark Core Day04
【Spark】Spark Core Day04
29 1
|
4月前
|
SQL 分布式计算 算法
【大数据处理框架】Spark大数据处理框架,包括其底层原理、架构、编程模型、生态圈
【大数据处理框架】Spark大数据处理框架,包括其底层原理、架构、编程模型、生态圈
115 0
|
5月前
|
分布式计算 搜索推荐 算法
大数据Spark MLlib基于模型的协同过滤
大数据Spark MLlib基于模型的协同过滤
47 0
|
6月前
|
分布式计算 监控 Java
Spark学习---7、Spark内核(源码提交流程、任务执行、Shuffle、内存管理)(一)
Spark学习---7、Spark内核(源码提交流程、任务执行、Shuffle、内存管理)(一)
|
8月前
|
分布式计算 Spark
SPARK outputDeterministicLevel的作用--任务全部重试或者部分重试
SPARK outputDeterministicLevel的作用--任务全部重试或者部分重试
39 0