获取PySpark中可见节点的数量

Question

我在PySpark中运行一些操作,最近增加了配置中的节点数(在Amazon EMR上).然而,即使我将节点数量增加了两倍(从4到12),性能似乎也没有改变.因此,我想看看Spark是否可以看到新节点.

我正在调用以下函数:

sc.defaultParallelism
>>>> 2

但我认为这告诉我分配给每个节点的任务总数,而不是Spark可以看到的代码总数.

如何查看PySpark在群集中使用的节点数量？

Answer 1

在pyspark上你仍然可以getExecutorMemoryStatus使用pyspark的py4j桥来调用scala API:

sc._jsc.sc().getExecutorMemoryStatus().size()

Answer 2

sc.defaultParallelism只是一个暗示.根据配置,它可能与节点数量无关.如果您使用带有分区计数参数但未提供分区计数参数的操作,则这是分区数.例如,sc.parallelize将从列表中创建新的RDD.您可以使用第二个参数告诉它在RDD中创建多少个分区.但是这个参数的默认值是sc.defaultParallelism.

您可以sc.getExecutorMemoryStatus在Scala API中获取执行程序的数量,但这不会在Python API中公开.

一般来说,建议在RDD中的分区大约是执行程序的4倍.这是一个很好的提示,因为如果任务所花费的时间存在差异,这将使其均匀.例如,一些执行器将处理5个更快的任务,而其他执行器处理3个较慢的任务.

你不需要对此非常准确.如果你有一个粗略的想法,你可以去估计.就像你知道你的CPU少于200个,你可以说500个分区就可以了.

因此,尝试使用此数量的分区创建RDD:

rdd = sc.parallelize(data, 500)     # If distributing local data.
rdd = sc.textFile('file.csv', 500)  # If loading data from a file.

或者如果不控制RDD的创建,则在计算之前重新分配RDD:

rdd = rdd.repartition(500)

您可以使用检查RDD中的分区数rdd.getNumPartitions().

Answer 3

应该可以使用此方法获取集群中的节点数（类似于上面@Dan 的方法，但更短且效果更好！）。

sc._jsc.sc().getExecutorMemoryStatus().keySet().size()