AI 基本要素
你跑火花任务时会发生什么？
在 Google 上的首选来源中添加 Unite.AI
导言
随着数据集的增长，由于CPU容量，内存和存储吞吐量的限制，在单机上高效处理它们会变得具有挑战性。
Apache Spark通过提供分布式数据处理引擎来应对这些挑战，该引擎可以将工作量划分为分区，并同时执行跨CPU核心和多个机器的任务。
然而，Spark不仅仅是要增加更多的机器。
了解其各组成部分——司机、执行者、区域数据数据库、数据库、分区、工作、阶段和任务——如何合作，对于设计有效和可靠的数据管道至关重要。
在文章中，我会用一个实用的电子商务销售汇总例子
1. 为什么Apache Spark - 并行数据处理
Apache Spark是一个分布式计算引擎，为大规模数据处理和数据工程设计。
Spark 处理利用并行数据处理的大数据：
与单一节点处理(右，图A)相比，Spark将工作分成使用多个节点的任务，在有资源时，在机器之间执行。
在这里，分布并不意味着多个机器是强制性的。 Spark可以在单机上以局部模式运行。 即使在那里，数据也可以平行分割和处理。
2. 核心Apache Spark架构
Spark由几个关键组件组成：
- Spark驱动程序(驱动程序):Spark应用程序的中央协调进程。
- 集群管理器：一个负责分配CPU核心和内存等硬件资源的外部服务。
- 工人节点(worker)：一个物理/虚拟机或计算集群中的节点。
- 执行者：在工人节点上分配工人加工发布。
- Java虚拟机(JVM)：执行 Spark 字节代码所需的运行时间环境( 因为 Spark 写于 Scala/ Java )。
首先，驱动程序协调一个Spark应用程序，并将其转换为物理执行任务的定向循环图(DAG)，并把它们排到执行者之间。
然后，群组管理员根据部署的资源管理系统分配资源。
工作节点为执行者提供计算资源，执行者在执行任务时执行任务并保存数据。
一些管理环境如Databricks使用每个工人一个执行人，但这不是通用的Spark规则。 工人可以根据组群配置，托管多个执行器。
2-1. 电子商务使用案例-输入数据结构
让我们想象一下一家精品电子商务公司。
该公司有1万份销售记录，每份都有库存股和销售数量。 假设三种可能的SKU:A,B和C.
表1. 电子商务销售记录细目和预期汇总实例。
我们的目标是计算每个SKU销售的总有效单位：
- 过滤记录谁的Qty QQ0，然后
- SKU的汇总样本。
在说明性完整的数据集中，假设在过滤后，10 000份记录中仍有9 800份保留。 Spark预计最终总数为：
[("A", 8000), ("B", 10500), ("C", 9200)]
2-2. 火花组
我将定义一个简化的集群
工人节点(工人)
我将设置3个工人节点，每个节点都有8个vCPU的例型
该系统总共提供了24个VCPU。
Spark 执行器和CPU/记忆分配
我将保证：
- 7名执行人。
- 每名执行员2个vCPU核心。
- 3GiB JVM堆积每个执行者的记忆。
Spark应用程序要求集群管理器将资源分配给这7名执行者。
然后，集群管理者根据每个工人的资源使用情况决定CPU和内存分配。
例如，它决定创建14个任务槽 – 每个执行者 2 vCPU 核心 x 7 执行者：
请注意，分配给执行者的3 GiB内存并不是其全部内存，因为它必须保证工人的内存管理费和其他处决。
假设单项任务需要1 vCPU(spark.task.cpus = 1)，每个工人分配其 vCPU:
此显示两个关键点：
每个工人都有其独特的CPU使用率(在图C-2中，工人A的使用率为75%，而工人B和C仍然是50%)。
- 尽管该系统有24个vCPU核心(8个vCPU/工人x3个工人)，但只能使用14个vCPU核心；其余10个仍然未用于这些任务。
2-3. Java虚拟机(JVM)在Spark后面
Apache Spark在Java虚拟机(JVM)上构建。
JVM 是执行 Spark 字节代码所需的运行时环境。
因为Spark在斯卡拉/贾瓦写道：
- 在应用中使用Scala或Java时：代码在JVM内部本土运行。
- 当使用PySpark(Python)：一个Python进程运行代码，但它通过像Py4J这样的桥梁与基于JVM的驱动程序进行通信。 工人节点上的Spark执行器仍然运行在一个JVM内部，然后管理Python工人执行Python代码和Python用户定义函数(Python UDFs).
无论在哪种情况下，JVM管理执行线程，内存分配，以及对驱动器和执行器的垃圾收集。
PySpark 应用程序中的驱动程序( E) JVMs
在PySpark应用程序中，Python驱动程序与运行在JVM中的Spark驱动程序进行通信；
- Python驱动程序：应用程序级的PySpark代码被写入和执行的地方。
- 驾驶员JVM：坐标Spark执行和调度。
- 执行者JVM：运行Spark任务并管理执行资源。 当任务需要Python执行时，涉及Python工人过程。
- 客户端JVM(客户流程或提交机):JVM运行于本地机上或边缘/通道节点。
如图所示，客户端JVM首先与驱动程序JVM连接，后者随后与执行者JVMs协调：
从这里开始，驱动程序JVM托管Spark Driver来管理应用程序，而执行者JVMs托管负责执行这些任务的单个执行者。
每个JVM驱动程序都有JVM堆积，可以高效地处理数据处理，缓存，以及用户对象。
2-4. 弹性分布数据集(RDD)深潜
一个可逆分布的数据集(RDD)是Spark可以并行处理的元素的不可改变，分割的集合。
RDDs跟踪变换之间的依赖性，使得Spark能够在许多失败情景中重算丢失的分区。
名称抓取三个属性：
- 耐火性：Spark记录线条——RDD取决于以前的RDD——并经常可以重建缺失的分区，而不是维持每个中间结果的复制品。
- 分布：数据集作为Spark可能在不同执行器上处理的分区表示。 不需要多个物理机器；本地模式也支持RDD.
- 数据集：RDD可以包含密钥值对，整数，字符串，扭矩，以及其他支持的元素——不仅仅是关系行。 RDD API包括地图，过滤器，减少ByKey，加入，以及聚合ByKey等操作。
RDD是逻辑的集合和计算抽象，而不是声明特定分区永久分配给特定工人。
分区
一个分区是数据的一个逻辑块，通常对应一个阶段中的一个任务。
我们创造100个分区(这些100个分区是明确的教学选择)
拥有10,000个源元素，如果输入分配大致均匀，则每个分区给出约100个元素：
3. RDD API vs DataFrame API：内部催化剂优化
最初的RDD示例帮助理解世系，分区和洗涤。
然而在典型的结构化ETL中，DataFrame API是首选的起点，因为它提供了列，schema信息，SQL类似表达式，以及优化的可见操作。
下图说明这种关系：
就像RDD有RDD线程来定义变换，DataFrame查询计划代表其结构化查询。
Spark从依赖和物理操作符构建执行工作。
DataFrame执行可能内部使用基于RDD的机制，但这并不意味着用户在写入DataFrame代码时需要手动构建RDD行。
3–1. 数据Frame查询计划
一个DataFrame查询计划是执行引擎生成的逻辑和物理步骤的顺序(图L中的purple框).
DataFrame 查询通过多个表达式移动：
- 逻辑计划：对请求的操作(扫描、过滤、组、聚合)的说明。
- 分析并优化逻辑计划：Spark 解决引用并应用适用的逻辑优化，如列普run或上游推倒。
- 物理计划：Spark选择可执行运算符，例如扫描、部分/最终聚合、交换、加入和类型。
- 执行：实际计划采用Spark的执行机制，最终涉及执行人的任务。
在我们的使用情况下，物理计划可以包括以下步骤：
- 部分散列聚合
- SKU的交换
- 最终散列集合。
3-2. 比较动作：RDD API对 DataFrame API
让我们比较RDD API和DataFrame API，以SKU汇总为例。
RDD API 软件
元素为(sku,qty)对。 我们的 Python 聚合逻辑是明确的，Spark 遵循 RDD 依赖性：
结果 rd=(
原始编号
.filter(lambda 行：行[1] > 0)
减少ByKey(lambda a, b: a + b, nummPartities=4)
)
rd 结果=结果 rd.collect ()
# 实例：
# [('A', 8000), ('B', 10500), ('C', 9200)]
数据Frame API
假设销售 df有列sku(字符串)和qty(数字):
从 pyspark.sql 导入函数为 F
结果 df=(
销售 df
.filter(F.col("qty") > 0)
. groupBy ("sku") (中文(简体) ).
.agg(F.sum("qty").alias("total_qty"))
)
df 结果=结果 df.collect ()
# 实例：
# [Row(sku='A'，共计 qty=8000) #
(sku='B'，共计 qty=10500)
(sku='C'，共计 qty=9200)].
在这里，这两个例子解决了同样的商业问题：
按 SKU 过滤无效销售记录并汇总有效数量。
在这两种情况下，收集()触发Spark执行，并将所有产生的记录退回给驱动程序。
这里的关键区别在于结果表述：
- RDD API：返回 Python 元素，如( sku, total qty) tuples。
- DataFrame API：回归结构带有命名字段的行对象，如 sku 和 total qty。
以下是RDD和DataFrame API的总体比较：
表2. RDD API vs. DataFrame API – 特征和抽象比较：(由Kuriko IWAI创建)
4. 包装
Spark调制的目的不是将分区计数或CPU的使用独立最大化。
这是在不造成不稳定的情况下，减少端到端的执行成本和延迟。
在本篇文章中，我们了解了该系统的核心关系：
- 资源：工人主机执行器；执行器运行任务；配置了CPU和内存限制的货币。
- RDD编程：转型产生不可变的RDD定义和线程；动作触发计算。
- DataFrame编程：转型产生逻辑查询操作；优化选择可执行的物理运算符；动作触发计算。
- 执行：Spark将工作组织成工作、阶段和任务，以及执行者的任务处理分区。 分散和其他依赖性在各阶段之间形成了协调的界限。
了解Spark如何执行分配的工作量只是基础。
可以使用RDD API学习七步计算流程。
本文由 AI图鉴自动抓取并翻译，版权归原作者所有；如需查看原文请点下方链接。
AI Fundamentals
What Actually Happens When You Run a Spark Job?
Add Unite.AI to your preferred sources on Google
Introduction
As datasets grow, processing them efficiently on a single machine can become challenging due to limitations in CPU capacity, memory, and storage throughput.
Apache Spark addresses these challenges by providing a distributed data processing engine that can divide workloads into partitions and execute tasks in parallel across CPU cores and multiple machines.
However, Spark is not simply about adding more machines.
Understanding how its components work together—Drivers, Executors, RDDs, DataFrames, Partitions, Jobs, Stages, and Tasks—is essential to designing efficient and reliable data pipelines.
In this article, I’ll walk through Spark’s architecture, execution mechanics, and key performance considerations using a practical e-commerce sales aggregation example.
1. Why Apache Spark – Parallel Data Processing
Apache Spark is a distributed computing engine designed for large-scale data processing and data engineering.
Spark processes big data leveraging parallel data processing:
Compared with a single node processing (right, Figure A), Spark divides work into tasks using multiple nodes, executing them across resources, when available, across machines.
Here, distributed does not mean multiple machines are mandatory. Spark can run in local mode on a single machine. Even there, data can be partitioned and processed in parallel.
2. Core Apache Spark Architecture
Spark consists of several key components:
- Spark driver (driver): The central coordination process of Spark application.
- Cluster manager: An external service responsible for allocating hardware resources like CPU cores and memory.
- Worker nodes (worker): A physical / virtual machine or compute node in the cluster.
- Executors: Distributed worker processes launched on worker nodes.
- Java Virtual Machine (JVM): The runtime environment required to execute Spark’s bytecode (since Spark is written in Scala/Java).
First, the driver coordinates a Spark application and transform it into a directed acyclic graph (DAG) of physical execution tasks, and schedules them across the executors.
Then, cluster manager allocates resources under the deployment’s resource-management system.
Worker nodes supply compute resources to executors, which execute tasks and hold data during the task execution.
Some managed environments like Databricks use one executor per worker, but this is not a universal Spark rule. A worker can host multiple executors depending on cluster configuration.
2-1. E-commerce Use Case – Input Data Structure
Let us imagine a boutique e-commerce company.
It has 10,000 sales records, each with a stock keeping unit (SKU) and quantity sold. Assume three possible SKUs: A, B, and C.
Table 1. E-Commerce sales record breakdown & expected aggregations example.
Our goal is to calculate total valid units sold per SKU:
- Filter out records whose qty <= 0, then
- Aggregate samples by SKU.
In the illustrative full dataset, suppose 9,800 of the 10,000 records remain after filtering. The final totals expected from Spark are:
[("A", 8000), ("B", 10500), ("C", 9200)]
2-2. Spark Cluster
I’ll define a simplified cluster for our use case.
Worker Nodes (Workers)
I’ll set 3 worker nodes, each of which has an instance type of 8 vCPU with 32 GiB RAM:
In total, 24 vCPU are available in the system.
Spark Executors and CPU/Memory Allocation
Let’s say, I’ll secure:
- 7 executors.
- 2 vCPU cores for each executor.
- 3GiB JVM heap memory for each executor.
Spark application requests the cluster manager to allocate resources to these 7 executors.
Then, the cluster manager decides CPUs and memory allocations based on the resource usage of each worker.
For instance, it decides to create 14 tasks slots – 2 vCPU core per executor x 7 executors:
Note that the 3 GiB memory allocated to Executor is not its total memory because it has to secure memory overhead for Workers and other executions.
Assuming that a single task requires 1 vCPU (spark.task.cpus = 1), each worker allocates its vCPU:
This shows two key points:
- Each worker has its unique CPU usage rate (In Figure C-2, Worker A’s usage is 75%, while Workers B and C remain 50%).
- Even though the system has 24 vCPU cores (8 vCPU/worker x 3 Workers), it can only use 14 vCPU cores; the rest 10 remain unused for these tasks.
2-3. Java Virtual Machine (JVM) Behind Spark
Apache Spark is built on the Java Virtual Machine (JVM).
JVM is a runtime environment required to execute Spark’s bytecode.
Since Spark is written in Scala/Java:
- When using Scala or Java in application: The code runs natively inside the JVM.
- When using PySpark (Python): A Python process runs the code, but it communicates with a JVM-based driver via a bridge such as Py4J. The Spark executors on the worker nodes still run inside a JVM, which then manages Python workers to execute Python code and Python User-Defined Functions (Python UDFs).
In either case, the JVM manages execution threads, memory allocation, and garbage collection for both the driver and executors.
Driver & Executor JVMs in PySpark Application
In a PySpark application, the Python driver communicates with the Spark driver running in the JVM;
- Python driver: Where application-level PySpark code is written and executed.
- Driver JVM: Coordinates Spark execution and scheduling.
- Executor JVM: Runs Spark tasks and manages execution resources. Python worker processes are involved when tasks require Python execution.
- Client JVM (client process or submission machine): JVM running on the local machine or an edge/gateway node.
As shown in the figure, the client JVM first connects to the driver JVM, which subsequently coordinates with the executor JVMs:
From here, the driver JVM hosts the Spark Driver to manage the application, while the executor JVMs host the individual executors responsible for executing those tasks.
Each JVM driver has JVM heap to handle data processing, caching, and user objects efficiently.
2-4. Resilient Distributed Datasets (RDDs) Deep Dive
A Resilient Distributed Dataset (RDD) is an immutable, partitioned collection of elements that Spark can process in parallel.
RDDs track dependencies between transformations, allowing Spark to recompute lost partitions in many failure scenarios.
The name captures three properties:
- Resilient: Spark records lineage—which RDD depends on which previous RDD—and can often reconstruct missing partitions instead of maintaining replicas of every intermediate result.
- Distributed: The dataset is represented as partitions that Spark may process on different executors. Multiple physical machines are not required; local mode also supports RDDs.
- Dataset: An RDD can contain key-value pairs, integers, strings, tuples, and other supported elements—not only relational rows. RDD APIs include operations such as map, filter, reduceByKey, join, and aggregateByKey.
An RDD is a logical collection and computation abstraction, not a declaration that particular partitions are permanently assigned to particular workers.
Partitions
A partition is a logical chunk of data and normally corresponds to one task within a stage.
Let’s say we create 100 partitions (These 100 initial partitions are an explicit teaching choice).
With 10,000 source elements, it gives approximately 100 elements per partition if the input is distributed roughly evenly:
3. RDD API vs DataFrame API: Internal Catalyst Optimization
The original RDD example helps understand lineage, partitions, and shuffle.
In typical structured ETL, however, the DataFrame API is the preferred starting point because it provides columns, schema information, SQL-like expressions, and optimizer-visible operations.
Below diagram illustrates the relationship:
Just like RDD has RDD lineage to define the transformation, DataFrame query plan represents its structured query.
Spark constructs execution work from dependencies and physical operators.
DataFrame execution may use RDD-based mechanisms internally, but that does not mean users need to manually build RDD lineage when writing DataFrame code.
3-1. DataFrame Query Plan
A DataFrame query plan is the sequence of logical and physical steps an execution engine generates (purple box in Figure L).
A DataFrame query moves through several representations:
- Logical plan: A description of the requested operations (scan, filter, group, aggregate).
- Analyzed and optimized logical plan: Spark resolves references and applies applicable logical optimizations, such as column pruning or predicate pushdown.
- Physical plan: Spark chooses executable operators, e.g. scans, partial/final aggregates, exchanges, joins, and sorts.
- Execution: The physical plan is run using Spark’s execution mechanisms, ultimately involving tasks on executors.
In our use case, the physical plan can include the following steps:
- Partial hash aggregation,
- Exchange by SKU, and
- Final hash aggregation.
3-2. Comparing Actions: RDD API vs DataFrame API
Let us compare RDD API and DataFrame API, taking the SKU aggregation for an example.
RDD API
The elements are (sku, qty) pairs. Our Python aggregation logic is explicit, and Spark follows RDD dependencies:
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
Assume sales_df has columns sku (string) and qty (numeric):
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)]
Here, both examples solve the same business problem:
Filtering invalid sales records and aggregating valid quantities by SKU.
And in both cases, collect() triggers Spark execution and returns all resulting records to the Driver.
The key difference here is in the result representation:
- RDD API: Returns Python elements, such as (sku, total_qty) tuples.
- DataFrame API: Returns structured Row objects with named fields, such as sku and total_qty.
Here’s an overall comparison of RDD and DataFrame APIs:
Table 2. RDD API vs. DataFrame API – feature & abstraction comparison:(Created by Kuriko IWAI)
4. Wrapping Up
The goal of Spark tuning is not to maximize partition count or CPU usage independently.
It is to reduce end-to-end execution cost and latency without introducing instability.
In this article, we learn the core relationships in its system:
- Resources: Workers host executors; executors run tasks; configured CPU and memory limit concurrency.
- RDD programming: Transformations produce immutable RDD definitions and lineage; actions trigger computation.
- DataFrame programming: Transformations produce logical query operations; optimization chooses executable physical operators; actions trigger computation.
- Execution: Spark organizes work into jobs, stages, and tasks, and tasks process partitions on executors. Shuffles and other dependencies create coordination boundaries among stages.
Understanding how Spark executes distributed workloads is only the foundation.
You can learn a seven-step computation flow using RDD API.
Auto-fetched and translated by AI Atlas. All rights belong to the original author.