我在yarn集群中运行我的Spark应用程序。在我的代码中,我使用队列可用核心的数量来为我的数据集创建分区:
Dataset ds = ...
ds.coalesce(config.getNumberOfCores());
我的问题是:如何通过编程方式获取队列可用核心数量,而不是通过配置方式?我在yarn集群中运行我的Spark应用程序。在我的代码中,我使用队列可用核心的数量来为我的数据集创建分区:
Dataset ds = ...
ds.coalesce(config.getNumberOfCores());
我的问题是:如何通过编程方式获取队列可用核心数量,而不是通过配置方式?从Spark中可以获取执行器数量和集群中核心数的方法。以下是我过去使用过的一些Scala实用程序代码。您应该很容易将其适应于Java。有两个关键想法:
工作进程数是执行器数减一或sc.getExecutorStorageStatus.length - 1
。
每个工作进程的核心数可以通过在工作进程上执行java.lang.Runtime.getRuntime.availableProcessors
来获得。
其余的代码是为了使用Scala隐式添加便捷方法到SparkContext
,这是模板代码。我几年前为1.x编写了这段代码,这就是为什么它没有使用SparkSession
的原因。
最后一个要点:通常最好将分区数合并为您的核心数的倍数,因为这可以改善数据倾斜的情况下的性能。在实践中,我使用1.5倍到4倍之间的任何值,具体取决于数据大小和作业是否在共享集群上运行。
import org.apache.spark.SparkContext
import scala.language.implicitConversions
class RichSparkContext(val sc: SparkContext) {
def executorCount: Int =
sc.getExecutorStorageStatus.length - 1 // one is the driver
def coresPerExecutor: Int =
RichSparkContext.coresPerExecutor(sc)
def coreCount: Int =
executorCount * coresPerExecutor
def coreCount(coresPerExecutor: Int): Int =
executorCount * coresPerExecutor
}
object RichSparkContext {
trait Enrichment {
implicit def enrichMetadata(sc: SparkContext): RichSparkContext =
new RichSparkContext(sc)
}
object implicits extends Enrichment
private var _coresPerExecutor: Int = 0
def coresPerExecutor(sc: SparkContext): Int =
synchronized {
if (_coresPerExecutor == 0)
sc.range(0, 1).map(_ => java.lang.Runtime.getRuntime.availableProcessors).collect.head
else _coresPerExecutor
}
}
更新
最近,getExecutorStorageStatus
已被删除。我们改用SparkEnv
的blockManager.master.getStorageStatus.length - 1
(减去1是为了再次排除driver)。通常通过SparkContext
的env
来访问它,但在org.apache.spark
包之外无法访问。因此,我们使用了一种封装违规模式:
package org.apache.spark
object EncapsulationViolator {
def sparkEnv(sc: SparkContext): SparkEnv = sc.env
}
在寻找答案时发现了这篇文章,问题和我遇到的差不多。
我发现:
Dataset ds = ...
ds.coalesce(sc.defaultParallelism());
这正是OP所寻求的功能。
例如,我的5个节点 x 8核心集群会将defaultParallelism
设置为40。
java.lang.Runtime.getRuntime.availableProcessors * (sc.statusTracker.getExecutorInfos.length -1)
import spark.implicits._
import scala.collection.JavaConverters._
import sys.process._
val procs = (1 to 1000).toDF.map(_ => "hostname".!!.trim -> java.lang.Runtime.getRuntime.availableProcessors).collectAsList().asScala.toMap
val nCpus = procs.values.sum
scala> :paste
// Entering paste mode (ctrl-D to finish)
import spark.implicits._
import scala.collection.JavaConverters._
import sys.process._
val procs = (1 to 1000).toDF.map(_ => "hostname".!!.trim -> java.lang.Runtime.getRuntime.availableProcessors).collectAsList().asScala.toMap
val nCpus = procs.values.sum
// Exiting paste mode, now interpreting.
import spark.implicits._
import scala.collection.JavaConverters._
import sys.process._
procs: scala.collection.immutable.Map[String,Int] = Map(ip-172-31-76-201.ec2.internal -> 2, ip-172-31-74-242.ec2.internal -> 2)
nCpus: Int = 4
如果您的集群通常有大量机器,请将零添加到您的范围中。即使在我的两台机器集群上,10000个任务也可以在几秒钟内完成。
如果您需要比sc.defaultParallelism()提供的更多信息(例如@SteveC的答案),则可能只有这才有用。
对于那些没有使用yarn集群的人:如果你在Python/Databricks中进行操作,这里有一个我编写的函数可以帮助解决这个问题。它将为您获取工作节点数量和CPU数量,并返回您的工作分配的最终CPU计数的乘积。
def GetDistCPUCount():
nWorkers = int(spark.sparkContext.getConf().get('spark.databricks.clusterUsageTags.clusterTargetWorkers'))
GetType = spark.sparkContext.getConf().get('spark.databricks.clusterUsageTags.clusterNodeType')
GetSubString = pd.Series(GetType).str.split(pat = '_', expand = True)
GetNumber = GetSubString[1].str.extract('(\d+)')
ParseOutString = GetNumber.iloc[0,0]
WorkerCPUs = int(ParseOutString)
nCPUs = nWorkers * WorkerCPUs
return nCPUs