Apache Spark：核心数与执行程序数

为了希望使这一切更加具体，这里有一个配置 Spark 应用程序以使用尽可能多的集群的工作示例：想象一个集群有六个运行 NodeManagers 的节点，每个节点配备 16 个内核和 64GB 内存。 NodeManager 容量，yarn.nodemanager.resource.memory-mb 和 yarn.nodemanager.resource.cpu-vcores，应该分别设置为 63 * 1024 = 64512（兆字节）和 15。我们避免将 100% 的资源分配给 YARN 容器，因为节点需要一些资源来运行操作系统和 Hadoop 守护进程。在这种情况下，我们为这些系统进程留下一个千兆字节和一个内核。 Cloudera Manager 通过考虑这些并自动配置这些 YARN 属性来提供帮助。可能的第一个冲动是使用 --num-executors 6 --executor-cores 15 --executor-memory 63G。但是，这是错误的方法，因为：63GB + 执行程序内存开销不适合 NodeManagers 的 63GB 容量。应用程序主控将在其中一个节点上占用一个核心，这意味着该节点上将没有空间容纳 15 核执行器。每个执行器 15 个内核会导致 HDFS I/O 吞吐量不佳。更好的选择是使用 --num-executors 17 --executor-cores 5 --executor-memory 19G。为什么？此配置会在所有节点上产生三个执行程序，但带有 AM 的节点除外，它将有两个执行程序。 --executor-memory 派生为（每个节点 63/3 个执行器）= 21. 21 * 0.07 = 1.47。 21 – 1.47 ~ 19。

Cloudera 的博客 How-to: Tune Your Apache Spark Jobs (Part 2) 中的一篇文章中给出了解释。

---

简短回答：我认为 tgbaggio 是对的。您达到了执行程序的 HDFS 吞吐量限制。

我认为这里的答案可能比这里的一些建议要简单一些。

我的线索在集群网络图中。对于运行 1，利用率稳定在 ~50 M 字节/秒。对于运行 3，稳定利用率翻了一番，约为 100 M 字节/秒。

在 DzOrd 分享的 the cloudera blog post 中，您可以看到以下重要引述：

我注意到 HDFS 客户端在处理大量并发线程时遇到了问题。一个粗略的猜测是，每个 executor 最多五个任务可以实现完整的写入吞吐量，因此最好将每个 executor 的核心数保持在该数字以下。

所以，让我们做一些计算，看看如果这是真的，我们期望什么性能。

运行 1：19 GB，7 个内核，3 个执行程序

个执行器 x 7 个线程 = 21 个线程

每个执行程序有 7 个内核，我们预计 HDFS 的 IO 有限（最多约 5 个内核）

有效吞吐量 ~= 3 执行器 x 5 线程 = 15 线程

运行 3：4 GB，2 个内核，12 个执行器

个执行器 x 12 个线程 = 24 个线程

每个执行器 2 个核心，因此 hdfs 吞吐量还可以

有效吞吐量 ~= 12 执行器 x 2 线程 = 24 线程

如果作业 100% 受并发（线程数）限制。我们希望运行时与线程数完全成反比。

ratio_num_threads = nthread_job1 / nthread_job3 = 15/24 = 0.625
inv_ratio_runtime = 1/(duration_job1 / duration_job3) = 1/(50/31) = 31/50 = 0.62

所以 ratio_num_threads ~= inv_ratio_runtime，看起来我们是网络受限的。

同样的效果解释了 Run 1 和 Run 2 之间的区别。

运行 2：19 GB，4 个内核，3 个执行程序

个执行器 x 4 个线程 = 12 个线程

每个执行器有 4 个内核，可以 IO 到 HDFS

有效吞吐量 ~= 3 执行器 x 4 线程 = 12 线程

比较有效线程数和运行时：

ratio_num_threads = nthread_job2 / nthread_job1 = 12/15 = 0.8
inv_ratio_runtime = 1/(duration_job2 / duration_job1) = 1/(55/50) = 50/55 = 0.91

它不像上次比较那样完美，但是当我们失去线程时，我们仍然看到类似的性能下降。

现在是最后一点：为什么我们会通过更多线程获得更好的性能，尤其是。线程数多于 CPU 数？

Rob Pike 的这篇精彩文章很好地解释了并行性（我们通过将数据划分到多个 CPU 上得到的）和并发性（当我们使用多个线程在单个 CPU 上工作时得到的）之间的区别：{ 1}。

简短的解释是，如果 Spark 作业与文件系统或网络交互，CPU 会花费大量时间等待与这些接口的通信，而不是花费大量时间实际“工作”。通过让这些 CPU 一次处理超过 1 个任务，它们等待的时间更少，工作的时间更多，你会看到更好的性能。

---

根据 Sandy Ryza，当您在 HDFS 上运行 Spark 应用程序时

我注意到 HDFS 客户端在处理大量并发线程时遇到了问题。一个粗略的猜测是，每个 executor 最多五个任务可以实现完整的写入吞吐量，因此最好将每个 executor 的核心数保持在该数字以下。

所以我相信你的第一个配置比第三个配置慢是因为 HDFS I/O 吞吐量不好

---

我自己没有玩过这些设置，所以这只是推测，但如果我们将此问题视为分布式系统中的普通内核和线程，那么在您的集群中，您最多可以使用 12 个内核（4 * 3 台机器）和 24 个线程（8 * 3 台机器）。在您的前两个示例中，您为您的工作提供了相当数量的内核（潜在的计算空间），但是在这些内核上运行的线程（工作）数量非常有限，以至于您无法使用分配的大部分处理能力因此即使分配了更多的计算资源，作业也会变慢。

您提到您关心的是洗牌步骤 - 虽然限制洗牌步骤中的开销很好，但利用集群的并行化通常更为重要。考虑一下极端情况——零随机播放的单线程程序。

---

从 RStudio's Sparklyr package page 提供的优秀资源：

SPARK 定义：为 Spark 命名法提供一些简单的定义可能会很有用： 节点：服务器 工作节点：作为集群一部分的服务器，可用于运行 Spark 作业 主节点：协调工作节点的服务器。 Executor：节点内的一种虚拟机。一个节点可以有多个执行者。驱动节点：启动 Spark 会话的节点。通常，这将是 sparklyr 所在的服务器。 Driver（Executor）：Driver Node也会出现在Executor列表中。

---

我认为主要原因之一是地方性。你的输入文件大小是165G，文件的相关块肯定分布在多个DataNode上，更多的executor可以避免网络复制。

尝试设置 executor num 等于块数，我认为可以更快。

---

Spark 动态分配提供了灵活性并动态分配资源。在这个数量的 min 和 max executors 中可以给出。此外，还可以给出在应用程序启动时必须启动的执行程序的数量。

请阅读以下内容：

http://spark.apache.org/docs/latest/configuration.html#dynamic-allocation

---

我认为前两种配置存在一个小问题。线程和核心的概念如下。线程的概念是，如果核心是理想的，则使用该核心来处理数据。所以在前两种情况下内存没有被充分利用。如果要对这个示例进行基准测试，请选择每台机器上具有 10 个以上内核的机器。然后做基准测试。

但是不要为每个执行程序提供超过 5 个内核，否则 i/o 性能会出现瓶颈。

因此，进行此基准测试的最佳机器可能是具有 10 个内核的数据节点。

数据节点机器规格： CPU：Core i7-4790（核心数：10，线程数：20） RAM：32GB (8GB x 4) HDD：8TB (2TB x 4)

---