前言

Spark Job主要阐述了Spark的内容操作流程以及不提交RDD 后各流程的操作原理等Spark计算逻辑。 Shuffle描述了从map任务输出数据到reduce任务输入的过程。 Shuffle是Map和Reduce之间的桥梁。
Map输出必须在Reduce中使用以通过shuffle链接。
shuffle一般分为两个部分：
Map阶段的数据准备和Reduce阶段的数据拷贝处理。
map侧的Shuffle通常称为Shuffle Write， Reduce侧的Shuffle称为Shuffle Read。

不提交RDD在说法上并不严谨：本义是指Spark中的惰性计算机制，具体来说：

1.RDD的转换操作（transformations）不会立即执行。
当调用 .map()、.filter()、join() 等方法时，Spark只是记录这些操作，构建一个计算的逻辑计划（DAG），但不会立即执行计算
只是定义了“要做什么”，而不是“现在就开始做”

2.只有行动操作（actions）才会触发真正的计算
只有当你调用 .collect()、.count()、.saveAsTextFile() 等行动操作时，Spark才会提交作业（Job） 并执行整个计算流程
这就是所谓的“不提交RDD”——在调用行动操作之前，RDD实际上并没有被计算

---

一、Spark job是什么？

（一）概念

Spark RDD的转化函数不会执行任何行动操作，当Spark执行RDD的行动函数时，我们会在应用转化操作时自动创建RDD谱系图（逻辑DAG）。

The transformation function of Spark RDD will not perform any action, and when Spark is executing the action function of RDD, RDD lineage graph is created automatically as we apply transformation(logical DAG).

Spark调度器（DAG scheduler）会构建执行图（graph）并启动一个Spark作业（job）。该作业由多个阶段组成，这些都是最终实现RDD数据转换所需的步骤。每个阶段由一组任务组成，这些任务在执行器上并行计算。

the Spark scheduler (DAG Scheduler) will construct the execution graph (graph) and Start a Spark job (Job). The job is composed of many stages, which are the steps required to realize the final RDD data in the conversion. Each stage consists of a set of tasks that are calculated in parallel on an executor.

（二）创建过程

下图说明了Job->Stage->Task的创建过程：
通过Spark Context或Spark Session执行应用程序。
如果应用程序中存在Action操作，则此时将生成并提交一个作业。
如果需要在作业中执行宽转换和其他操作，则将创建多个Stage

The following figure illustrates the creation process of Job->Stage->Task:
◼ Execute the application through Spark Context or Spark Session
◼ If there is an Action operation in the application, a job will be generated at this time and submitted
◼ If you need to perform wide conversion and other operations in the Job, several Stages will be created
◼ Stage will create a series of Tasks, which will be executed in parallel in each executor

二、Spark Job Introduction

（一）有关术语概念

1.DAG (Directed Acyclic Graph)

Directed – the data flows in 1 direction (step -> next step)
数据流向1个方向（step ->下一步）
Acyclic – No loop (data never goes back to previous step)
没有循环（数据永远不会回到上一步）
Graph – It’s a network of steps
图-这是一个步骤网络

why we need DAG in spark?

Because we can know:
1.Which steps must happen first
2.Which steps can run in parallel
3.Where shuffle required
4.How to split work into stages and tasks.

1.哪些步骤必须先发生
2.哪些步骤可以并行运行
3.需要洗牌的地方
4.如何将工作分成阶段和任务。

2.DAG scheduler

