在 Apache Spark 中，Job 对象是执行逻辑的核心组件之一，它代表了对一系列数据操作（如 transformations 和 actions）的提交。理解 Job 的本质和它在 Spark 中的运行机制，有助于深入理解 Spark 的任务调度、执行模型和容错机制。
Spark Job 对象的定义与作用
Spark 中的 Job 主要用于表示一个具体的计算作业，它是由用户提交的 Action（例如 count(), collect(), saveAsTextFile() 等）触发的。这些动作会生成一个 Job 对象，最终调度并执行一系列与之相关的任务。
主要作用
- 调度的基本单元：
Job是 Spark 中由调度器提交给集群调度系统的最小执行单元。每次用户调用Action时都会触发一个新的Job。 - 执行依赖解析：在
Job中，Spark 会解析由 RDD transformations 构建的执行 DAG（Directed Acyclic Graph，有向无环图），将整个 DAG 划分为多个阶段（Stages），并将每个阶段的计算划分为多个任务（Tasks）。 - 生命周期管理：
Job还负责跟踪其执行状态，包括成功、失败、重试等。调度器负责管理Job的整个生命周期。 - 结果汇总与返回：
Job的最终结果会返回给提交的客户端，并供用户程序使用。
底层架构与执行流程
Spark 中 Job 的执行流程可以分为以下几个步骤：
-
用户触发 Action：
当用户调用 RDD 的Action操作（如collect()）时，Spark 会触发一个Job的创建。每个Job与一个 Action 一一对应。 -
DAG 划分：
Spark 的调度器会将 RDD 的 transformations 构建的 DAG 划分为多个阶段（Stages）。这些阶段之间通过宽依赖（Shuffle Dependencies）进行划分，每个Stage是一组可以并行执行的操作。 -
生成任务（Task）：
每个Stage会被进一步分解为多个Task。这些Task通常与数据分区（Partition）相对应。每个Task会在集群的不同节点上执行，并行处理数据。 -
调度执行：
每个Stage中的Task通过TaskSet被提交到TaskScheduler，由调度器在集群中的不同节点上执行。调度器会根据可用资源、节点健康状况等因素进行调度。 -
结果返回与 Job 完成：
在所有Stage完成后，Job被标记为完成，最后的结果会被返回给用户，供进一步处理。
代码层面解释
在 Spark 源码中，Job 的相关实现可以在 DAGScheduler 和 Job 类中找到。DAGScheduler 是调度层的核心组件，它负责将用户的高层操作分解为具体的作业（Job）和任务（Task）。
1. Job 对象的类结构
在 Spark 代码中，Job 由 DAGScheduler 负责创建。每个 Job 都有一个唯一的 jobId。其定义主要存在于 DAGScheduler.scala 文件中。
// DAGScheduler.scala (部分代码)
class Job(
val jobId: Int,
val finalStage: Stage,
val callSite: CallSite,
val listener: JobListener,
val properties: Properties) {
def finished(result: JobResult): Unit = {
listener.jobSucceeded(result)
}
}
在上述代码中，Job 对象中有几个关键字段：
jobId：作业的唯一标识符。finalStage：该Job的最后一个Stage，作业的完成意味着该阶段的完成。callSite：作业执行时的代码位置信息。listener：用于监听Job执行状态的监听器，通常用于执行完成时通知上层。properties：包含一些与作业相关的配置信息。
2. DAGScheduler 的作用
DAGScheduler 是 Spark 调度器的核心组件，负责管理 Job 的生命周期，包括划分阶段、提交任务、重试失败任务等。
DAGScheduler 的部分代码如下：
// DAGScheduler.scala (简化示例)
private[scheduler] class DAGScheduler(
taskScheduler: TaskScheduler,
listenerBus: LiveListenerBus,
mapOutputTracker: MapOutputTracker,
blockManagerMaster: BlockManagerMaster,
env: SparkEnv,
clock: Clock = new SystemClock()) extends Logging {
private val jobIdToActiveJob = new HashMap[Int, ActiveJob]
def submitJob[T, U](
rdd: RDD[T],
func: (TaskContext, Iterator[T]) => U,
partitions: Seq[Int],
callSite: CallSite,
resultHandler: (Int, U) => Unit,
properties: Properties): JobWaiter[U] = {
// 创建一个新的 Job
val jobId = nextJobId.getAndIncrement()
val finalStage = createResultStage(rdd, func, partitions, jobId, callSite)
val job = new Job(jobId, finalStage, callSite, resultHandler, properties)
// 提交 Job
jobIdToActiveJob(jobId) = new ActiveJob(job, finalStage)
submitStage(finalStage)
return job.waiter
}
}
这个代码展示了 DAGScheduler 是如何接收用户的 Action 调用，创建 Job 并提交执行的：
submitJob方法会基于传入的 RDD 和操作函数创建一个新的Job。- 调用
createResultStage方法将 RDD DAG 分解为Stage，并创建该Job的最终Stage。 submitStage方法负责将阶段提交到底层的TaskScheduler，执行该阶段中的任务。
3. Job 与 ActiveJob 的关系
Job 是一个抽象的高层次的概念，而 ActiveJob 是其运行时状态的一个封装。ActiveJob 代表一个正在运行的 Job，包含了更多的运行时状态信息。
// ActiveJob.scala
private[spark] class ActiveJob(
val jobId: Int,
val finalStage: Stage,
val func: (TaskContext, Iterator[_]) => _,
val partitions: Array[Int],
val callSite: CallSite,
val listener: JobListener,
val properties: Properties) {
val numTasks = partitions.length
var numFinished = 0
def stageFinished(stage: Stage): Unit = {
if (numFinished == numTasks) {
listener.jobSucceeded(this)
}
}
}
总结
Job的核心作用：Job是 Spark 中用于管理由Action操作触发的计算任务。它通过DAGScheduler划分执行阶段（Stages），并调度相应的任务执行，最终将计算结果返回给用户。- 代码实现：
Job在 Spark 源码中作为调度系统的一个重要组成部分，由DAGScheduler创建并管理。DAGScheduler负责将用户的作业拆解为可执行的阶段和任务，并交由TaskScheduler执行。 - 调度逻辑：
Job包含了执行依赖、分区信息和调度状态等。通过与Stage和Task的结合，Job的执行能够在大规模分布式环境中高效并行化。
了解这些底层机制有助于理解 Spark 在执行任务时的调度流程和容错处理机制，也为优化 Spark 作业的性能提供了更深入的视角。
