apachespark:内核数量与执行器数量的对比

zy1mlcev  于 2021-06-03  发布在  Hadoop
关注(0)|答案(8)|浏览(423)

我试图了解在Yarn上运行Spark作业时,芯数和执行器数之间的关系。
测试环境如下:
数据节点数:3
数据节点计算机规范:
cpu:core i7-4790(内核数:4,线程数:8)
内存:32gb(8gb x 4)
硬盘:8tb(2tb x 4)
网络:1gb
spark版本:1.0.0
hadoop版本:2.4.0(hortonworks hdp 2.1)
spark作业流:sc.textfile->filter->map->filter->maptopair->reducebykey->map->saveastextfile
输入数据
类型:单个文本文件
大小:165gb
线路数:4568833
输出
第二次过滤后的行数:310640717
结果文件行数:99848268
结果文件大小:41gb
作业使用以下配置运行: --master yarn-client --executor-memory 19G --executor-cores 7 --num-executors 3 (每个数据节点的执行器,使用尽可能多的核心) --master yarn-client --executor-memory 19G --executor-cores 4 --num-executors 3 (减少芯数) --master yarn-client --executor-memory 4G --executor-cores 2 --num-executors 12 (核心越少,执行者越多)
运行时间:
50分15秒
55分48秒
31分23秒
令我惊讶的是,(3)快得多。
我认为(1)会更快,因为洗牌时执行者之间的通信会更少。
尽管(1)的#芯数少于(3),但#芯数并不是关键因素,因为(2)的表现不错。
(在pwilmot的回答之后添加了以下内容。)
有关信息,性能监视器屏幕截图如下所示:
(1)的ganglia数据节点摘要-作业在04:37开始。

(3)的ganglia数据节点摘要-作业在19:47开始。在此之前请忽略图表。

图表大致分为两部分:
第一:从开始到减少关键:cpu密集型,无网络活动
第二:在reducebykey:cpu降低后,网络i/o完成。
如图所示,(1)可以使用给定的cpu功率。所以,这可能不是线程数的问题。
如何解释这个结果?

wmtdaxz3

wmtdaxz31#

我认为前两种配置有一个小问题。线程和内核的概念如下。线程的概念是,如果核心是理想的,那么就使用该核心来处理数据。所以在前两种情况下,内存没有得到充分利用。如果您想在本例中做基准点,请选择每台机器上有10个以上芯的机器。然后做基准点。
但每个执行器的内核数不要超过5个,否则会影响i/o性能。
所以最好的机器做这个基准点可能是数据节点有10个核心。
数据节点机器规格:cpu:core i7-4790(核心数:10,线程数:20)ram:32gb(8gb x 4)hdd:8tb(2tb x 4)

j2cgzkjk

j2cgzkjk2#

简而言之:我认为tgbaggio是对的。您的执行器达到了hdfs吞吐量限制。
我认为这里的答案可能比这里的一些建议要简单一些。
我的线索在集群网络图中。对于运行1,利用率稳定在~50 m字节/秒。对于运行3,稳定的利用率提高了一倍,约为100 m字节/秒。
从dzord分享的cloudera博客文章中,您可以看到以下重要引语:
我注意到hdfs客户机有大量并发线程的问题。粗略估计,每个执行器最多有五个任务可以实现完全的写吞吐量,因此最好将每个执行器的内核数保持在这个数字以下。
所以,让我们做一些计算,看看如果这是真的,我们期望的性能如何。

运行1:19 gb,7核,3个执行器

3个执行器x 7个线程=21个线程
由于每个执行器有7个核,我们希望hdfs的io有限(最多5个核)
有效吞吐量~=3个执行器x 5个线程=15个线程

运行3:4 gb,2核,12个执行器

2个执行器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 ,看起来我们的网络有限。
同样的效果解释了运行1和运行2之间的差异。

