Spark YARN 集群仍未得到充分利用

Posted 2023-04-18

【中文标题】Spark YARN 集群仍未得到充分利用

【英文标题】：Spark YARN cluster remains underutilized

【发布时间】：2018-05-25 04:49:07

【问题描述】：

使用 Microsoft Azure，我有一个包含 2 个主节点和 50 个工作节点的集群。所有节点都有 8 个内核和 64 GB 的 RAM。

我正在使用 pyspark 运行 spark-submit 作业。我的 Python 代码中最相关的部分是我创建一个长度为 72 的元组的 Python 列表 arg_list。（每个元组有大约 6 个元素，没有两个元组是相同的。）然后我创建一个具有 72 个分区的 RDD，如下所示,

sc = spark.sparkContext
rdd = sc.parallelize(arg_list, len(arg_list))

我的目标是同时处理所有 72 个分区。处理分区涉及以下内容：

根据收到的元组元素的值从 Blob (HDFS) 存储中读取两个文件。 运行名为 gensim 的 Python 包，以使用读取的两个文件的内容执行 NLP 任务。

为此，我按如下方式提交我的 spark-submit 作业，

spark-submit --master yarn --num-executors 100 --executor-cores 3 --executor-memory 19G /path/to/my/file.py

思路如下。将每个工作节点用于 2 个执行程序。每个执行器有 3 个核心，因此每个节点有 8-2*3=2 个核心用于节点管理器和任何其他开销任务。因为我每个节点可以获得 2 个执行器，并且我有 50 个节点，所以我有 100 个执行器。 （我承认在这种情况下集群比需要的要大一些。）

当我运行这段代码时，我可以使用 Ambari 监控工作节点的活动。我曾预计 72/2 = 36 个工作节点处于忙碌状态（如平均负载所证明的那样），而 50-36 = 14 个节点处于空闲状态。相反，我看到只有 12 个节点很忙，并且每个节点似乎都在运行 6 个进程。

任务数 6*12=72 可能不是巧合。就好像 Spark/YARN 决定忽略我的参数，并将我的分区处理塞进尽可能少的节点中。

我还注意到，完成 72 项任务中的任何一项似乎都需要很长时间。我这样说是基于看到一个典型的任务在串行模式下运行需要 3 个小时，而我的 Spark 作业运行了 6 个小时却完全没有输出。

问题：

为什么我只使用 50 个工作节点中的 12 个？ 为什么我的代码运行这么慢？

我已阅读有关 spark-submit/Yarn 参数的指南，并认为我所写的内容是有道理的。我缺少一些其他参数设置吗？

【参考方案1】：

Spark 将根据您正在运行的作业可用的内核总数来处理每个分区。

假设您的 Spark 作业有 100 个执行程序，每个执行程序有 3 个核心。这意味着您将能够同时处理 100 x 3 = 300 个分区，假设 spark.task.cpus 设置为 1。

spark.task.cpus 是为每个任务分配的核心数，--executor-cores 指定每个执行程序的核心数。

具有 2 个执行器的工作节点，进程 2 x 3 = 6 个分区。并且默认 spark.default.parallelism = 12。所以 6x12 = 72。

Spark 中用于在运行时调整分区数的两个配置属性如下：

将默认并行度提高

--conf spark.default.parallelism=36 --conf spark.default.parallelism=36

设置 spark.task.cpus=2 和 --executor-cores 4（在 spark 提交命令中）。所以每个节点将只处理 (4/2 =) 2 个分区。在这种情况下，36 个节点将用于并行处理数据。

【讨论】：

非常感谢您的回答。需要明确的是，没有必要更改任何底层配置文件。我可以通过 --conf spark.default.parallelism=36 spark.task.cpus=2 在 spark-submit 命令中设置这两个变量（spark.task.cpus 和 spark.default.parallelism）。对吗？

为了更清楚，您说“通过 --conf spark.default.parallelism=36 spark.default.parallelism=36 增加默认并行度”。我假设你的意思是 --conf spark.default.parallelism=36 spark.task.cpus=2?

是的。如果您通过 --conf spark.task.cpus=2 --executor-cores=4 那么您的每个节点将处理 2 个分区。有 72 个分区，因此将有 36 个节点。还提供这些参数以更改默认并行度 --conf spark.default.parallelism=36 --conf spark.default.parallelism=36