Spark TaskSet提交及执行准备

ゝ一世哀愁。 2022-11-15 15:59 157阅读 0赞

参考文章：[Spark TaskSet提交及执行准备][Spark TaskSet]

一个stage是由一组相同运算的task组成，他们分别计算不同的partition，stage的提交实际是向调度器提交一组包含相同计算的task，这里调度器的名字是TaskScheduler，其调度单位是taskset，stage在提交时使用了递归算法，会先提交没有parent stage的stage，代码如下：

private def submitStage(stage: Stage) {
        val jobId = activeJobForStage(stage)
        if (jobId.isDefined) {
          logDebug("submitStage(" + stage + ")")
          if (!waitingStages(stage) && !runningStages(stage) && !failedStages(stage)) {
            val missing = getMissingParentStages(stage).sortBy(_.id)
            logDebug("missing: " + missing)
            if (missing == Nil) {
              logInfo("Submitting " + stage + " (" + stage.rdd + "), which has no missing parents")
              //这里会提交stage至TaskScheduler
              submitMissingTasks(stage, jobId.get)
            } else {
              for (parent <- missing) {
                submitStage(parent)
              }
              waitingStages += stage
            }
          }
        } else {
          abortStage(stage, "No active job for stage " + stage.id)
        }
      }

stage类型分为shuffleMap和result，stage的rdd会被序列化，ShuffleMapTask会序列化rdd、shuffleDep，ResultTask会序列化rdd和func。序列化后的task会被广播到各节点，等待执行指令。各task会以taskset的形式打包到一起供TaskScheduler调度:

private def submitMissingTasks(stage: Stage, jobId: Int) {
        logDebug("submitMissingTasks(" + stage + ")")
        // 数据结构清理，马上要运行，挂起的task先清理掉
        stage.pendingTasks.clear()
     
     
        // First figure out the indexes of partition ids to compute.
        val partitionsToCompute: Seq[Int] = {
          if (stage.isShuffleMap) {
            (0 until stage.numPartitions).filter(id => stage.outputLocs(id) == Nil)
          } else {
            val job = stage.resultOfJob.get
            (0 until job.numPartitions).filter(id => !job.finished(id))
          }
        }
     
     
        val properties = if (jobIdToActiveJob.contains(jobId)) {
          jobIdToActiveJob(stage.jobId).properties
        } else {
          // this stage will be assigned to "default" pool
          null
        }
     
     
        runningStages += stage
        // SparkListenerStageSubmitted should be posted before testing whether tasks are
        // serializable. If tasks are not serializable, a SparkListenerStageCompleted event
        // will be posted, which should always come after a corresponding SparkListenerStageSubmitted
        // event.
        stage.latestInfo = StageInfo.fromStage(stage, Some(partitionsToCompute.size))
        outputCommitCoordinator.stageStart(stage.id)
        listenerBus.post(SparkListenerStageSubmitted(stage.latestInfo, properties))
     
     
        // TODO: Maybe we can keep the taskBinary in Stage to avoid serializing it multiple times.
        // Broadcasted binary for the task, used to dispatch tasks to executors. Note that we broadcast
        // the serialized copy of the RDD and for each task we will deserialize it, which means each
        // task gets a different copy of the RDD. This provides stronger isolation between tasks that
        // might modify state of objects referenced in their closures. This is necessary in Hadoop
        // where the JobConf/Configuration object is not thread-safe.
        //序列化task，并广播到各节点
        var taskBinary: Broadcast[Array[Byte]] = null
        try {
          // For ShuffleMapTask, serialize and broadcast (rdd, shuffleDep).
          // For ResultTask, serialize and broadcast (rdd, func).
          val taskBinaryBytes: Array[Byte] =
            if (stage.isShuffleMap) {
              closureSerializer.serialize((stage.rdd, stage.shuffleDep.get) : AnyRef).array()
            } else {
              closureSerializer.serialize((stage.rdd, stage.resultOfJob.get.func) : AnyRef).array()
            }
          taskBinary = sc.broadcast(taskBinaryBytes)
        } catch {
          .....
        }
    		//计算task集合创建taskset，传递给scheduler调度
        val tasks: Seq[Task[_]] = if (stage.isShuffleMap) {
          partitionsToCompute.map { id =>
            val locs = getPreferredLocs(stage.rdd, id)
            val part = stage.rdd.partitions(id)
            new ShuffleMapTask(stage.id, taskBinary, part, locs)
          }
        } else {
          val job = stage.resultOfJob.get
          partitionsToCompute.map { id =>
            val p: Int = job.partitions(id)
            val part = stage.rdd.partitions(p)
            val locs = getPreferredLocs(stage.rdd, p)
            new ResultTask(stage.id, taskBinary, part, locs, id)
          }
        }
     
     
        if (tasks.size > 0) {
          logInfo("Submitting " + tasks.size + " missing tasks from " + stage + " (" + stage.rdd + ")")
          stage.pendingTasks ++= tasks
          logDebug("New pending tasks: " + stage.pendingTasks)
          //注意，这里开始调度
          taskScheduler.submitTasks(
            new TaskSet(tasks.toArray, stage.id, stage.newAttemptId(), stage.jobId, properties))
          stage.latestInfo.submissionTime = Some(clock.getTimeMillis())
        } else {
          // Because we posted SparkListenerStageSubmitted earlier, we should mark
          // the stage as completed here in case there are no tasks to run
          markStageAsFinished(stage, None)
          logDebug("Stage " + stage + " is actually done; %b %d %d".format(
            stage.isAvailable, stage.numAvailableOutputs, stage.numPartitions))
        }
      }
      task的触发实际是通过akka向woker的actor发送了一个指令
        override def submitTasks(taskSet: TaskSet) {
        val tasks = taskSet.tasks
        logInfo("Adding task set " + taskSet.id + " with " + tasks.length + " tasks")
        this.synchronized {
          val manager = createTaskSetManager(taskSet, maxTaskFailures)
          activeTaskSets(taskSet.id) = manager
          schedulableBuilder.addTaskSetManager(manager, manager.taskSet.properties)
     
     
          if (!isLocal && !hasReceivedTask) {
            starvationTimer.scheduleAtFixedRate(new TimerTask() {
              override def run() {
                if (!hasLaunchedTask) {
                  logWarning("Initial job has not accepted any resources; " +
                    "check your cluster UI to ensure that workers are registered " +
                    "and have sufficient resources")
                } else {
                  this.cancel()
                }
              }
            }, STARVATION_TIMEOUT, STARVATION_TIMEOUT)
          }
          hasReceivedTask = true
        }
        backend.reviveOffers()
      }

