运行 Spark 作业时，究竟会发生什么？

引言

随着数据集不断增大，受 CPU 容量、内存和存储吞吐量的限制，在单台机器上高效处理数据可能会变得越来越困难。

Apache Spark通过提供分布式数据处理引擎来应对这些挑战。该引擎可以将工作负载拆分为多个分区，并在多个 CPU 核心和多台机器上并行执行任务。

然而，Spark 并不只是增加机器这么简单。

要设计高效、可靠的数据管道，必须了解各个组件如何协同工作：Driver、Executor、RDD、DataFrame、分区、作业、阶段和任务。

本文将以一个实际的电子商务销售额汇总示例为例，介绍 Spark 的架构、执行机制和关键性能考量。

1. Apache Spark 的优势——并行数据处理

Apache Spark是一种分布式计算引擎，专为大规模数据处理和数据工程而设计。

Spark 借助并行数据处理来处理大数据：

图 A. Apache Spark 分布式集群架构概览（集群、Driver 和 Worker 节点），对比单节点执行方式（由 Kuriko IWAI 制作）

与单节点处理方式相比（右侧，图 A），Spark 会使用多个节点将工作拆分为任务，并在机器上的可用资源间分配执行。

这里的分布式并不意味着必须使用多台机器。Spark 可以在单台机器上以本地模式运行。即便如此，数据仍可分区并行处理。

2. Apache Spark 核心架构

Spark 由几个关键组件构成：

• Spark Driver（Driver）：Spark 应用程序的中央协调进程。
• 集群管理器：负责分配 CPU 核心、内存等硬件资源的外部服务。
• Worker 节点（Worker）：集群中的物理或虚拟机器，或计算节点。
• Executor：在 Worker 节点上启动的分布式工作进程。
• Java Virtual Machine (JVM)：执行 Spark 字节码所需的运行环境（因为 Spark 使用 Scala/Java 编写）。

首先，Driver负责协调 Spark 应用程序，并将其转换为由有向无环图（DAG）表示的物理执行任务，再将这些任务安排到各个Executor上。

然后，集群管理器会根据部署所采用的资源管理系统分配资源。

Worker 节点为 Executor 提供计算资源；Executor 负责执行任务，并在任务执行期间保存数据。

Databricks 等一些托管环境会在每个 Worker 上运行一个 Executor，但这并不是 Spark 的通用规则。根据集群配置，一个 Worker 可以承载多个 Executor。

2-1. 电子商务用例——输入数据结构

设想一家精品电子商务公司。

该公司有10,000 条销售记录，每条记录都包含一个库存单位（SKU）和售出数量。假设有三种可能的 SKU： A, B，以及 C.

表 1. 电子商务销售记录细分及预期汇总示例。

我们的目标是计算每种 SKU 的有效售出总件数：

1. 筛除……字段为……的记录 qty <= 0，然后
2. 按 SKU 汇总样本。

在这个用于说明的完整数据集中，假设筛选后，10,000 条记录中有 9,800 条保留下来。Spark 预期得出的最终总数如下：

[("A", 8000), ("B", 10500), ("C", 9200)]

2-2. Spark 集群

我将为这个用例搭建一个简化的集群。

Worker 节点（Worker）

我将设置 3 个 Worker 节点，每个节点的实例类型均为 8 vCPU、32 GiB RAM：

图 B. Worker 节点间的资源分配（由 Kuriko IWAI 制作）

系统总共有 24 个 vCPU 可用。

Spark 执行器与 CPU/内存分配

假设我要申请以下资源：

• 7 个执行器。
• 每个执行器分配 2 个 vCPU 核心。
• 每个执行器分配 3GiB JVM 堆内存。

Spark 应用程序会请求集群管理器为这 7 个执行器分配资源。

随后，集群管理器会根据各工作节点的资源使用情况决定 CPU 和内存的分配。

例如，它决定创建 14 个任务槽位：每个执行器 2 个 vCPU 核心 × 7 个执行器：

图 C-1. 工作节点之间的 vCPU 核心、内存和执行器分配情况，工作节点 A 存在资源不均衡

