Spark 资源调度包 stage 类解析
spark 资源调度包 Stage(阶段) 类解析
类注释:
/** * A stage is a set of parallel 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. * 一个阶段是所有计算相同功能的并行任务集合, 作为spark作业的一部分, 这些任务都有相同的 shuffle 依赖 * * 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. * 每个由调度器运行的任务的有向无环图在shuffle发生的分界处分化成不同的阶段, 并且这些有向无环图的调度器 * 将以拓扑排序来运行这些 不同阶段的任务 * * Each Stage can either be a shuffle map stage, in which case its tasks' results are input for * other stage(s), or a result stage, in which case its tasks directly compute a Spark action * (e.g. count(), save(), etc) by running a function on an RDD. * 每个阶段都可以是一个 shuffle 匹配阶段, 该任务的结果可以被其他阶段作为导入, 也可以是 一个结果阶段, * 它的任务 通过 在RDD上运行 一个公式 来直接计算 一个 spark 行为(计数, 保存等...) * * For shuffle map stages, we also track the nodes that each output partition is on. * 对于 shuffle 匹配阶段, 我们会追踪 每个输出的分区的节点 * * Each Stage also has a firstJobId, 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. * 每个阶段都有一个 首任务Id, 用于辨识第一个提交到该阶段的 任务。 当 先进先出 调度策略被使用时, 这会允 * 许 更早 的 作业 的阶段先被计算, 或者 在失败之后更早的恢复。 * * Finally, a single stage can be re-executed in multiple attempts due to fault recovery. In * that case, the Stage object will track multiple StageInfo objects to pass to listeners or * the web UI. * 最终, 由于 容错恢复机制, 单个阶段可以在多次尝试运行中 被重新执行。在这种情况下, 该阶段对象会追踪 许 * 多 阶段信息对象 并 将信息传递给监听者 或者 web UI 界面 * * The latest one will be accessible through latestInfo. * 最新的阶段可以通过 最新的信息获取 * * @param id Unique stage ID id: 唯一的阶段Id * @param rdd RDD that this stage runs on: for a shuffle map stage, it's the RDD we run map * tasks on, while for a result stage, it's the target RDD that we ran an action on * rdd:该 阶段 所运行在的 RDD:对于shuffle匹配阶段, 该RDD是我们 运行匹配任务 所在的RDD, 但对于结果阶 * 段,该RDD是我们运行 行为所在的RDD * @param numTasks Total number of tasks in stage; 当前阶段的任务数量 * result stages in particular may not need to compute all partitions, e.g. for first(), * lookup(), and take(). * 结果阶段 实际上不一定需要计算所有分区, 比如使用 first(), lookup() 和 take() 等算子 * @param parents List of stages that this stage depends on (through shuffle dependencies). * parent: 该阶段 所依赖(shuffle依赖) 的 阶段列表 * @param firstJobId ID of the first job this stage was part of, for FIFO scheduling. * firstJobId: 该阶段的第一个任务Id, 用于先进先出调度 * @param callSite Location in the user program associated with this stage: either where the * target RDD was created, for a shuffle map stage, or where the action for a result stage was * called. * callSite: 与该阶段相关的用户程序的位置: 对于 shuffle 匹配阶段 是 目标 RDD创建的位置, 对于 结果 * 阶段, 是 行为被调用的位置 (调用位点) */
代码:
private[scheduler] abstract class Stage( val id: Int, val rdd: RDD[_], val numTasks: Int, val parents: List[Stage], val firstJobId: Int, val callSite: CallSite) extends Logging { // 根据数组长度确定分区数 val numPartitions = rdd.partitions.length /** Set of jobs that this stage belongs to. 该阶段 的任务Id集合*/ val jobIds = new HashSet[Int] /** The ID to use for the next new attempt for this stage. * 该阶段的下一次新尝试的 Id */ private var nextAttemptId: Int = 0 // 短型的调用位点 val name: String = callSite.shortForm // 长型的调用位点 val details: String = callSite.longForm /** * Pointer to the [[StageInfo]] object for the most recent attempt. * 指向阶段信息对象的指针, 用于获取最近最多的尝试 * This needs to be initialized here, before any attempts have actually been created, because * the DAGScheduler uses this StageInfo to tell SparkListeners when a job starts (which * happens before any stage attempts have been created). * 在任何尝试被实际创建之前, 它需要被初始化, 因为 有向无环图的调度器 使用了该阶段信息来告诉spark的监 * 听者有任务启动了(这发生在任何阶段的尝试被发生之前) */ private var _latestInfo: StageInfo = StageInfo.fromStage(this, nextAttemptId) /** * Set of stage attempt IDs that have failed with a FetchFailure. * 拉取失败的的阶段尝试Id集 * We keep track of these failures in order to avoid endless retries if a stage keeps failing * with a FetchFailure. * 我们追踪这些失败时为了防止 一个阶段一直发生拉取失败后 无尽的重试 * We keep track of each attempt ID that has failed to avoid recording duplicate failures if * multiple tasks from the same stage attempt fail (SPARK-5945). * 我们追踪失败的每次尝试的Id以防止 如果许多任务在同一个阶段的尝试中失败 会导致 记录重复的失败 */ val fetchFailedAttemptIds = new HashSet[Int] // 清除失败Id private[scheduler] def clearFailures() : Unit = { fetchFailedAttemptIds.clear() } /** Creates a new attempt for this stage by creating a new StageInfo with a new attempt ID. * 通过创建一个新的附带新的尝试Id的状态信息 来为当前阶段创建一个新的尝试 */ def makeNewStageAttempt( // 需要计算的分区 numPartitionsToCompute: Int, // 本地优先的任务 taskLocalityPreferences: Seq[Seq[TaskLocation]] = Seq.empty): Unit = { // TaskMetrics是内部累加器的一个包装类 val metrics = new TaskMetrics metrics.register(rdd.sparkContext) // 获取最近的阶段信息 _latestInfo = StageInfo.fromStage( this, nextAttemptId, Some(numPartitionsToCompute), metrics, taskLocalityPreferences) nextAttemptId += 1 } /** Returns the StageInfo for the most recent attempt for this stage. * 返回最近最多尝试的 当前阶段的 阶段信息 */ def latestInfo: StageInfo = _latestInfo override final def hashCode(): Int = id override final def equals(other: Any): Boolean = other match { case stage: Stage => stage != null && stage.id == id case _ => false } /** Returns the sequence of partition ids that are missing (i.e. needs to be computed). * 返回缺失的分区 id 的序列 */ def findMissingPartitions(): Seq[Int] }
相关推荐
Hhanwen 2020-07-05
zilianxiaozhu 2018-09-12
yanqianglifei 2020-02-21
89463661 2020-02-20
xiaodaiwang 2019-12-19
fengjing81 2019-12-02
dxyadc 2019-10-27
lfjjia 2010-11-09
herryyy 2012-12-18
PANH 2019-07-01
starksummer 2019-03-31
上海彭彭 2019-06-30
CloasGao 2019-06-29
82530995 2019-06-28
雪飞海 2019-06-26
成长之路 2018-10-18
adayan0 2018-07-20
jinkun00 2015-05-09
FEvivi 2018-02-22