本地模式的运行逻辑如下,具体的处理逻辑位于LocalBAckend中：

override def reviveOffers() {
        localActor ! ReviveOffers //发送指令开始执行
      }

def reviveOffers() {
        val offers = Seq(new WorkerOffer(localExecutorId, localExecutorHostname, freeCores))
        val tasks = scheduler.resourceOffers(offers).flatten
        for (task <- tasks) {
        	//更新资源的统计信息并发送launchTask指令
          freeCores -= scheduler.CPUS_PER_TASK
          executor.launchTask(executorBackend, taskId = task.taskId, attemptNumber = task.attemptNumber,
            task.name, task.serializedTask)
        }
        if (tasks.isEmpty && scheduler.activeTaskSets.nonEmpty) {
          // Try to reviveOffer after 1 second, because scheduler may wait for locality timeout
          context.system.scheduler.scheduleOnce(1000 millis, self, ReviveOffers)
        }
      }

如果是集群模式，稍微复杂些，会由CoarseGrainedSchedulerBackend来接受执行指令，内部封装DriverActor:

def makeOffers() {
          launchTasks(scheduler.resourceOffers(executorDataMap.map { case (id, executorData) =>
            new WorkerOffer(id, executorData.executorHost, executorData.freeCores)
          }.toSeq))
        }

// Launch tasks returned by a set of resource offers
      def launchTasks(tasks: Seq[Seq[TaskDescription]]) {
          for (task <- tasks.flatten) {
            val ser = SparkEnv.get.closureSerializer.newInstance()
            val serializedTask = ser.serialize(task)
            if (serializedTask.limit >= akkaFrameSize - AkkaUtils.reservedSizeBytes) {
              val taskSetId = scheduler.taskIdToTaskSetId(task.taskId)
              scheduler.activeTaskSets.get(taskSetId).foreach { taskSet =>
                try {
                  var msg = "Serialized task %s:%d was %d bytes, which exceeds max allowed: " +
                    "spark.akka.frameSize (%d bytes) - reserved (%d bytes). Consider increasing " +
                    "spark.akka.frameSize or using broadcast variables for large values."
                  msg = msg.format(task.taskId, task.index, serializedTask.limit, akkaFrameSize,
                    AkkaUtils.reservedSizeBytes)
                  taskSet.abort(msg)
                } catch {
                  case e: Exception => logError("Exception in error callback", e)
                }
              }
            }
            else {
            	//向executor发送task启动指令
              val executorData = executorDataMap(task.executorId)
              executorData.freeCores -= scheduler.CPUS_PER_TASK
              executorData.executorActor ! LaunchTask(new SerializableBuffer(serializedTask))
            }
          }
        }

[Spark TaskSet]: https://blog.csdn.net/lihm0_1/article/details/49785591