请注意，分配给执行器的 3 GiB 内存并非其全部内存，因为还必须为工作节点和其他执行预留内存开销。

假设单个任务需要 1 个 vCPU（spark.task.cpus = 1），每个工作节点分配其 vCPU：

图 C-2. 工作节点之间的 vCPU 并行分配

这说明了两个关键点：

1. 每个工作节点都有自己独特的 CPU 使用率（在图 C-2中，工作节点 A 的使用率为 75%，而工作节点 B 和 C 均为 50%）。
2. 尽管系统共有 24 个 vCPU 核心（每个工作节点 8 个 vCPU × 3 个工作节点），但只能使用其中 14 个 vCPU 核心；其余 10 个无法用于这些任务。

2-3. Spark 背后的 Java 虚拟机（JVM）

Apache Spark 构建于Java 虚拟机（JVM）之上。

JVM 是执行 Spark 字节码所需的运行时环境。

由于 Spark 使用 Scala/Java 编写：

• 在应用程序中使用 Scala 或 Java 时：代码会在 JVM 中原生运行。
• 使用 PySpark（Python）时：Python 进程会运行代码，但它会通过诸如Py4J这样的桥接机制与基于 JVM 的驱动程序通信。工作节点上的 Spark 执行器仍运行在 JVM 中，由 JVM 管理 Python 工作进程，以执行 Python 代码和Python 用户自定义函数（Python UDF）。

无论哪种情况，JVM 都会管理驱动程序和执行器的执行线程、内存分配及垃圾回收。

PySpark 应用程序中的驱动程序和执行器 JVM

在 PySpark 应用程序中，Python 驱动程序会与运行在 JVM 中的 Spark 驱动程序通信；

• Python 驱动程序：编写并执行应用程序级 PySpark 代码的地方。
• 驱动程序 JVM：协调 Spark 的执行和调度。
• 执行器 JVM：运行 Spark 任务并管理执行资源。任务需要执行 Python 代码时，会涉及 Python 工作进程。
• 客户端 JVM（客户端进程或提交任务的机器）：运行在本地机器或边缘/网关节点上的 JVM。

如图所示，客户端 JVM首先连接到驱动程序 JVM，随后由后者与执行器 JVM协调：

图 D. 客户端、驱动程序和执行器 JVM 之间的整体关系

从这里可以看到，驱动程序 JVM 托管 Spark 驱动程序以管理应用程序，而执行器 JVM 则托管各个执行器，由它们负责执行这些任务。

每个 JVM 驱动程序都拥有 JVM 堆内存，以便高效处理数据、缓存数据和管理用户对象。

2-4. 深入了解弹性分布式数据集（RDD）

弹性分布式数据集（RDD）是一种由不可变、分区化的元素集合构成的数据抽象，Spark 可以对其进行并行处理。

RDD 会跟踪转换之间的依赖关系，因此在许多故障场景下，Spark 都能重新计算丢失的分区。

这个名称体现了它的三项特性：

• 弹性：Spark 会记录血统信息，即每个 RDD 依赖于哪个前序 RDD；在许多情况下，Spark 可以重建缺失的分区，而不必为每个中间结果保留副本。
• 分布式：数据集以分区的形式表示，Spark 可以在不同的执行器上处理这些分区。不要求使用多台物理机器；本地模式也支持 RDD。
• 数据集：RDD 可以包含键值对、整数、字符串、元组以及其他受支持的元素，不仅限于关系型数据行。RDD API 包含诸如以下操作 map, filter, reduceByKey, join，以及 aggregateByKey.

RDD 是一种逻辑集合与计算抽象，并不意味着特定分区会永久分配给特定工作节点。

分区

分区是数据的逻辑分块，通常对应一个阶段中的一项任务。

假设我们创建 100 个分区（这 100 个初始分区是为了便于讲解而明确选定的）。

对于 10,000 个源数据元素，如果输入大致均匀分布，则每个分区约有 100 个元素：

图 E. 将数据并行分配到 100 个初始分区

3. RDD API 与 DataFrame API：Catalyst 内部优化

