blocks|key|4530193|text|sc.defaultParallelism只是一个提示。根据配置的不同，它可能与节点数量没有关系。如果您使用带有分区计数参数的操作，但没有提供该参数，则这是分区的数量。例如，sc.parallelize将从列表中创建一个新的RDD。您可以用第二个参数告诉它要在RDD中创建多少个分区。但此参数的默认值为sc.defaultParallelism。|type|unstyled|depth|inlineStyleRanges|offset|length|style|CODE|entityRanges|data|4530194|您可以在Scala+API中使用sc.getExecutorMemoryStatus获取执行器的数量，但是在Python+API中没有公开。|4530195|一般而言，建议RDD中的分区数量是执行器数量的4倍左右。这是一个很好的技巧，因为如果任务花费的时间有差异，这将使它变得平衡。例如，一些执行器将处理5个较快的任务，而另一些执行器将处理3个较慢的任务。|4530196|你不需要对此非常准确。如果你有一个大致的想法，你可以用一个估计值。比如，如果你知道你的CPU少于200个，你可以说500个分区就可以了。|4530197|因此，尝试使用以下数量的分区创建RDD：|4530198|rdd+=+sc.parallelize(data,+500)+++++#+If+distributing+local+data.
rdd+=+sc.textFile('file.csv',+500)++#+If+loading+data+from+a+file.|code-block|syntax|javascript|4530199|或者在计算之前重新划分RDD，如果您不控制RDD的创建：|4530200|rdd+=+rdd.repartition(500)|4530201|您可以使用rdd.getNumPartitions()查看RDD中的分区数量。|4530202|entityMap^0|0|L|2F|E|47|L|0|G|Q|0|0|0|0|0|0|0|5|M|0^^$0|@$1|2|3|4|5|6|7|10|8|@$9|11|A|12|B|C]|$9|13|A|14|B|C]|$9|15|A|16|B|C]]|D|@]|E|$]]|$1|F|3|G|5|6|7|17|8|@$9|18|A|19|B|C]]|D|@]|E|$]]|$1|H|3|I|5|6|7|1A|8|@]|D|@]|E|$]]|$1|J|3|K|5|6|7|1B|8|@]|D|@]|E|$]]|$1|L|3|M|5|6|7|1C|8|@]|D|@]|E|$]]|$1|N|3|O|5|P|7|1D|8|@]|D|@]|E|$Q|R]]|$1|S|3|T|5|6|7|1E|8|@]|D|@]|E|$]]|$1|U|3|V|5|P|7|1F|8|@]|D|@]|E|$Q|R]]|$1|W|3|X|5|6|7|1G|8|@$9|1H|A|1I|B|C]]|D|@]|E|$]]|$1|Y|3|-4|5|6|7|1J|8|@]|D|@]|E|$]]]|Z|$]]

<code>sc.defaultParallelism</code> is just a hint. Depending on the configuration it may not have a relation to the number of nodes. This is the number of partitions if you use an operation that takes a partition count argument but you don't provide it. For example <code>sc.parallelize</code> will make a new RDD from a list. You can tell it how many partitions to create in the RDD with the second argument. But the default value for this argument is <code>sc.defaultParallelism</code>.

You can get the number of executors with <code>sc.getExecutorMemoryStatus</code> in the Scala API, but this is not exposed in the Python API.

In general the recommendation is to have around 4 times as many partitions in an RDD as you have executors. This is a good tip, because if there is variance in how much time the tasks take this will even it out. Some executors will process 5 faster tasks while others process 3 slower tasks, for example.

You don't need to be very accurate with this. If you have a rough idea, you can go with an estimate. Like if you know you have less than 200 CPUs, you can say 500 partitions will be fine.

So try to create RDDs with this number of partitions:

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

Or repartition the RDD before the computation if you don't control the creation of the RDD:

<pre><code>rdd = rdd.repartition(500)
</code></pre>

You can check the number of partitions in an RDD with <code>rdd.getNumPartitions()</code>.

blocks|key|4530231|text|在pyspark上，你仍然可以使用pyspark的py4j桥来调用scala：|type|unstyled|depth|inlineStyleRanges|entityRanges|data|4530232|sc._jsc.sc().getExecutorMemoryStatus().size()|code-block|syntax|javascript|4530233|entityMap^0|0|0^^$0|@$1|2|3|4|5|6|7|I|8|@]|9|@]|A|$]]|$1|B|3|C|5|D|7|J|8|@]|9|@]|A|$E|F]]|$1|G|3|-4|5|6|7|K|8|@]|9|@]|A|$]]]|H|$]]

On pyspark you could still call the scala <code>getExecutorMemoryStatus</code> API using pyspark's py4j bridge:

<pre><code>sc._jsc.sc().getExecutorMemoryStatus().size()
</code></pre>

blocks|key|22684|text|我发现有时我的会话会被遥控器终止，给出一个奇怪的Java错误|type|unstyled|depth|inlineStyleRanges|entityRanges|data|22685|Py4JJavaError:+An+error+occurred+while+calling+o349.defaultMinPartitions.
:+java.lang.IllegalStateException:+Cannot+call+methods+on+a+stopped+SparkContext.|code-block|syntax|javascript|22686|我通过以下方式避免了这一点|22687|def+check_alive(spark_conn):
++++"""Check+if+connection+is+alive.+``True``+if+alive,+``False``+if+not"""
++++try:
++++++++get_java_obj+=+spark_conn._jsc.sc().getExecutorMemoryStatus()
++++++++return+True
++++except+Exception:
++++++++return+False

