Spark作业的执行遵循**惰性计算（Lazy Evaluation）**原则，即只有当一个**行动（Action）**操作（如`saveAsTextFile()`）被调用时，所有之前定义的**转换（Transformation）**操作（如`reduceByKey()`）才会作为一个整体作业（Job）被提交并实际执行。这个过程涉及到RDD血统、DAG构建、Stage划分、Shuffle和任务调度等多个核心机制 [ref_1][ref_4][ref_5]。 下面我们详细解析这段典型WordCount代码的每一步执行过程。 ### 一、RDD血统与DAG构建 首先，假设已有一个`SparkContext`对象`sc`，并且我们基于一个文本文件创建了初始RDD。 ```python # 1. 创建初始RDD：这是一个转换操作，但此时不读取数据，仅记录“从某路径读取”的元数据。 lines = sc.textFile("hdfs://data/input.txt") # Transformation # 2. 扁平化：将每行文本拆分成单词，生成一个新的RDD。 words = lines.flatMap(lambda line: line.split(" ")) # Transformation # 3. 映射：将每个单词映射为(单词, 1)的键值对。 pairs = words.map(lambda word: (word, 1)) # Transformation # 4. 按键聚合：对相同key(单词)的value进行累加。 wordCounts = pairs.reduceByKey(lambda a, b: a + b) # Transformation (Shuffle) # 5. 保存结果：这是一个行动操作，它将触发整个计算流程。 wordCounts.saveAsTextFile("hdfs://data/output") # Action (触发Job执行) ``` **关键点分析**： 1. **惰性计算**：在执行到第5步`saveAsTextFile()`之前，代码的第1-4步仅构建了一个记录了转换关系的**RDD血统图**（Lineage Graph），并没有发生任何实际的数据读取或计算 [ref_4]。 2. **宽依赖**：`reduceByKey`操作是一个**宽依赖（Shuffle Dependency）** 操作。为了将相同key的数据汇集到同一个分区进行处理，它要求进行全量的数据洗牌（Shuffle），这构成了一个Stage的边界 [ref_1]。 3. **行动触发**：`saveAsTextFile()`是行动操作，它的目的是将计算结果写入外部存储系统。它需要最终的计算结果，因此会强制Spark调度器执行所有上游的转换操作。 ### 二、执行流程与调度详解 当`saveAsTextFile()`被调用时，Spark会触发以下一系列内部操作： 1. **Job提交**：`SparkContext`将`saveAsTextFile`及其上游的转换链提交给**DAGScheduler**。 2. **Stage划分**： * DAGScheduler分析RDD的血统图，并**以宽依赖（即Shuffle操作）为边界**，将整个计算流程划分为多个Stage（阶段） [ref_2][ref_4]。 * 在这个WordCount例子中，`reduceByKey`是唯一的宽依赖操作。因此，DAG会被划分为两个Stage： * **Stage 0**：包含从`textFile`到`map`的所有窄依赖转换。其最后的RDD是`pairs`。 * **Stage 1**：以`reduceByKey`开始，以`saveAsTextFile`结束。`reduceByKey`内部的Shuffle写操作是Stage 0和Stage 1的分界线。 3. **Task生成**： * 每个Stage被进一步划分为多个**Task**。Task的数量等于该Stage最后一个RDD的分区数 [ref_2]。 * **Stage 0**：假设`lines` RDD有N个分区，那么`flatMap`和`map`这些窄依赖操作可以沿着分区进行流水线优化。因此，Stage 0会产生N个**ShuffleMapTask**。每个Task负责读取一个数据分片，执行`flatMap`和`map`，然后为后续的`reduceByKey`准备Shuffle数据。 * **Stage 1**：`reduceByKey`之后的分区数（可通过`numPartitions`参数设置，默认为父RDD分区数）决定了Task的数量。假设设置为M，那么Stage 1会产生M个**ResultTask**。每个Task负责拉取（Fetch）属于自己分区的Shuffle数据，执行聚合（`reduceByKey`的`reduce`函数），并将最终结果写入HDFS。 4. **Task调度与执行**： * TaskScheduler将每个Stage的Task集提交到集群管理器（如YARN），由集群管理器分配到各个工作节点（Worker Node）的Executor上执行 [ref_1]。 * **Stage的执行顺序是串行的**。只有当前一个Stage（Stage 0）的所有Task都执行完毕，并完成了Shuffle数据的写入（本地磁盘），下一个Stage（Stage 1）的Task才能开始拉取数据进行计算 [ref_2]。 ### 三、`reduceByKey`的Shuffle过程详解 `reduceByKey`是理解该执行流程的核心，其内部包含了一个完整的Shuffle过程，分为`map`端和`reduce`端： 1. **Map端（对应Stage 0的最后阶段）**： * 每个ShuffleMapTask处理自己分区内的`(word, 1)`数据。 * 它首先会应用一个可选的**combine**（合并）函数。`reduceByKey`会在map端对每个分区内的数据进行局部聚合（例如，将分区内相同的`word`的`1`相加）。这一步极大地减少了需要通过网络传输到reduce端的数据量，是`reduceByKey`优于`groupByKey`的关键 [ref_1][ref_5]。 * 然后，根据目标分区器（默认是HashPartitioner），计算每条数据的`key`应该属于哪个下游分区（假设有M个分区）。 * 最后，将数据写入内存缓冲区，并根据分区ID排序，最终溢写（Spill）到本地磁盘，生成一个包含M个文件的Shuffle输出文件（每个文件对应一个下游分区）。 2. **Reduce端（对应Stage 1的开始阶段）**： * 每个ResultTask需要计算一个最终的分区结果。 * Task启动后，会向所有运行过Stage 0的Executor节点发起网络请求，拉取（Fetch）属于自己分区的Shuffle数据片段。 * 拉取到的数据会进行归并排序。 * 然后，对已排序的、属于同一个`key`的数据流，应用`reduce`函数（即代码中的`lambda a, b: a + b`）进行全局聚合，得到每个单词的最终计数。 * 最后，将聚合结果传递给下游的`saveAsTextFile`操作，由每个Task负责将结果写入HDFS的指定路径。 ### 四、`saveAsTextFile`的执行 `saveAsTextFile`作为行动操作，它会触发Stage 1的计算。其执行逻辑由每个ResultTask完成： 1. 每个Task处理自己分区内已经计算好的最终数据（即`(word, count)`键值对列表）。 2. 每个Task会在HDFS的输出目录下，写入一个以自己分区ID命名的文件（例如`part-00000`, `part-00001`）。 3. 所有Task执行完毕，`saveAsTextFile`操作完成，整个Job结束。HDFS输出目录中会生成M个结果文件（M为Stage 1的分区数），以及一个_SUCCESS标记文件。 ### 五、执行过程总结表 为了更清晰地展示整个流程，下表概括了从代码到最终输出的关键步骤和内部机制： | 步骤 | 用户代码 | Spark内部操作 | 依赖类型 | Stage划分 | 任务类型 | 数据流向 | | :--- | :--- | :--- | :--- | :--- | :--- | :--- | | **1. 转换定义** | `lines = sc.textFile(...)` | 记录血统（从HDFS读取） | 无 | - | - | 无 | | | `words = lines.flatMap(...)` | 记录血统（扁平化转换） | 窄依赖 | Stage 0 | ShuffleMapTask | 逻辑关系 | | | `pairs = words.map(...)` | 记录血统（映射转换） | 窄依赖 | Stage 0 | ShuffleMapTask | 逻辑关系 | | | `wordCounts = pairs.reduceByKey(...)` | 记录血统（聚合转换） | **宽依赖** | Stage 1 | ResultTask | **Shuffle边界** | | **2. 行动触发** | `wordCounts.saveAsTextFile(...)` | DAGScheduler提交Job | - | - | - | - | | **3. Stage 0执行** | - | `textFile` -> `flatMap` -> `map` -> **Shuffle Write** | 窄依赖链 | Stage 0 | ShuffleMapTask | 数据读取、转换、按Key分区写入本地磁盘 | | **4. Shuffle** | - | **数据交换** | - | Stage 0/1边界 | - | ResultTask从各节点拉取属于自己分区的数据 | | **5. Stage 1执行** | - | **Shuffle Read** -> `reduce` -> `saveAsTextFile` | - | Stage 1 | ResultTask | 数据拉取、聚合、写入HDFS | | **6. 作业完成** | - | 所有Task完成，Driver收到成功报告 | - | - | - | HDFS生成结果文件 | 综上所述，这段简短的WordCount代码背后，是Spark一套完整的**惰性计算、DAG调度、Shuffle管理和容错恢复**的分布式执行框架在运作。`reduceByKey`定义了需要进行数据重分布和聚合的关键转换，而`saveAsTextFile`作为行动操作，是点燃整个计算引擎的火花，最终驱动数据从输入源经过复杂的处理流程，流向输出目的地 [ref_1][ref_2][ref_4][ref_5]。