📚 课程目录第2章 Spark核心架构与运行流程

2. Spark核心架构与运行流程

理解Spark的运行流程，是面试中区分“会用API”和“懂引擎”的关键分水岭。很多候选人能写出rdd.map().filter().reduceByKey()，却说不清这段代码提交后究竟在集群里发生了什么。本节我们沿着一条作业的完整生命周期，把Driver、Executor、Cluster Manager、DAG调度、Stage划分和Task执行串成一条清晰的链路。

一、四个核心角色：谁在干什么

Spark应用在运行时涉及四个关键角色，面试中常被要求逐一说明职责边界。

Driver（驱动器）是应用的“大脑”。它运行用户的main()方法，创建SparkContext或SparkSession，把用户代码转换成RDD或DataFrame的血缘关系图（Lineage），并负责DAG的构建、Stage划分、Task生成与调度。Driver还通过心跳监控Executor状态，收集任务执行结果。需要强调：Driver不是“协调者”这么简单，它承担了几乎所有逻辑决策，一旦Driver进程崩溃，整个应用即告失败。

Executor（执行器）是真正干活的“工人”。每个Executor是一个JVM进程，运行在集群的Worker节点上，内部以线程池方式并发执行Task。它负责：执行Task中的计算逻辑、把RDD分区数据缓存在内存或磁盘、通过BlockManager提供Shuffle数据的读写服务、向Driver汇报心跳与任务状态。一个Executor内的多个Task共享同一JVM内存，这也是为什么Executor内存和核数配置会直接影响并发度与OOM风险。

Cluster Manager（集群管理器）负责资源的分配与回收。Spark支持Standalone、YARN、Kubernetes、Mesos等模式。它的职责是：按Driver的申请，在集群节点上启动Executor进程，并在应用结束时回收资源。注意区分：Cluster Manager只管“给资源、起进程”，不参与Task调度和计算逻辑。

Application（应用）是用户提交的整个程序，包含一个Driver和多个Executor，它们共同构成一个独立的运行单元。

二、作业提交：从spark-submit到Executor就绪

以YARN集群模式为例，一条典型的提交流程如下：

spark-submit \
  --master yarn \
  --deploy-mode cluster \
  --executor-memory 4g \
  --executor-cores 2 \
  --num-executors 10 \
  --class com.example.WordCount \
  app.jar hdfs:///input/words.txt

执行后依次发生：客户端向YARN ResourceManager申请启动ApplicationMaster；AM中运行Driver（cluster模式下Driver在AM内）；Driver初始化SparkContext，向RM申请容器；RM在各NodeManager上分配容器并启动Executor；Executor反向注册到Driver，汇报自身可用的核数与内存。当Driver看到足够Executor注册后，作业才真正开始调度。这里有一个面试常考点：cluster模式与client模式的区别——client模式下Driver运行在提交机器上，便于本地调试但提交机需保持在线；cluster模式下Driver在集群内，适合生产环境。

三、DAG调度：从代码到执行计划

用户写的转换算子（map、filter、join等）是惰性的，只有遇到行动算子（count、collect、saveAsTextFile）才会触发真正的执行。触发时，Driver中的DAGScheduler开始工作：

它首先根据RDD之间的依赖关系构建DAG。依赖分为两类：窄依赖（Narrow Dependency）指父RDD的每个分区最多被子RDD的一个分区使用，如map、filter、union；宽依赖（Wide Dependency / Shuffle Dependency）指父RDD的一个分区可能被子RDD的多个分区使用，如groupByKey、reduceByKey、join（未co-partition时）。宽依赖意味着数据需要跨节点重新分发，也就是Shuffle。

DAGScheduler以宽依赖为边界，把DAG切分成多个Stage。切分规则是：从触发行动算子的RDD向前回溯，遇到宽依赖就断开，形成一个Stage；每个Stage内部由一系列窄依赖组成，可以流水线（pipeline）执行。Stage分为两种：ResultStage（最后一个Stage，直接产出结果）和ShuffleMapStage（为下游Shuffle准备数据）。

举个具体例子。假设代码为：

val rdd = sc.textFile("hdfs:///input/words.txt")
val counts = rdd.flatMap(_.split(" "))
                .map((_, 1))
                .reduceByKey(_ + _)
                .filter(_._2 > 10)
counts.saveAsTextFile("hdfs:///output")

这里reduceByKey是宽依赖，因此DAG被切成两个Stage：Stage 0包含textFile → flatMap → map，是ShuffleMapStage；Stage 1包含reduceByKey → filter → saveAsTextFile，是ResultStage。Stage 0先执行，把按key分区后的中间数据写到磁盘，Stage 1再拉取这些数据完成聚合。

四、Task生成与执行：Stage内部的并行

Stage划分完成后，DAGScheduler为每个Stage生成一组TaskSet，交给底层的TaskScheduler。Task数量由该Stage最后一个RDD的分区数决定——例如Stage 0的textFile若产生200个分区，就会生成200个ShuffleMapTask。

TaskScheduler按照“数据本地性优先”的原则分配Task，本地性从高到低为：PROCESS_LOCAL（同JVM）、NODE_LOCAL（同节点）、RACK_LOCAL（同机架）、ANY（任意）。它会等待一小段时间（默认3秒，可通过spark.locality.wait调整）以争取更高本地性，超时后降级分配，避免资源空转。

Executor收到Task后，在线程池中启动线程执行。对于ShuffleMapTask，它读取输入分区，执行窄依赖链，把结果按分区器写到本地磁盘的Shuffle文件中，并向Driver汇报MapStatus（包含数据位置和大小）。对于ResultTask，它通过BlockManager从各Map节点拉取对应分区数据，完成聚合后把结果返回Driver。

整个过程中，Driver持续通过心跳跟踪Task状态，某个Task失败会自动重试（默认4次，spark.task.maxFailures），若某个Stage因Executor丢失而失败，DAGScheduler会重新提交该Stage的父Stage，这就是Spark容错的核心机制——基于Lineage的重算。

五、面试高频追问与判断依据

掌握上述链路后，可以应对几类典型追问：

“为什么要划分Stage？”——因为宽依赖必须等上游全部写完才能开始下游拉取，Stage是Shuffle的天然边界，也是失败重算的最小单位。

“Task数量怎么定？”——由Stage最后RDD的分区数决定，而分区数又受textFile的block数、reduceByKey的并行度参数、repartition等影响。分区太少会导致并发不足，太多则Task调度开销上升。

“Driver和Executor内存分别管什么？”——Driver内存主要承载DAG、Task元数据和结果收集，结果过大（如collect）会OOM；Executor内存分为执行内存（Shuffle、Join、Sort）和存储内存（Cache、Broadcast），由统一内存管理器动态借用。

建议在本地用spark-shell跑一遍上述WordCount，打开http://localhost:4040的Spark UI，在Stages页签中观察Stage数量、每个Stage的Task数、Shuffle Read/Write大小与耗时。当你能对着UI说出“这个Stage为什么有200个Task、Shuffle Write为什么是这些字节”时，这条全链路才算真正打通。