文章目录
Stage及Spark计算模式
一、Stage
当我们创建一个 Spark 程序后，可以看做创建了一个 Spark Application，一个 Application 中可以有多个 RDD，Spark 任务会根据 RDD 之间的依赖关系形成一个 DAG 有向无环图，Spark 会按照 RDD 宽窄依赖关系将 DAG 划分为相互依赖的多个 Stage，遇到 RDD 宽依赖就划分 Stage，每个 Stage 包含一个或多个 Task 任务，这些 Task 最终发送到计算节点执行，可见 Stage 是由一组并行的 Task 组成。
Stage 的划分规则：从 finalRDD 开始，从后往前依次递归，遇到 RDD 宽依赖就生成 Stage，因此，Stage 分为 Shuffle Map Stage 和 Result Stage 两种类型。
如上图所示，Spark 从 RDD G 开始，按照 RDD 之间的依赖关系从后向前遍历。遇到 Shuffle 依赖时，就会生成一个新的 Stage，其中 Stage3 为 Result Stage，Stage1 和 Stage2 为 Shuffle Map Stage。
二、Spark计算模式
在 WordCount 示例中，划分 Stage 如下图所示，RDD 的每个 Partition 并不直接存储数据，而是包含对应的计算逻辑（如：textFile、flatMap、map、reduceByKey），每个 RDD 可以视为一个迭代器，在同一个 Stage 中，多个 RDD 的 Partition 通过一个 Task 从数据源开始，依次执行各自的计算逻辑，直到数据最终落地，由于 Stage 中的 RDD 有多个分区，因此每个 Stage 中会有多个并行的 Task 处理数据，这也是每个 Stage 由一组并行 Task 组成的原因。
总结：这种 Spark 的计算模式叫做 pipeline（管道）处理模式，也可以叫做迭代器处理模式，这种处理数据模式减少了中间数据的存储和读取，提高了计算效率。
关于 Spark 计算模式，需要注意如下几点：
- Stage 中的并行度由谁决定？由 Stage 中最后一个 RDD 的 partition 个数决定。
- 管道中的数据何时落地？当发生 Shuffle 操作、对数据进行持久化（如调用 persist 或 cache 方法），或者计算结果需要写入磁盘时，数据会落地。
- 如何提高 Stage 的并行度？可以使用带有 Shuffle 的算子（如 reduceByKey、groupByKey 等）对 RDD 增加分区数量，从而提高并行度。代码并行度设置多少合适？建议每个 core 处理 2~3 个 task。
如下代码验证 Spark 中 pipeline 管道计算模式。
三、小结
本节围绕 Stage 的划分与 Spark 的计算模式展开，核心结论可以归纳为以下几点：
- Stage 的本质：Stage 是由一组并行 Task 组成的执行单元，其边界由 RDD 的宽依赖（Shuffle 依赖）决定，划分时从 finalRDD 出发由后向前递归，遇到宽依赖即切分。
- 两类 Stage：Shuffle Map Stage 负责产出 Shuffle 所需的中间数据，Result Stage 负责产出最终结果，二者共同构成一个完整的 Job 执行链路。
- Pipeline 计算模式：同一 Stage 内的多个 RDD 以迭代器方式串联，数据在内存中依次流转，避免了中间结果的反复落盘，显著提升了计算效率。
- 并行度调优：Stage 的并行度取决于最后一个 RDD 的分区数，可通过带 Shuffle 的算子调整分区；同时应关注数据落地时机，合理使用 persist 或 cache 缓存热点数据。
理解 Stage 的划分逻辑与 pipeline 执行机制，是后续进行 Shuffle 调优、任务并行度设计和作业性能诊断的基础。在实际开发中，建议结合 Spark UI 观察每个 Stage 的 Task 数量与执行耗时，从而更直观地定位性能瓶颈。
Java代码：
SparkConf conf = new SparkConf().setMaster("local").setAppName("pipelineTest");
JavaSparkContext sc = new JavaSparkContext(conf);
JavaRDD<String> rdd1 = sc.parallelize(Arrays.asList("a", "b", "c", "d"));
JavaRDD<String> rdd2 = rdd1.map(x -> {
System.out.println("map------: " + x);
return x;
});
JavaRDD<String> rdd3 = rdd2.filter(x -> {
System.out.println("filter******: " + x);
return true;
});
rdd3.foreach(x->System.out.println("foreach======: " + x));
sc.stop();Scala代码：
val conf: SparkConf = new SparkConf().setMaster("local").setAppName("PipelineTest")
val sc = new SparkContext(conf)
val rdd1: RDD[String] = sc.parallelize(List("a", "b", "c", "d"))
val rdd2: RDD[String] = rdd1.map(x => {
println(s"map------:$x")
x
})
val rdd3: RDD[String] = rdd2.filter(x => {
println(s"filter******:$x")
true
})
rdd3.foreach(x => {
println(s"foreach======:$x")
})
sc.stop()
输出结果：
map------: a
filter******: a
foreach======: a
map------: b
filter******: b
foreach======: b
map------: c
filter******: c
foreach======: c
map------: d
filter******: d
foreach======: d- 📢博客主页：https://lansonli.blog.csdn.net
- 📢欢迎点赞 👍 收藏 ⭐留言 📝 如有错误敬请指正！
- 📢本文由 Lansonli 原创，首发于 CSDN博客🙉
- 📢停下休息的时候不要忘了别人还在奔跑，希望大家抓紧时间学习，全力奔赴更美好的生活✨
转载自 CSDN-专业IT技术社区
原文链接：https://blog.csdn.net/xiaoweite1/article/details/170235087