运行2:19 gb,4核,3个执行器

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上工作得到的结果)之间的区别:并发性不是并行性。
简单的解释是,如果spark作业与文件系统或网络交互,cpu会花费大量时间等待与这些接口的通信,而不是花费大量时间实际“工作”。通过一次给这些cpu分配多个任务,它们将花费更少的等待时间和更多的工作时间,您将看到更好的性能。

ou6hu8tu

ou6hu8tu3#

spark动态分配提供了灵活性并动态分配资源。在这种情况下,可以给出最小和最大执行器的数量。此外,还可以给出在申请开始时必须启动的执行者的数量。
请阅读以下内容:
http://spark.apache.org/docs/latest/configuration.html#dynamic-分配

7eumitmz

7eumitmz4#

sandy ryza说,当你在hdfs上运行spark应用程序时
我注意到hdfs客户机有大量并发线程的问题。粗略估计,每个执行器最多有五个任务可以实现完全的写吞吐量,因此最好将每个执行器的内核数保持在这个数字以下。
所以我认为第一个配置比第三个慢是因为hdfs的i/o吞吐量不好

nvbavucw

nvbavucw5#

从rstudio的SparkyR软件包页面提供的优秀资源中:
Spark定义:
为Spark术语提供一些简单的定义可能很有用:
节点:服务器
工作节点:作为集群一部分的服务器,可以运行spark作业
主节点:协调工作节点的服务器。
执行器:节点内的一种虚拟机。一个节点可以有多个执行器。
驱动程序节点:启动spark会话的节点。通常,这将是SparkyR所在的服务器。
驱动程序(executor):驱动程序节点也将显示在executor列表中。

uelo1irk

uelo1irk6#

为了让这一切更具体一点,下面是一个配置spark应用程序以尽可能多地使用集群的工作示例:想象一个集群有6个节点运行NodeManager,每个节点都配备了16核和64gb内存。nodemanager的容量yarn.nodemanager.resource.memory-mb和yarn.nodemanager.resource.cpu-vcores可能应该分别设置为63*1024=64512(兆字节)和15。我们避免将100%的资源分配给yarn容器,因为节点需要一些资源来运行os和hadoop守护进程。在本例中,我们为这些系统进程保留一个千兆字节和一个内核。cloudera manager通过自动计算和配置这些属性来提供帮助。
可能的第一个冲动是使用--num executors 6--executor cores 15--executor内存63g。但是,这是错误的方法,因为:
63gb+的执行器内存开销不适合节点管理器的63gb容量。应用程序主节点将占用其中一个节点上的一个核心,这意味着该节点上没有容纳15个核心执行器的空间。每个执行器15个内核可能会导致hdfs i/o吞吐量下降。
更好的选择是使用--num executors 17--executor cores 5--executor内存19g。为什么?
此配置会在所有节点上产生三个执行器,但具有am的节点除外,该节点将有两个执行器--执行器内存的计算公式为(每个节点63/3个执行器)=21。21 * 0.07 = 1.47. 21 – 1.47 ~ 19.
cloudera的博客how-to:tune your apache spark jobs(第2部分)中的一篇文章对此进行了解释。

px9o7tmv

px9o7tmv7#

我认为主要原因之一是地理位置。您的输入文件大小是165g,文件的相关块肯定分布在多个数据节点上,更多的执行者可以避免网络复制。
尝试设置executor num equal blocks count,我认为可以更快。

zsohkypk

zsohkypk8#

我自己没有使用过这些设置,所以这只是猜测,但是如果我们把这个问题看作是分布式系统中的普通内核和线程,那么在您的集群中,您最多可以使用12个内核(43台机器)和24个线程(83台机器)。在前两个示例中,您为作业分配了相当数量的核心(潜在计算空间),但要在这些核心上运行的线程(作业)数量非常有限,因此您无法使用分配的大部分处理能力,因此即使分配了更多计算资源,作业也会变慢。
您提到您的关注点是在shuffle步骤中—虽然在shuffle步骤中限制开销是很好的,但通常更重要的是利用集群的并行化。想一想极端的情况-一个单线程程序的零洗牌。

相关问题