如何充分利用集群中的所有 Spark 节点？答案

【问题标题】：How to fully utilize all Spark nodes in cluster?如何充分利用集群中的所有 Spark 节点？
【发布时间】：2023-03-18 14:35:01
【问题描述】：

我已经在 Spark 的独立模式下使用 ec2-script 启动了一个 10 节点集群。我正在从 PySpark shell 中访问 s3 存储桶中的数据，但是当我在 RDD 上执行转换时，只使用了一个节点。例如下面将从 CommonCorpus 中读取数据：

bucket = ("s3n://@aws-publicdatasets/common-crawl/crawl-data/CC-MAIN-2014-23/"
          "/segments/1404776400583.60/warc/CC-MAIN-20140707234000-00000-ip-10"
          "-180-212-248.ec2.internal.warc.gz")

data = sc.textFile(bucket)
data.count()

当我运行它时，我的 10 个从站中只有一个处理数据。我知道这一点，因为从 Spark Web 控制台查看时，只有一个从站 (213) 有任何活动日志。当我查看 Ganglia 中的活动时，同一节点 (213) 是唯一在活动运行时内存使用量激增的从站。

此外，当我使用只有一个从属设备的 ec2 集群运行相同的脚本时，我的性能完全相同。我正在使用 Spark 1.1.0，非常感谢任何帮助或建议。

【问题讨论】：

标签： amazon-ec2 apache-spark pyspark

【解决方案1】：

...ec2.internal.warc.gz

我认为您在 gzip 压缩文件中遇到了一个相当典型的问题，即它们无法并行加载。更具体地说，单个 gzip 压缩文件不能由多个任务并行加载，因此 Spark 将使用 1 个任务加载它，从而为您提供一个具有 1 个分区的 RDD。

（但是请注意，Spark 可以并行加载 10 个 gzip 压缩文件；只是这 10 个文件中的每一个只能由 1 个任务加载。您仍然可以在跨个文件中获得并行性，只是不在文件中。）

您可以通过明确检查 RDD 中的分区数来确认您只有 1 个分区：

data.getNumPartitions()

可以在 RDD 上并行运行的任务数的上限是 RDD 中的分区数或集群中的从属核心数，以较低者为准。

在您的情况下，它是 RDD 分区的数量。您可以通过如下方式重新分区您的 RDD 来增加它：

data = sc.textFile(bucket).repartition(sc.defaultParallelism * 3)

为什么是sc.defaultParallelism * 3？

Spark Tuning 指南建议使用2-3 tasks per core，sc.defaultParalellism 为您提供集群中的核心数量。

【讨论】：