Spark 源码分析 -- Stage
2023-03-14 22:35:46 时间
理解stage, 关键就是理解Narrow Dependency
和Wide Dependency
, 可能还是觉得比较难理解
关键在于是否需要shuffle, 不需要shuffle是可以随意并发的, 所以stage的边界就是需要shuffle的地方, 如下图很清楚
并且Stage分为两种,
shuffle map stage, in which case its tasks' results are input for another stage
其实就是,非最终stage, 后面还有其他的stage, 所以它的输出一定是需要shuffle并作为后续的输入
result stage, in which case its tasks directly compute the action that initiated a job (e.g. count(), save(), etc)
最终的stage, 没有输出, 而是直接产生结果或存储
1 stage class
这个注释写的很清楚
可以看到stage的RDD参数只有一个RDD, final RDD, 而不是一系列的RDD
因为在一个stage中的所有RDD都是map, partition不会有任何改变, 只是在data依次执行不同的map function
所以对于task scheduler而言, 一个RDD的状况就可以代表这个stage
/** * A stage is a set of independent tasks all computing the same function that need to run as part * of a Spark job, where all the tasks have the same shuffle dependencies. Each DAG of tasks run * by the scheduler is split up into stages at the boundaries where shuffle occurs, and then the * DAGScheduler runs these stages in topological order. * * Each Stage can either be a shuffle map stage, in which case its tasks' results are input for * another stage, or a result stage, in which case its tasks directly compute the action that * initiated a job (e.g. count(), save(), etc). For shuffle map stages, we also track the nodes * that each output partition is on. * * Each Stage also has a jobId, identifying the job that first submitted the stage. When FIFO * scheduling is used, this allows Stages from earlier jobs to be computed first or recovered * faster on failure. */ private[spark] class Stage( val id: Int, val rdd: RDD[_], // final RDD val shuffleDep: Option[ShuffleDependency[_,_]], // Output shuffle if stage is a map stage val parents: List[Stage], // 父stage val jobId: Int, callSite: Option[String]) extends Logging { val isShuffleMap = shuffleDep != None // 是否是shuffle map stage, 取决于是否有shuffleDep val numPartitions = rdd.partitions.size val outputLocs = Array.fill[List[MapStatus]](numPartitions)(Nil) // 用于buffer每个shuffle中每个maptask的MapStatus var numAvailableOutputs = 0 private var nextAttemptId = 0
def isAvailable: Boolean = { if (!isShuffleMap) { true } else { numAvailableOutputs == numPartitions } }
}
2 newStage
如果是shuffle map stage, 需要在这里向mapOutputTracker注册shuffle
/** * Create a Stage for the given RDD, either as a shuffle map stage (for a ShuffleDependency) or * as a result stage for the final RDD used directly in an action. The stage will also be * associated with the provided jobId. */ private def newStage( rdd: RDD[_], shuffleDep: Option[ShuffleDependency[_,_]], jobId: Int, callSite: Option[String] = None) : Stage = { if (shuffleDep != None) { // 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.origin + ")") mapOutputTracker.registerShuffle(shuffleDep.get.shuffleId, rdd.partitions.size) } val id = nextStageId.getAndIncrement() val stage = new Stage(id, rdd, shuffleDep, getParentStages(rdd, jobId), jobId, callSite) stageIdToStage(id) = stage stageToInfos(stage) = StageInfo(stage) stage }
3 getMissingParentStages
可以根据final stage的deps找出所有的parent stage
private def getMissingParentStages(stage: Stage): List[Stage] = { val missing = new HashSet[Stage] val visited = new HashSet[RDD[_]] def visit(rdd: RDD[_]) { if (!visited(rdd)) { visited += rdd if (getCacheLocs(rdd).contains(Nil)) { for (dep <- rdd.dependencies) { dep match { case shufDep: ShuffleDependency[_,_] => // 如果发现ShuffleDependency, 说明遇到新的stage val mapStage = getShuffleMapStage(shufDep, stage.jobId) // check shuffleToMapStage, 如果该stage已经被创建则直接返回, 否则newStage if (!mapStage.isAvailable) { missing += mapStage } case narrowDep: NarrowDependency[_] => // 对于NarrowDependency, 说明仍然在这个stage中 visit(narrowDep.rdd) } } } } } visit(stage.rdd) missing.toList }
本文章摘自博客园,原文发布日期:2013-12-26
相关文章
- 八个步骤让你的企业“数据化”
- 给你的 Linux 桌面披上 Windows 8 的外衣
- Bash脚本15分钟进阶教程
- SCL项目:在Red Hat系列系统上安装同一软件的不同版本
- 2017年即将过半 免费OA系统行业OA厂商大转变
- 敢偷用我的WiFi? 看我怎么治你
- 如何在Linux命令行下查询字典?
- Xperience UI 设计理念:优雅的 Linux 桌面设计欣赏
- 用于记录一些东西,避免遗忘
- Linux硬件信息命令大全
- 值得关注:Ubuntu 14.04服务器版提供了虚拟化、自动化、存储相关更新
- How AI can fight the phenomenon of fake news
- 如何关闭一个开放的DNS解析器
- 五个你可能不了解的killall选项
- 如何在Linux中“快速阅读”?
- 图解css3:核心技术与案例实战. 2.2 基本选择器
- rhel源更换为centos源
- 图解css3:核心技术与案例实战. 1.2 浏览器对CSS3的支持状况
- 图解css3:核心技术与案例实战. 1.3 渐进增强
- 如何给任意一款 Linux 发行版添加漂亮的字体