(1）概念

在Spark任务调度的高层，Spark会根据RDD的依赖关系为每个作业构建一个有向无环图（Directed Acyclic Graph， DAG）。在Spark中，这称为DAG调度器（DAG Scheduler）。

At the high level of Spark task scheduling, Spark will build a Directed Acyclic Graph (DAG) for each Job based on the RDD dependencies. In Spark, this is called DAG Scheduler.

(2）作用

DAG调度器负责获取RDD，然后在shuffle边界将作业划分为阶段，创建阶段图（执行DAG），并将各个阶段发送给任务调度器。

The DAG Scheduler is responsible for taking the RDD lineage then Splitting the job into stages at shuffle boundaries, create a stage graph (execution DAG) and sending stages to the Task Scheduler.

(3）阶段图

阶段图由多个阶段组成，每个阶段包含可以并行执行的任务。各个阶段由数据混洗依赖关系决定，每个阶段的边界就是RDD之间进行数据混洗的位置。

The stage graph consists of stages, where each stage contains tasks that can be executed in parallel. Stages are determined by shuffle dependencies—each stage boundary is where data shuffling happens between RDDs.

Spark Web-UI端口号：4040

(4）阶段依赖

在一个阶段中，所有任务都独立并行运行。在各个阶段之间，任务依赖于前几个阶段的输出。

Within a stage, all tasks run independently and in parallel. Between stages, tasks depend on the output of previous stages.

2.TASK scheduler

（1）Role

作用：TaskScheduler负责在集群的工作节点上启动任务。 它从DAG调度器接收任务并将其分配给执行器进程。

The TaskScheduler is responsible for launching tasks on the cluster’s worker nodes. It receives tasks from the DAG Scheduler and assigns them to executors.

（2）Task Execution

执行：TaskScheduler处理任务的调度，监控它们的执行，并处理故障。 它确保任务被分发并在可用的集群资源上运行。

TaskScheduler handles the scheduling of tasks, monitors their execution, and deals with failures. It ensures that tasks are distributed and run across the available cluster resources.

（3）Data Locality

局部性（Locality）： TaskScheduler试图将任务放在数据已经存在的执行器节点上，以减少网络传输。

TaskScheduler tries to place tasks on executors where the data already exists to reduce network transfer.

3.Job

作业由[ActiveJob]表示，是提交给调度器的顶级工作项。例如,
当用户调用一个操作操作函数（例如count()）时，作业将被提交
submitJob。每个作业（job）可能需要执行多个阶段（Stage）来构建中间数据。

Job is represented by [ActiveJob] and is the top-level work item submitted to the scheduler. For example,when the user calls an action operation function (such as count()), the job will be submitted through submitJob. Each job (Job) may need to perform multiple stages (Stage) to build intermediate data.

4.Stages and Tasks

（1）Stages

Spark中的阶段（stage）是一组可以一起执行的任务的集合。各阶段由shuffle操作分开。

A stage in Spark is a collection of tasks that can be executed together. Stages are separated by shuffle operations.

（2）Tasks

Spark中最小的工作单元，用于操作RDD的单个分区。有两种类型的任务。

The smallest unit of work in Spark, which operates on a single partition of an RDD. There are two types of tasks:

ShuffleMapTask

生成中间结果，这些中间结果会被混洗到下一阶段（为混洗做好准备）。

Produces intermediate results that will be shuffled to the next stage (prepare it for shuffle).
reduceByKey(), groupByKey()，既然shuffle，说明数据需要跨节点移动，为宽依赖

ResultTask

生成阶段的最终结果，并返回给驱动程序。

Produces the final result of a stage, which is returned to the driver.
Collect(), saveAsTextFile()

（二）Dependency Types 依赖类型

1.ShuffleDependency 宽依赖

父RDD的每个分区可以被多个子分区使用，这就需要通过网络进行数据混洗。将作业分成多个阶段。shuffle依赖的存在会触发新阶段的创建。

Each partition of the parent RDD can be used by multiple child partitions, requiring a shuffle of data across the network.Causes the job to be split into multiple stages.The presence of a ShuffleDependency triggers the creation of a new stage.

例如：groupByKey, reduceByKey, join

2.NarrowDependency 窄依赖

父RDD的每个分区最多只能被子RDD的一个分区使用。 像map、filter和union这样的操作会创建窄依赖。这些依赖关系不会导致阶段边界，并且包含在同一个阶段中。

例如：Map, FlatMap, Filter

3.两种依赖的区分

窄依赖 = “各忙各的”（分区独立处理，高效并行）
窄依赖使得Spark可以流水线化执行多个操作
宽依赖 = “先集中，再分配”（需要shuffle）
必须等待前面所有任务完成

二、Spark Deployment Architecture

（一）整体介绍

从部署图中可以看到，整个集群分为Master node和Worker node，相当于Hadoop的Master node和Slave node。 主进程驻留在主节点上，负责管理所有工作节点。 Worker进程位于Worker节点上，负责与Master节点通信并管理执行程序。

驱动程序的官方解释是“运行应用程序的main()函数并创建SparkContext的过程”。 应用程序是用户自己编写的Spark程序（驱动程序），如WordCount.scala。 如果驱动程序在主控机上运行，如在主控机上运行

（二）组成部分

1.Cluster Overview

集群分工: 集群分为主节点和工作节点，类似于Hadoop的主节点和从节点。

Cluster Division:The cluster is divided into Master and Worker nodes, similar to Hadoop’s Master and Slave nodes.

2.Master Node

主节点： 驻留在主节点上，管理所有工作节点。

Master Daemon:Resides on the Master node.Manages all Worker nodes.

3.Worker Node

**工作节点：**驻留在工作节点上。与主节点通信。管理 Executors。
Worker Daemon：Resides on Worker nodes.Communicates with the Master node.Manages Executors.

4.Driver

驱动程序运行应用程序的main（）函数并创建SparkContext。
该应用是用户编写的Spark程序，如WordCount.scala。

The Driver runs the main() function of the application and creating the SparkContext.
The application is the user-written Spark program, such as WordCount.scala.

5.Deployment

(1)在主节点上运行

/bin/run-example SparkPi 10

SparkPi在Master上充当驱动程序。

(2)在YARN集群中

驱动程序可能运行在一个工作节点上

(3)直接在PC上运行

如果直接在PC上运行，则驱动程序会在PC上运行。如：

val sc = new SparkContext("spark://master:7077", "AppName")

但是，不建议这样做，因为如果PC和worker不在同一个本地网络上，可能会导致通信缓慢。

this is not recommended due to potential slow communication if the PC and Workers are not on the same local network.

6.Executor

（1）ExecutorBackend过程:

每个Worker上都存在一个或多个ExecutorBackend进程。

One or more ExecutorBackend processes exist on each Worker.

每个进程都包含一个Executor对象，该对象持有一个线程池，每个线程都能够执行一个任务。

Each process contains an Executor object that holds a thread pool, with each thread capable of executing a task.

在独立模式下，ExecutorBackend被实例化为一个CoarseGrainedExecutorBackend进程。

In Standalone mode, ExecutorBackend is instantiated as a CoarseGrainedExecutorBackend process.

每个Worker通常运行一个CoarseGrainedExecutorBackend进程。在运行多个应用程序时，可能会产生多个进程，但这种设置可能需要进一步的试验来确认。

Each Worker typically runs one CoarseGrainedExecutorBackend process. Multiple processes might be generated when running multiple applications, though this setup might require further experimentation to confirm.

（2） Application Structure:

每个应用都由一个驱动器程序和多个执行器进程组成。
Each application consists of a Driver and multiple Executors.

运行在每个Executor中的task都属于同一个应用程序。
Tasks running in each Executor belong to the same application.

import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
import seaborn as sns
import warnings
warnings.filterwarnings('ignore')
import  ssl
ssl._create_default_https_context = ssl._create_unverified_context

---

三、Spark Job Execution

典型的Job逻辑执行图如上图所示，最终的执行结果可以通过以下四个步骤获得：

（一）步骤

1.Reading Data and Creating the Initial RDD:读取数据并创建初始化的RDD

(1)Data Source

数据来源多种多样，如本地文件、内存数据结构、HDFS、HBase等。

Data Source: The data can come from various sources such as local files, in-memory data structures, HDFS, HBase, etc.

(2) create RDD

初始的RDD是通过parallelize（）、textFile（）或特定的数据源连接器从源中读取数据创建的。例如，parallelize（）用于从内存中的集合创建RDD。

The initial RDD is created by reading data from the source using methods like parallelize(), textFile(), or specific data source connectors. For instance, parallelize() is used to create an RDD from an in-memory collection.

val sc = new SparkContext(conf)
val data = sc.textFile("hdfs:///data/input.txt")

2.Transformation Operations on RDD::RDD上的转换操作

(1)转化概念（transformation）

对初始的RDD执行一系列转化操作。每个转换都会生成一个新的RDD。转化操作是惰性操作，这意味着它们不会立即执行，而是会形成一个RDD转化操作的谱系**。

oA series of transformation operations are performed on the initial RDD. Each transformation generates a new RDD. Transformations are lazy operations, meaning they are not executed immediately but build up a lineage of RDD transformations.

(2)转换类型

map（）、filter（）、flatMap（）、reduceByKey（）等。每个转换操作都会生成一个包含不同类型数据的新RDD。

val words = data.flatMap(line => line.split(" "))
val pairs = words.map(word => (word, 1))
val wordCounts = pairs.reduceByKey(_ + _)

3.Action Operations on the Final RDD:

（1）action 概念

动作触发转换的执行。 当对最终的RDD执行一个Action时，整个转化操作的DAG就会被执行。（只有aciton触发 才会执行 惰性）

Actions trigger the execution of the transformations. When an action is performed on the final RDD, the entire DAG of transformations is executed.

（2）Execution 执行

执行：RDD的每个分区都会根据定义的转化操作计算出来，然后合并结果。

操作类型：count（）、collect（）、saveAsTextFile（）等。

val result = wordCounts.collect()

4.Returning Results and Final Computation:

(1)结果收集 Final Computation:

将操作的结果返回到驱动程序。例如，collect（）返回整个数据集给驱动程序，而count（）返回元素的个数。

Result Collection: The result of the action is returned to the driver program. For example, collect() returns the entire dataset to the driver, while count() returns the number of elements.

(2)最终计算 Final Computation

对收集到的结果进行的任何最终计算或操作都可以在驱动器程序中执行。这可以包括进一步处理、聚合或将结果保存到外部存储系统。

Any final computations or operations on the collected results can be performed within the driver program. This can include further processing, aggregation, or saving the results to an external storage system.

result.foreach(println)

四、Spark Case Analysis 实例分析

Spark使用RDD （Resilient Distributed Datasets）实现数据存储。 RDD可以理解为在集群服务器的内存中以分布式形式存储大型数据集合。 从物理上讲，RDD将一个大数据集划分为一系列数组。 这些数组以分布式的方式存储在集群中的每个节点上。 每个数据块都有一个标识符，称为BlockID。 这样，就可以通过记录每个数据块的元数据（Meta data）来管理数据块。 每个数据块可以存储在节点的内存中，也可以持久化在硬盘上。 为了便于组织和处理，这些数据块在逻辑上被划分成多个分区（Partition）。 用户可以使用一系列转换操作符来处理rdd，并使用动作操作符来触发实际操作。 根据相应的转换计算过程，Spark在输入RDD的分区数据上计算出新的结果RDD。 通过这种方式，用户可以以类似于编写独立程序的方式执行分布式计算。

（一）四种操作符

1.创建操作符（Creation）

用于在内存中创建集合或用外部文件作为RDD对象。

used to create a collection or external files in memory as RDD objects

2.转换操作符（Transformation）

用于将一个RDD转换为另一个RDD。

used to transform one RDD into another RDD.

3.缓存操作符（Cache）

用于将RDD缓存到磁盘或内存中，以便后续计算可以重用。
used to cache RDD in disk or memory, so that subsequent calculations can be reused.

4.Action操作符（Action）

触发Spark作业执行，并将计算结果RDD保存为Scala集合或标量，或者保存到外部文件或数据库中。

trigger Spark job execution and save the calculation result RDD as a Scala collection or scalar, or save it to an external file or database.

五、Spark Shuffle Principle

（一）什么是shuffle

Shuffle描述了从map任务输出数据到reduce任务输入的过程。 Shuffle是Map和Reduce之间的桥梁。 Map输出必须经过shuffle的链接才能在Reduce中使用。 shuffle的性能直接影响到整个程序的性能和吞吐量。 因为在分布式情况下，reduce任务需要跨节点拉取其他节点上map任务的结果。 这个过程会产生网络资源的消耗和内存、磁盘IO的消耗。

Shuffle describes the process of data output from map task to input of reduce task. Shuffle is the bridge between Map and Reduce. Map output must go through the link of shuffle to be used in Reduce. The performance of shuffle directly affects the performance and throughput of the entire program. Because in a distributed situation, the reduce task needs to pull the results of the map task on other nodes across nodes.
This process will produce network resource consumption and memory, disk IO consumption.

shuffle一般分为两个部分：Map阶段的数据准备和Reduce阶段的数据拷贝处理。 map侧的Shuffle通常称为Shuffle Write， Reduce侧的Shuffle称为Shuffle Read。

Shuffle is divided into two parts: data preparation in Map phase and data copy processing in Reduce phase.Shuffle on the map side is generally called Shuffle Write, and Shuffle on the Reduce side is called Shuffle Read.

（二）Spark shuffle

Spark shuffle是Apache Spark中的一种机制，它将数据重新分布到不同的分区中，以执行需要对数据进行分组或重新分区的操作。 它是Spark分布式数据处理的重要组成部分，但由于集群中节点之间的数据移动，它经常带来显著的性能开销。

Spark shuffle is a mechanism in Apache Spark that redistributes data across different partitions to perform operations that require grouping or re-partitioning the data. It’s an essential part of distributed data processing in Spark but is often associated with significant performance overhead due to the movement of data between nodes in a cluster.

当Spark需要在集群的节点间重组数据时，就会发生shuffle。当使用下面这样的变换时，就会发生这种情况：

groupByKey, reduceByKey, aggregateByKey, join, repartition

在Spark中，负责shuffle过程的执行、计算和处理的组件主要是ShuffleManager，即shuffle管理器。 随着Spark的发展，ShuffleManager有了HashShuffleManager和SortShuffleManager两种实现方法

In Spark, the component responsible for the execution, calculation, and processing of the shuffle process is mainly ShuffleManager, which is the shuffle manager. With the development of Spark, ShuffleManager has two implementation methods, namely HashShuffleManager and SortShuffleManager, so Spark Shuffle has Hash Shuffle and Sort Shuffle.

（三）ShuffleManager的发展历史

在Spark版本的开发过程中，ShuffleManager不断迭代，越来越先进。 在Spark 1.2之前，默认的shuffle计算引擎是HashShuffleManager。 ShuffleManager和HashShuffleManager有一个非常严重的缺点，那就是会生成大量的中间磁盘文件，然后进行大量的磁盘IO操作，影响性能。

因此，在Spark 1.2以后的版本中，默认的ShuffleManager更改为SortShuffleManager。 与HashShuffleManager相比，SortShuffleManager有一些改进。 主要原因是，当每个Task执行shuffle操作时，虽然也会生成更多的临时磁盘文件，但最终所有临时文件都会合并到一个磁盘文件中，因此每个Task只有一个磁盘文件。 当下一阶段的shuffle读取任务提取自己的数据时，根据索引读取每个磁盘文件中的部分数据就足够了。

（四）重点：SortShuffleManager

SortShuffleManager是Apache Spark默认的洗牌管理器， 用于优化数据洗牌的过程，即在各个阶段之间跨分区重新分配数据。它是Spark在reduceByKey、join（宽依赖）或其他需要重组数据的操作中处理数据移动的关键部分。

The SortShuffleManager is the default shuffle manager in Apache Spark, designed to optimize the process of data shuffling, which is the redistribution of data across partitions between stages. It is a key part of how Spark handles data movement during operations like reduceByKey, join, or any operation that requires reshuffling data.

Spark中的SortShuffleManager主要有两种操作机制。

1.Ordinary Operation Mechanism 普通运行机制

这是SortShuffleManager的标准工作方式。 它在数据混洗时将数据写入磁盘之前按键排序。当任务需要稍后访问数据时，这有助于提高效率，因为它最小化了文件数量，优化了读取性能。

This is the standard way SortShuffleManager works.It sorts the data by key before writing it to disk during a shuffle. This helps improve efficiency when tasks need to access the data later, as it minimizes the number of files and optimizes read performance.

它在保存数据之前按键组织数据。你可以把它想象成在把文件放入盒子之前先把它们分类到文件夹中。这使得以后更容易找到你需要的东西。因为数据是有序的，所以当你需要再次使用它时，读取它所需的时间更短。

It organizes the data by key before saving it. Think of it like sorting your files into folders before putting them in a box. This makes it easier to find what you need later.Because the data is sorted, it takes less time to read it when you need to use it again.

2.Bypass Operation Mechanism 旁路运行机制

当分区数量低于某个阈值（默认值是200）时，将使用这种机制。

Spark开发者发现：当分区数超过200时，排序带来的读取性能提升，超过了排序的时间成本！所以选了200作为默认阈值。

This mechanism is used when the number of partitions is below a certain threshold (default is 200).

它不排序数据，而是直接使用基于散列的方法将数据写入磁盘。这减少了较小数据集的开销，在不需要排序的情况下使过程更快。

Instead of sorting the data, it directly writes the data to disk using a hash-based approach. This reduces overhead for smaller datasets, making the process quicker when sorting isn’t necessary.

它不会对数据进行排序，而是直接将其放入文件中，而不会首先对其进行组织。这就像把你的文件扔进一个盒子里，但没有分类，如果你没有太多的话。这节省了时间和精力。

Instead of sorting the data, it just puts it directly into files without organizing it first. It’s like throwing your files into a box without sorting them if you don’t have too many. This saves time and effort.

“多就排序，少就直给”
分区多（>200）→ 用普通机制（排序）
分区少（≤200）→ 用旁路机制（不排序）

---

总结

本章节主要讲解了Spark工作流程的核心步骤及要点，例如：为什么要使用DAG？以及部署架构、Shuffle的使用等等