为什么 Spark 慢？-Python教程-PHP中文网

为什么 Spark 慢？

Mary-Kate Olsen

发布： 2024-12-11 19:43:11

原创

888 人浏览过

为什么 Spark 慢？

从一个引人注目的标题“Spark 为什么这么慢？”开始，值得注意的是，称 Spark“慢”可能意味着多种含义。聚合速度慢吗？数据加载？存在不同的情况。此外，“Spark”是一个广泛的术语，其性能取决于编程语言和使用上下文等因素。因此，在深入讨论之前，让我们将标题改进得更加精确。

由于我主要在 Databricks 上使用 Spark 和 Python，因此我将进一步缩小范围。

优化后的标题将是：

“Spark 的第一印象：‘听说它很快，但为什么感觉很慢？’初学者的视角”

写作动机（随意的想法）

作为广泛使用 pandas、NumPy 和机器学习库的人，我钦佩 Spark 通过并行和分布式处理处理大数据的能力。当我终于在工作中使用 Spark 时，我对它看起来比 pandas 慢的场景感到困惑。不确定出了什么问题，我发现了一些见解并想与大家分享。

你的火花什么时候会变慢？

在进入主题之前

我们简单介绍一下Spark的基本架构。

Why Is Spark Slow??

（集群模式概述）

Spark 集群由执行实际处理的 工作节点和协调和计划执行的驱动程序节点组成。这种架构会影响下面讨论的所有内容，因此请记住这一点。

现在，进入要点。

1. 数据集不够大

Spark 针对大规模数据处理进行了优化，但它也可以处理小型数据集。然而，看看这个基准：

Why Is Spark Slow??

（在单节点机器上对 Apache Spark 进行基准测试）

结果表明，对于 15GB 以下的数据集，pandas 在聚合任务中优于 Spark。为什么？简而言之，Spark 优化的开销超过了小数据集的好处。

该链接显示了 Spark 并不慢的情况，但这些情况通常处于本地集群模式。对于独立设置，由于节点之间的网络通信开销，较小的数据集可能是一个缺点。

pandas：在一台机器上处理内存中的所有内容，无需网络或存储 I/O。
Spark：使用 RDD（弹性分布式数据集），涉及 Workers 之间的网络通信（如果分布式），并会在组织数据以进行并行处理时产生开销。

2. 理解惰性求值

Spark 采用惰性求值，这意味着转换不会立即执行，而是推迟到某个操作（例如收集、计数、显示）触发计算为止。

示例（熊猫）：

df = spark.read.table("tpch.lineitem").limit(1000).toPandas()
df["l_tax_percentage"] = df["l_tax"] * 100
for l_orderkey, group_df in df.groupby("l_orderkey"):
    print(l_orderkey, group_df["l_tax_percentage"].mean())

登录后复制

执行时间：3.04秒

Spark 中的等效项：

from pyspark.sql import functions as F
sdf = spark.read.table("tpch.lineitem").limit(1000)
sdf = sdf.withColumn("l_tax_percentage", F.col("l_tax") * 100)

for row in sdf.select("l_orderkey").distinct().collect():
    grouped_sdf = sdf.filter(F.col("l_orderkey") == row.l_orderkey).groupBy("l_orderkey").agg(
        F.mean("l_tax_percentage").alias("avg_l_tax_percentage")
    )
    print(grouped_sdf.show())

登录后复制

执行时间：3分钟后仍在运行。

为什么？

惰性求值：所有转换都会排队，并且仅在表演等动作期间执行。
Driver 到 Worker 的通信：收集和显示等操作涉及从 Workers 到 Driver 的数据传输，导致延迟。

Spark 代码在 pandas 中有效地执行了此操作：

for l_orderkey, group_df in df.groupby("l_orderkey"):
    df["l_tax_percentage"] = df["l_tax"] * 100
    print(l_orderkey, group_df["l_tax_percentage"].mean())

登录后复制

通过使用 Spark 的缓存或重构逻辑以尽量减少重复计算来避免此类模式。