当前位置：首页 > news >正文

Spark 核心之 Application 和 Job 原理剖析

news 2026/8/5 10:12:05

摘要：你是否清楚一个 Spark 应用中到底有几个 Job？为什么.collect()会触发 Job 而.map()不会？Application、Job、Stage、Task 之间的关系到底是什么？本文从四层执行层级全景图、Action 触发 Job 的源码链路、DAGScheduler 的 Stage 切分规则、多 Job 应用实战示例四个维度，配合 1 张原创深色架构图 + 完整源码分析，带你彻底理清 Spark 的执行层级体系。
关键词：Spark Application, Job, Stage, Task, Action, DAGScheduler, SparkContext, 执行层级

一、开篇：你写的一个 Application 到底有几个 Job？

先看一段代码，你能准确说出它会产生几个 Job 吗？

vallines=sc.textFile("hdfs:///data/words.txt")// Transformationvalwords=lines.flatMap(_.split(" "))// Transformationvalpairs=words.map((_,1))// Transformationvalcounts=pairs.reduceByKey(_+_)// Transformationcounts.collect()// Action 1 → Job 0counts.count()// Action 2 → Job 1counts.saveAsTextFile("hdfs:///output")// Action 3 → Job 2

答案：3 个 Job。因为每个 Action 算子都会触发一个新的 Job。而textFile、flatMap、map、reduceByKey都是 Transformation——它们只是构建 DAG，不触发任何计算。

二、执行层级全景图

2.1 四层模型

Application (SparkContext) │ ├── Job-0 (调用 collect() 触发) │ ├── Stage 0 (ShuffleMapStage: 2 Tasks) │ └── Stage 1 (ResultStage: 3 Tasks) │ ├── Job-1 (调用 count() 触发) │ ├── Stage 2 (ShuffleMapStage: 2 Tasks) │ └── Stage 3 (ResultStage: 3 Tasks) │ └── Job-2 (调用 saveAsTextFile() 触发) ├── Stage 4 (ShuffleMapStage: 2 Tasks) └── Stage 5 (ResultStage: 3 Tasks)

三、Action 触发 Job 的源码链路 🔥

// 源码：RDD.scala - collect()defcollect():Array[T]=withScope{valresults=sc.runJob(this,(iter:Iterator[T])=>iter.toArray)results.flatten}// 源码：RDD.scala - count()defcount():Long=sc.runJob(this,Utils.getIteratorSize _).sum// 源码：RDD.scala - saveAsTextFile()defsaveAsTextFile(path:String):Unit={// ... 内部最终调用 sc.runJob()}// 核心链路// Action → sc.runJob() → DAGScheduler.runJob()// → DAGScheduler.handleJobSubmitted()// → 创建 ActiveJob → Stage 切分 → submitStage()

3.1 哪些算子是 Action？

算子	返回类型	说明
`collect()`	Array[T]	拉取所有数据到 Driver
`count()`	Long	计数
`take(n)`	Array[T]	取前 n 个
`reduce(f)`	T	聚合
`foreach(f)`	Unit	遍历
`saveAsTextFile()`	Unit	保存到文件
`first()`	T	取第一个

反直觉点：reduceByKey不是 Action！它是 Transformation，触发 Shuffle 但不触发 Job。

四、Stage 切分规则

// 源码：DAGScheduler.scalaprivatedefgetMissingParentStages(stage:Stage):List[Stage]={stage.rdd.dependencies.flatMap{caseshufDep:ShuffleDependency[_,_,_]=>// WideDep → 切分 → 创建父 ShuffleMapStagegetOrCreateShuffleMapStage(shufDep,stage.firstJobId)case_=>Nil// NarrowDep → 不切分，保持在同一 Stage}.toList}

一句话规则：遇到 ShuffleDependency (Wide Dependency) 即切分 Stage。

五、多 Job 实战示例

valrdd=sc.parallelize(1to1000,4)// 4 Partitions// Job 1: countprintln(s"Count:${rdd.count()}")// Action → Job-1// Job 2: collectvalarr=rdd.collect()// Action → Job-2 (无 Shuffle, 1 Stage)// Job 3: saverdd.saveAsTextFile("hdfs:///output")// Action → Job-3// 总计: 3 个 Job, 3×1=3 个 ResultStage

// 带 Shuffle 的场景valrdd=sc.parallelize(1to1000,4).map(x=>(x%10,x))// Narrow.groupByKey()// Wide! Stage 边界rdd.count()// Action → Job-1: Stage 0 (ShuffleMapStage) + Stage 1 (ResultStage)

六、Application/Job/Stage/Task 对比表

层级	定义	触发条件	数量
Application	SparkContext 实例	spark-submit	1
Job	一个 Action 的完整计算	Action 算子	1~N
Stage	Shuffle 边界切分的计算阶段	ShuffleDependency	每个 Job 1~N
Task	处理一个 Partition 的最小单元	Stage 内 Partition 数	每个 Stage 1~N

七、总结

要点	总结
层级关系	1 App = N Jobs = N×M Stages = N×M×P Tasks
Job 触发	每个 Action 算子调用 sc.runJob() 创建新 Job
Stage 切分	遇到 ShuffleDependency 即切分
Task 生成	Stage 最后一个 RDD 的 Partition 数量 = Task 数