最初的 RDD 示例有助于理解血缘关系、分区和 shuffle。

不过，在典型的结构化 ETL 中，DataFrame API 是更推荐的起点，因为它提供列、架构信息、类似 SQL 的表达式，以及可供优化器识别的操作。

下图展示了它们之间的关系：

图 F. Spark 图概念之间的关系：RDD 血缘关系、DataFrame 和 Spark 执行 DAG

RDD 使用 RDD 血缘关系来定义转换，与此类似，DataFrame 查询计划则表示其结构化查询。

Spark 根据依赖关系和物理运算符构建执行任务。

DataFrame 执行在内部可能会使用基于 RDD 的机制，但这并不意味着用户编写 DataFrame 代码时需要手动构建 RDD 血缘关系。

3-1. DataFrame 查询计划

DataFrame 查询计划是执行引擎生成的一系列逻辑和物理步骤（图 L中的紫色方框）。

DataFrame 查询会经过多种表示形式：

1. 逻辑计划：描述所请求的操作（扫描、筛选、分组、聚合）。
2. 经过分析和优化的逻辑计划：Spark 会解析引用，并应用适用的逻辑优化，例如列裁剪或谓词下推。
3. 物理计划：Spark 会选择可执行的运算符，例如扫描、局部/最终聚合、交换、连接和排序。
4. 执行：物理计划通过 Spark 的执行机制运行，最终由执行器上的任务完成。

在我们的用例中，物理计划可能包括以下步骤：

• 局部哈希聚合，
• 按 SKU 进行交换，以及
• 最终哈希聚合。

3-2. 比较操作：RDD API 与 DataFrame API

下面以按 SKU 聚合为例，比较 RDD API 和 DataFrame API。

RDD API

这些元素是 (sku, qty) 对。Python 聚合逻辑由我们显式编写，Spark 则会遵循 RDD 依赖关系：

result_rdd = (
    raw_rdd
    .filter(lambda row: row[1] > 0)
    .reduceByKey(lambda a, b: a + b, numPartitions=4)
)

rdd_results = result_rdd.collect()

# Example:
# [('A', 8000), ('B', 10500), ('C', 9200)]

DataFrame API

假设 sales_df 包含以下列 sku （字符串）和 qty （数值）：

from pyspark.sql import functions as F

result_df = (
    sales_df
    .filter(F.col("qty") > 0)
    .groupBy("sku")
    .agg(F.sum("qty").alias("total_qty"))
)

df_results = result_df.collect()

# Example:
# [Row(sku='A', total_qty=8000),
#  Row(sku='B', total_qty=10500),
#  Row(sku='C', total_qty=9200)]

这里的两个示例解决的是同一个业务问题：

筛除无效销售记录，并按 SKU 汇总有效数量。

而且在这两种情况下， collect() 都会触发 Spark 执行，并将所有结果记录返回给 Driver。

这里的主要区别在于结果的表示形式：

• RDD API：返回 Python 元素，例如 (sku, total_qty) 元组。
• DataFrame API：返回结构化 Row 对象，其中包含命名字段，例如 sku 和 total_qty.

以下是 RDD API 与 DataFrame API 的整体比较：

表 2。 RDD API 与 DataFrame API——功能与抽象层对比：（由 Kuriko IWAI 制作）

4. 总结

Spark 调优的目标并非单独追求分区数量或 CPU 使用率最大化。

而是在不引入不稳定因素的前提下，降低端到端执行成本和延迟。

在本文中，我们了解了 Spark 系统中的核心关系：

• 资源：工作节点承载执行器；执行器运行任务；配置的 CPU 和内存限制并发度。
• RDD 编程：转换操作生成不可变的 RDD 定义及其血缘关系；行动操作触发计算。
• DataFrame 编程：转换操作生成逻辑查询操作；优化器选择可执行的物理算子；行动操作触发计算。
• 执行：Spark 将工作组织为作业、阶段和任务，任务在执行器上处理分区。数据混洗及其他依赖关系会在阶段之间形成协调边界。

了解 Spark 如何执行分布式工作负载，只是基础。

你可以了解使用 RDD API 的七步计算流程。