摘要

本文目录

spark的RDD的逻辑执行链

----------------------------------------------

----------------------------------------------

----------------------------------------------

----------------------------------------------

----------------------------------------------

----------------------------------------------

----------------------------------------------

----------------------------------------------

----------------------------------------------

• 形成ResultStage依赖的父Stage
  */
  private def getOrCreateParentStages(rdd: RDD[_], firstJobId: Int): List[Stage] = {
  getShuffleDependencies(rdd).map { shuffleDep =>
  getOrCreateShuffleMapStage(shuffleDep, firstJobId)
  }.toList
  }
  /**
• 采用的是深度优先遍历找到Action算子的父依赖中的宽依赖
• 这个是最主要的方法，要看懂这个方法，其实后面的就好理解，最好结合这例子上面给出的RDD逻辑依赖图，比*
•   较容易看出来，根据上面的RDD逻辑依赖图，其返回的ShuffleDependency就是RDD2和RDD1，RDD7和RDD6的依
  赖,如果存在A<-B<-C,这两个都是shuffle依赖，那么对于C其只返回B的shuffle依赖，而不会返回A
  /
  private[scheduler] def getShuffleDependencies(
  rdd: RDD[]): HashSet[ShuffleDependency[, , ]] = {
  //用来存放依赖
  val parents = new HashSet[ShuffleDependency[, , ]]
  //遍历过的RDD放入这个里面
  val visited = new HashSet[RDD[]]
  //创建一个待遍历RDD的栈结构
  val waitingForVisit = new ArrayStack[RDD[]]
  //压入finalRDD，逻辑图中的RDD9
  waitingForVisit.push(rdd)
  //循环遍历这个栈结构
  while (waitingForVisit.nonEmpty) {
  val toVisit = waitingForVisit.pop()
  // 如果RDD没有被遍历过执行其中的代码
  if (!visited(toVisit)) {
  //然后把其放入已经遍历队列中
  visited += toVisit
  //得到依赖，我们知道依赖中存放的有父RDD的对象
  toVisit.dependencies.foreach {
  //如果这个依赖是shuffle依赖，则放入返回队列中
  case shuffleDep: ShuffleDependency[, , ] =>
  parents += shuffleDep
  case dependency =>
  //如果不是shuffle依赖，把其父RDD压入待访问栈中，从而进行循环
  waitingForVisit.push(dependency.rdd)
  }
  }
  }
  parents
  }
  /创建shuffleMapStage，根据上面得到的两个Shuffle对象，分别创建了两个shuffleMapStage
  /
  /
  def createShuffleMapStage(shuffleDep: ShuffleDependency[, , _], jobId: Int): ShuffleMapStage = {
  //这个RDD其实就是RDD1和RDD6
  val rdd = shuffleDep.rdd
  val numTasks = rdd.partitions.length
  val parents = getOrCreateParentStages(rdd, jobId) //查看这两个ShuffleMapStage是否存在父Shuffle的Stage
  val id = nextStageId.getAndIncrement()
  //创建ShuffleMapStage，下面是更新一下SparkContext的状态
  val stage = new ShuffleMapStage(
  id, rdd, numTasks, parents, jobId, rdd.creationSite, shuffleDep, mapOutputTracker)
  stageIdToStage(id) = stage
  shuffleIdToMapStage(shuffleDep.shuffleId) = stage
  updateJobIdStageIdMaps(jobId, stage)
  
       if (!mapOutputTracker.containsShuffle(shuffleDep.shuffleId)) {
  // Kind of ugly: need to register RDDs with the cache and map output tracker here
  // since we can't do it in the RDD constructor because # of partitions is unknown
  logInfo("Registering RDD " + rdd.id + " (" + rdd.getCreationSite + ")")
  mapOutputTracker.registerShuffle(shuffleDep.shuffleId, rdd.partitions.length)
  }
  stage
  }
  

  ----------------------------------------------

  通过上面的源代码分析，结合RDD的逻辑执行图，我们可以看出，这个job拥有三个Stage，一个ResultStage，两个ShuffleMapStage，一个ShuffleMapStage中的RDD是RDD1，另一个stage中的RDD是RDD6,从而，以上完成了RDD到Stage的切分工作。当切分完成后在handleJobSubmitted这个方法的最后，调用提交stage的方法。
  

----------------------------------------------

----------------------------------------------

----------------------------------------------

----------------------------------------------

spark的TaskSchedule的调度

----------------------------------------------

----------------------------------------------

----------------------------------------------

• 然后其会调用TaskSchedulerImpl中的resourceOffers方法，来进行筛选，筛选出符合请求资源的机器，来执行我们当前的任务
  /
  private def makeOffers() {
  // Filter out executors under killing
  val activeExecutors = executorDataMap.filterKeys(executorIsAlive)
  val workOffers = activeExecutors.map { case (id, executorData) =>
  new WorkerOffer(id, executorData.executorHost, executorData.freeCores)
  }.toIndexedSeq
  launchTasks(scheduler.resourceOffers(workOffers))
  }

•   /
  def resourceOffers(offers: IndexedSeq[WorkerOffer]): Seq[Seq[TaskDescription]] = synchronized {
  // Mark each slave as alive and remember its hostname
  // Also track if new executor is added
  var newExecAvail = false
  //检查work是否已经存在了，把不存在的加入到work调度池中
  for (o <- offers) {
  if (!hostToExecutors.contains(o.host)) {
  hostToExecutors(o.host) = new HashSet[String]()
  }
  if (!executorIdToRunningTaskIds.contains(o.executorId)) {
  hostToExecutors(o.host) += o.executorId
  executorAdded(o.executorId, o.host)
  executorIdToHost(o.executorId) = o.host
  executorIdToRunningTaskIds(o.executorId) = HashSet[Long]()
  newExecAvail = true
  }
  for (rack <- getRackForHost(o.host)) {
  hostsByRack.getOrElseUpdate(rack, new HashSet[String]()) += o.host
  }
  }
  // 打乱work机器的顺序，以免每次分配任务时都在同一个机器上进行。避免某一个work计算压力太大。
  val shuffledOffers = Random.shuffle(offers)
  //对于每一work，创建一个与其核数大小相同的数组，数组的大小决定了这台work上可以并行执行task的数目.
  val tasks = shuffledOffers.map(o => new ArrayBufferTaskDescription)
  //取出每台机器的cpu核数
  val availableCpus = shuffledOffers.map(o => o.cores).toArray
  //从task任务调度池中，按照我们的调度算法，取出需要执行的任务
  val sortedTaskSets = rootPool.getSortedTaskSetQueue
  for (taskSet <- sortedTaskSets) {
  logDebug("parentName: %s, name: %s, runningTasks: %s".format(
  taskSet.parent.name, taskSet.name, taskSet.runningTasks))
  if (newExecAvail) {
  taskSet.executorAdded()
  }
  }
  // 下面的这个循环，是用来标记task根据work的信息来标定数据本地化的程度的。当我们在yarn资源管理器，以--driver-mode配置
  //为client时，我们就会在打出来的日志上看出每一台机器上运行task的数据本地化程度。同时还会选择每个task对应的work机器
  // NOTE: the preferredLocality order: PROCESS_LOCAL, NODE_LOCAL, NO_PREF, RACK_LOCAL, ANY
  for (taskSet <- sortedTaskSets) {
  var launchedAnyTask = false
  var launchedTaskAtCurrentMaxLocality = false
  for (currentMaxLocality <- taskSet.myLocalityLevels) {
  do {
  launchedTaskAtCurrentMaxLocality = resourceOfferSingleTaskSet(
  taskSet, currentMaxLocality, shuffledOffers, availableCpus, tasks)
  launchedAnyTask |= launchedTaskAtCurrentMaxLocality
  } while (launchedTaskAtCurrentMaxLocality)
  }
  if (!launchedAnyTask) {
  taskSet.abortIfCompletelyBlacklisted(hostToExecutors)
  }
  }
  
       if (tasks.size > 0) {
  hasLaunchedTask = true
  }
  //返回taskDescription对象
  return tasks
  }
  

•   /
  private def resourceOfferSingleTaskSet(
  taskSet: TaskSetManager,
  maxLocality: TaskLocality,
  shuffledOffers: Seq[WorkerOffer],
  availableCpus: Array[Int],
  tasks: IndexedSeq[ArrayBuffer[TaskDescription]]) : Boolean = {
  var launchedTask = false
  //循环所有的机器，找适合此机器的task
  for (i <- 0 until shuffledOffers.size) {
  val execId = shuffledOffers(i).executorId
  val host = shuffledOffers(i).host
  //判断其剩余的cpu核数是否满足我们的最低配置，满足则为其分配任务，否则不为其分配任务。
  if (availableCpus(i) >= CPUS_PER_TASK) {
  try {
  //这个for中的resourOffer就是来判断其标记任务数据本地化的程度的。task(i)其实是一个数组，数组大小和其cpu核心数大小相同。
  for (task <- taskSet.resourceOffer(execId, host, maxLocality)) {
  tasks(i) += task
  val tid = task.taskId
  taskIdToTaskSetManager(tid) = taskSet
  taskIdToExecutorId(tid) = execId
  executorIdToRunningTaskIds(execId).add(tid)
  availableCpus(i) -= CPUS_PER_TASK
  assert(availableCpus(i) >= 0)
  launchedTask = true
  }
  } catch {
  case e: TaskNotSerializableException =>
  logError(s"Resource offer failed, task set ${taskSet.name} was not serializable")
  // Do not offer resources for this task, but don't throw an error to allow other
  // task sets to be submitted.
  return launchedTask
  }
  }
  }
  return launchedTask
  }
  

  ----------------------------------------------

  以上完成了从TaskSet到task和work机器的绑定过程的所有任务。下面就是如何发送task到executor进行执行。在makeOffers()方法中调用了launchTasks方法,这个方法其实就是发送task作业到指定的机器上。只此，spark TaskSchedule的调度就此结束。
  

executor如何执行task以及我们定义的函数

----------------------------------------------

•   /
  private def run(
  driverUrl: String,
  executorId: String,
  hostname: String,
  cores: Int,
  appId: String,
  workerUrl: Option[String],
  userClassPath: Seq[URL]) {
  
       Utils.initDaemon(log)
  
       SparkHadoopUtil.get.runAsSparkUser { () =>
  // Debug code
  Utils.checkHost(hostname)
  
       // Bootstrap to fetch the driver's Spark properties.
  val executorConf = new SparkConf
  val port = executorConf.getInt("spark.executor.port", 0)
  val fetcher = RpcEnv.create(
  "driverPropsFetcher",
  hostname,
  port,
  executorConf,
  new SecurityManager(executorConf),
  clientMode = true)
  val driver = fetcher.setupEndpointRefByURI(driverUrl)
  val cfg = driver.askWithRetrySparkAppConfig
  val props = cfg.sparkProperties ++ Seq[(String, String)](("spark.app.id", appId))
  fetcher.shutdown()
  
       // Create SparkEnv using properties we fetched from the driver.
  val driverConf = new SparkConf()
  for ((key, value) <- props) {
  // this is required for SSL in standalone mode
  if (SparkConf.isExecutorStartupConf(key)) {
  driverConf.setIfMissing(key, value)
  } else {
  driverConf.set(key, value)
  }
  }
  if (driverConf.contains("spark.yarn.credentials.file")) {
  logInfo("Will periodically update credentials from: " +
  driverConf.get("spark.yarn.credentials.file"))
  SparkHadoopUtil.get.startCredentialUpdater(driverConf)
  }
  
       val env = SparkEnv.createExecutorEnv(
  driverConf, executorId, hostname, port, cores, cfg.ioEncryptionKey, isLocal = false)
  
       env.rpcEnv.setupEndpoint("Executor", new CoarseGrainedExecutorBackend(
  env.rpcEnv, driverUrl, executorId, hostname, cores, userClassPath, env))
  workerUrl.foreach { url =>
  env.rpcEnv.setupEndpoint("WorkerWatcher", new WorkerWatcher(env.rpcEnv, url))
  }
  env.rpcEnv.awaitTermination()
  SparkHadoopUtil.get.stopCredentialUpdater()
  }
  }
  

  ----------------------------------------------

  其执行函数的调用过程如下：
  
  

----------------------------------------------

----------------------------------------------

----------------------------------------------

----------------------------------------------

----------------------------------------------

----------------------------------------------

----------------------------------------------

----------------------------------------------