马上注册,结交更多数据大咖,获取更多知识干货,轻松玩转大数据
您需要 登录 才可以下载或查看,没有帐号?立即注册
x
前两篇文章写了Shuffle Read的一些实现细节。但是要想彻底理清楚这里边的实现逻辑,还是需要更多篇幅的;本篇开始,将按照Job的执行顺序,来讲解Shuffle。即,结果数据(ShuffleMapTask的结果和ResultTask的结果)是如何产生的;结果是如何处理的;结果是如何读取的。 在Worker上接收Task执行命令的是org.apache.Spark.executor.CoarseGrainedExecutorBackend。它在接收到LaunchTask的命令后,通过在Driver创建SparkContext时已经创建的org.apache.spark.executor.Executor的实例的launchTask,启动Task:
deflaunchTask(
context: ExecutorBackend, taskId: Long, taskName: String,serializedTask: ByteBuffer) {
val tr = new TaskRunner(context, taskId, taskName, serializedTask)
runningTasks.put(taskId, tr)
threadPool.execute(tr) // 开始在executor中运行
}
最终Task的执行是在org.apache.spark.executor.Executor.TaskRunner#run。org.apache.spark.executor.ExecutorBackend是Executor与Driver通信的接口,它实际上是一个trait:
private[spark] trait ExecutorBackend {
defstatusUpdate(taskId: Long, state: TaskState, data: ByteBuffer)
}
TaskRunner会将Task执行的状态汇报给Driver(org.apache.spark.scheduler.cluster.CoarseGrainedSchedulerBackend.DriverActor)。 而Driver会转给org.apache.spark.scheduler.TaskSchedulerImpl#statusUpdate。
在Executor运行Task时,得到计算结果会存入org.apache.spark.scheduler.DirectTaskResult。在将结果回传到Driver时,会根据结果的大小有不同的策略:对于“较大”的结果,将其以taskid为key存入org.apache.spark.storage.BlockManager;如果结果不大,那么直接回传给Driver。那么如何判定这个阈值呢? 这里的回传是直接通过akka的消息传递机制。因此这个大小首先不能超过这个机制设置的消息的最大值。这个最大值是通过spark.akka.frameSize设置的,单位是Bytes,默认值是10MB。除此之外,还有200KB的预留空间。因此这个阈值就是conf.getInt("spark.akka.frameSize", 10) * 1024 *1024 – 200KB。
// directSend = sending directly back to the driver
val (serializedResult, directSend) = {
if (resultSize >=akkaFrameSize - AkkaUtils.reservedSizeBytes) { //如果结果太大,那么存入BlockManager
val blockId = TaskResultBlockId(taskId)
env.blockManager.putBytes(
blockId, serializedDirectResult,StorageLevel.MEMORY_AND_DISK_SER)
(ser.serialize(new IndirectTaskResult[Any](blockId)), false)
} else { // 如果大小合适,则直接发送结果给Driver
(serializedDirectResult, true)
}
}
execBackend.statusUpdate(taskId, TaskState.FINISHED, serializedResult)
TaskRunner将Task的执行状态汇报给Driver后,Driver会转给org.apache.spark.scheduler.TaskSchedulerImpl#statusUpdate。而在这里不同的状态有不同的处理: 1. 如果类型是TaskState.FINISHED,那么调用org.apache.spark.scheduler.TaskResultGetter#enqueueSuccessfulTask进行处理。 2. 如果类型是TaskState.FAILED或者TaskState.KILLED或者TaskState.LOST,调用org.apache.spark.scheduler.TaskResultGetter#enqueueFailedTask进行处理。对于TaskState.LOST,还需要将其所在的Executor标记为failed, 并且根据更新后的Executor重新调度。 enqueueSuccessfulTask的逻辑也比较简单,就是如果是IndirectTaskResult,那么需要通过blockid来获取结果:sparkEnv.blockManager.getRemoteBytes(blockId);如果是DirectTaskResult,那么结果就无需远程获取了。然后调用 1. org.apache.spark.scheduler.TaskSchedulerImpl#handleSuccessfulTask 2. org.apache.spark.scheduler.TaskSetManager#handleSuccessfulTask 3. org.apache.spark.scheduler.DAGScheduler#taskEnded 4. org.apache.spark.scheduler.DAGScheduler#eventProcessActor 5. org.apache.spark.scheduler.DAGScheduler#handleTaskCompletion 进行处理。核心逻辑都在第5个调用栈。如果task是ResultTask,处理逻辑比较简单,停止job,更新一些状态,发送一些event即可。
if (!job.finished(rt.outputId)){
job.finished(rt.outputId) =true
job.numFinished += 1
// If the whole job hasfinished, remove it
if (job.numFinished ==job.numPartitions) {
markStageAsFinished(stage)
cleanupStateForJobAndIndependentStages(job)
listenerBus.post(SparkListenerJobEnd(job.jobId,JobSucceeded))
}
// taskSucceeded runs someuser code that might throw an exception.
// Make sure we areresilient against that.
try {
job.listener.taskSucceeded(rt.outputId, event.result)
} catch {
case e: Exception =>
// TODO: Perhaps we wantto mark the stage as failed?
job.listener.jobFailed(new SparkDriverExecutionException(e))
}
}
如果task是ShuffleMapTask,那么它需要将结果通过某种机制告诉下游的Stage,以便于其可以作为下游Stage的输入。这个机制是怎么实现的? 实际上,对于ShuffleMapTask来说,其结果实际上是org.apache.spark.scheduler.MapStatus;其序列化后存入了DirectTaskResult或者IndirectTaskResult中。而DAGScheduler#handleTaskCompletion通过下面的方式来获取这个结果: val status =event.result.asInstanceOf[MapStatus] 通过将这个status注册到org.apache.spark.MapOutputTrackerMaster,就实现了
mapOutputTracker.registerMapOutputs(
stage.shuffleDep.get.shuffleId,
stage.outputLocs.map(list=> if (list.isEmpty) null else list.head).toArray,
changeEpoch = true)
来自群组: Spark精英汇 |