def+get_number_of_executors(spark_conn):
++++if+not+check_alive(spark_conn):
++++++++raise+Exception('Unexpected+Error:+Spark+Session+has+been+killed')
++++try:
++++++++return+spark_conn._jsc.sc().getExecutorMemoryStatus().size()
++++except:
++++++++raise+Exception('Unknown+error')|22688|entityMap^0|0|0|0|0^^$0|@$1|2|3|4|5|6|7|M|8|@]|9|@]|A|$]]|$1|B|3|C|5|D|7|N|8|@]|9|@]|A|$E|F]]|$1|G|3|H|5|6|7|O|8|@]|9|@]|A|$]]|$1|I|3|J|5|D|7|P|8|@]|9|@]|A|$E|F]]|$1|K|3|-4|5|6|7|Q|8|@]|9|@]|A|$]]]|L|$]]

I found sometimes my sessions were killed by the remote giving a strange Java error

<pre><code>Py4JJavaError: An error occurred while calling o349.defaultMinPartitions.
: java.lang.IllegalStateException: Cannot call methods on a stopped SparkContext.
</code></pre>

I avoided this by the following

<pre><code>def check_alive(spark_conn):
 """Check if connection is alive. ``True`` if alive, ``False`` if not"""
 try:
 get_java_obj = spark_conn._jsc.sc().getExecutorMemoryStatus()
 return True
 except Exception:
 return False

def get_number_of_executors(spark_conn):
 if not check_alive(spark_conn):
 raise Exception('Unexpected Error: Spark Session has been killed')
 try:
 return spark_conn._jsc.sc().getExecutorMemoryStatus().size()
 except:
 raise Exception('Unknown error')
</code></pre>

blocks|key|22711|text|其他答案提供了一种获取执行者数量的方法。下面是一种获取节点数量的方法。这包括头节点和工作节点。|type|unstyled|depth|inlineStyleRanges|entityRanges|data|22712|s+=+sc._jsc.sc().getExecutorMemoryStatus().keys()
l+=+str(s).replace("Set(","").replace(")","").split(",+")

d+=+set()
for+i+in+l:
++++d.add(i.split(":")[0])
len(d)++|code-block|syntax|javascript|22713|entityMap^0|0|0^^$0|@$1|2|3|4|5|6|7|I|8|@]|9|@]|A|$]]|$1|B|3|C|5|D|7|J|8|@]|9|@]|A|$E|F]]|$1|G|3|-4|5|6|7|K|8|@]|9|@]|A|$]]]|H|$]]

The other answers provide a way to get the number of executors. Here is a way to get the number of nodes. This includes head and worker nodes. 

<pre><code>s = sc._jsc.sc().getExecutorMemoryStatus().keys()
l = str(s).replace("Set(","").replace(")","").split(", ")

d = set()
for i in l:
 d.add(i.split(":")[0])
len(d) 
</code></pre>

blocks|key|22750|text|应该可以使用此方法获取集群中的节点数量(类似于上面的@Dan的方法，但更短，效果更好！)。|type|unstyled|depth|inlineStyleRanges|entityRanges|data|22751|sc._jsc.sc().getExecutorMemoryStatus().keySet().size()|code-block|syntax|javascript|22752|entityMap^0|0|0^^$0|@$1|2|3|4|5|6|7|I|8|@]|9|@]|A|$]]|$1|B|3|C|5|D|7|J|8|@]|9|@]|A|$E|F]]|$1|G|3|-4|5|6|7|K|8|@]|9|@]|A|$]]]|H|$]]

It should be possible to get the number of nodes in the cluster using this (similar to @Dan's method above, but shorter and works better!).

<pre><code>sc._jsc.sc().getExecutorMemoryStatus().keySet().size()
</code></pre>

I'm running some operations in PySpark, and recently increased the number of nodes in my configuration (which is on Amazon EMR). However, even though I tripled the number of nodes (from 4 to 12), performance seems not to have changed. As such, I'd like to see if the new nodes are visible to Spark.
I'm calling the following function:
<pre><code>sc.defaultParallelism
&gt;&gt;&gt;&gt; 2
</code></pre>
But I think this is telling me the total number of tasks distributed to each node, not the total number of nodes that Spark can see.
How do I go about seeing the amount of nodes that PySpark is using in my cluster?

getting number of visible nodes in PySpark

Python

Spark 

Java

我在PySpark中运行了一些操作，最近在我的配置(在Amazon EMR上)中增加了节点数量。但是，即使我将节点数量增加了两倍(从4个增加到12个)，性能似乎没有变化。因此，我想看看新节点对Spark是否可见。我正在调用以下函数：sc.defaultParallelism>>>> 2但我认为这是在告诉我分配给每个节点...

问获取PySpark中可见节点数
EN

社区

活动

资源

关于

腾讯云开发者

热门产品

热门推荐

更多推荐

问获取PySpark中可见节点数EN

社区

活动

资源

关于

腾讯云开发者

热门产品

热门推荐

更多推荐

问获取PySpark中可见节点数
EN