I haven't played with these settings myself so this is just speculation but if we think about this issue as normal cores and threads in a distributed system then in your cluster you can use up to 12 cores (4 * 3 machines) and 24 threads (8 * 3 machines). In your first two examples you are giving your job a fair number of cores (potential computation space) but the number of threads (jobs) to run on those cores is so limited that you aren't able to use much of the processing power allocated and thus the job is slower even though there is more computation resources allocated.

您提到您关心的是shuffle步骤—虽然在shuffle步骤中限制开销很好，但通常更重要的是利用集群的并行化。考虑一个极端的情况——一个没有shuffle的单线程程序。

To hopefully make all of this a little more concrete, here’s a worked example of configuring a Spark app to use as much of the cluster as possible: Imagine a cluster with six nodes running NodeManagers, each equipped with 16 cores and 64GB of memory. The NodeManager capacities, yarn.nodemanager.resource.memory-mb and yarn.nodemanager.resource.cpu-vcores, should probably be set to 63 * 1024 = 64512 (megabytes) and 15 respectively. We avoid allocating 100% of the resources to YARN containers because the node needs some resources to run the OS and Hadoop daemons. In this case, we leave a gigabyte and a core for these system processes. Cloudera Manager helps by accounting for these and configuring these YARN properties automatically. The likely first impulse would be to use --num-executors 6 --executor-cores 15 --executor-memory 63G. However, this is the wrong approach because: 63GB + the executor memory overhead won’t fit within the 63GB capacity of the NodeManagers. The application master will take up a core on one of the nodes, meaning that there won’t be room for a 15-core executor on that node. 15 cores per executor can lead to bad HDFS I/O throughput. A better option would be to use --num-executors 17 --executor-cores 5 --executor-memory 19G. Why? This config results in three executors on all nodes except for the one with the AM, which will have two executors. --executor-memory was derived as (63/3 executors per node) = 21. 21 * 0.07 = 1.47. 21 – 1.47 ~ 19.

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

Spark动态分配提供了灵活性，动态分配资源。在此范围内，可以给出最小和最大执行程序的数量。同样，在应用程序启动时必须启动的执行程序的数量也可以给定。

请阅读以下相同的内容:

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

我认为其中一个主要原因是地方性。您的输入文件大小为165G，文件的相关块当然分布在多个datanode上，更多的执行器可以避免网络复制。

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

I haven't played with these settings myself so this is just speculation but if we think about this issue as normal cores and threads in a distributed system then in your cluster you can use up to 12 cores (4 * 3 machines) and 24 threads (8 * 3 machines). In your first two examples you are giving your job a fair number of cores (potential computation space) but the number of threads (jobs) to run on those cores is so limited that you aren't able to use much of the processing power allocated and thus the job is slower even though there is more computation resources allocated.

您提到您关心的是shuffle步骤—虽然在shuffle步骤中限制开销很好，但通常更重要的是利用集群的并行化。考虑一个极端的情况——一个没有shuffle的单线程程序。

来自RStudio的Sparklyr包页面的优秀资源:

SPARK DEFINITIONS: It may be useful to provide some simple definitions for the Spark nomenclature: Node: A server Worker Node: A server that is part of the cluster and are available to run Spark jobs Master Node: The server that coordinates the Worker nodes. Executor: A sort of virtual machine inside a node. One Node can have multiple Executors. Driver Node: The Node that initiates the Spark session. Typically, this will be the server where sparklyr is located. Driver (Executor): The Driver Node will also show up in the Executor list.