You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by me...@apache.org on 2019/05/23 18:46:33 UTC
[spark] branch master updated: [SPARK-27488][CORE] Driver interface
to support GPU resources
This is an automated email from the ASF dual-hosted git repository.
meng pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new 74e5e41 [SPARK-27488][CORE] Driver interface to support GPU resources
74e5e41 is described below
commit 74e5e41eebf9ed596b48e6db52a2a9c642e5cbc3
Author: Thomas Graves <tg...@nvidia.com>
AuthorDate: Thu May 23 11:46:13 2019 -0700
[SPARK-27488][CORE] Driver interface to support GPU resources
## What changes were proposed in this pull request?
Added the driver functionality to get the resources.
The user interface is: SparkContext.resources - I called it this to match the TaskContext.resources api proposed in the other PR. Originally it was going to be called SparkContext.getResources but changed to be consistent, if people have strong feelings I can change it.
There are 2 ways the driver can discover what resources it has.
1) user specifies a discoveryScript, this is similar to the executors and is meant for yarn and k8s where they don't tell you what you were allocated but you are running in isolated environment.
2) read the config spark.driver.resource.resourceName.addresses. The config is meant to be used with standalone mode where the Worker will have to assign what GPU addresses the Driver is allowed to use by setting that config.
When the user runs a spark application, if they want the driver to have GPU's they would specify the conf spark.driver.resource.gpu.count=X where x is the number they want. If they are running on yarn or k8s they will also have to specify the discoveryScript as specified above, if they are on standalone mode and cluster is setup properly they wouldn't have to specify anything else. We could potentially get rid of the spark.driver.resources.gpu.addresses config which is really meant [...]
- This PR also has changes to be consistent about using resourceName everywhere.
- change the config names from POSTFIX to SUFFIX to be more consistent with other areas in Spark
- Moved the config checks around a bit since now used by both executor and driver. Note those might overlap a bit with https://github.com/apache/spark/pull/24374 so we will have to figure out which one should go in first.
## How was this patch tested?
Unit tests and manually test the interface.
Closes #24615 from tgravescs/SPARK-27488.
Authored-by: Thomas Graves <tg...@nvidia.com>
Signed-off-by: Xiangrui Meng <me...@databricks.com>
---
.../org/apache/spark/ResourceDiscoverer.scala | 88 ++++++++++++++-----
.../main/scala/org/apache/spark/SparkConf.scala | 60 +++++++++++++
.../main/scala/org/apache/spark/SparkContext.scala | 47 +++++++++++
.../executor/CoarseGrainedExecutorBackend.scala | 64 +++-----------
.../org/apache/spark/internal/config/package.scala | 5 +-
.../org/apache/spark/ResourceDiscovererSuite.scala | 83 ++++++++++++++----
.../scala/org/apache/spark/SparkConfSuite.scala | 43 ++++++++++
.../scala/org/apache/spark/SparkContextSuite.scala | 98 ++++++++++++++++++++++
.../CoarseGrainedExecutorBackendSuite.scala | 61 ++++----------
docs/configuration.md | 29 +++++--
10 files changed, 436 insertions(+), 142 deletions(-)
diff --git a/core/src/main/scala/org/apache/spark/ResourceDiscoverer.scala b/core/src/main/scala/org/apache/spark/ResourceDiscoverer.scala
index 1963942..d3b3860 100644
--- a/core/src/main/scala/org/apache/spark/ResourceDiscoverer.scala
+++ b/core/src/main/scala/org/apache/spark/ResourceDiscoverer.scala
@@ -29,10 +29,10 @@ import org.apache.spark.internal.config._
import org.apache.spark.util.Utils.executeAndGetOutput
/**
- * Discovers resources (GPUs/FPGAs/etc). It currently only supports resources that have
- * addresses.
+ * Discovers information about resources (GPUs/FPGAs/etc). It currently only supports
+ * resources that have addresses.
* This class finds resources by running and parsing the output of the user specified script
- * from the config spark.{driver/executor}.resource.{resourceType}.discoveryScript.
+ * from the config spark.{driver/executor}.resource.{resourceName}.discoveryScript.
* The output of the script it runs is expected to be JSON in the format of the
* ResourceInformation class.
*
@@ -42,28 +42,41 @@ private[spark] object ResourceDiscoverer extends Logging {
private implicit val formats = DefaultFormats
- def findResources(sparkConf: SparkConf, isDriver: Boolean): Map[String, ResourceInformation] = {
- val prefix = if (isDriver) {
- SPARK_DRIVER_RESOURCE_PREFIX
- } else {
- SPARK_EXECUTOR_RESOURCE_PREFIX
- }
- // get unique resource types by grabbing first part config with multiple periods,
- // ie resourceType.count, grab resourceType part
- val resourceNames = sparkConf.getAllWithPrefix(prefix).map { case (k, _) =>
- k.split('.').head
- }.toSet
+ /**
+ * This function will discover information about a set of resources by using the
+ * user specified script (spark.{driver/executor}.resource.{resourceName}.discoveryScript).
+ * It optionally takes a set of resource names or if that isn't specified
+ * it uses the config prefix passed in to look at the executor or driver configs
+ * to get the resource names. Then for each resource it will run the discovery script
+ * and get the ResourceInformation about it.
+ *
+ * @param sparkConf SparkConf
+ * @param confPrefix Driver or Executor resource prefix
+ * @param resourceNamesOpt Optionally specify resource names. If not set uses the resource
+ * configs based on confPrefix passed in to get the resource names.
+ * @return Map of resource name to ResourceInformation
+ */
+ def discoverResourcesInformation(
+ sparkConf: SparkConf,
+ confPrefix: String,
+ resourceNamesOpt: Option[Set[String]] = None
+ ): Map[String, ResourceInformation] = {
+ val resourceNames = resourceNamesOpt.getOrElse(
+ // get unique resource names by grabbing first part config with multiple periods,
+ // ie resourceName.count, grab resourceName part
+ SparkConf.getBaseOfConfigs(sparkConf.getAllWithPrefix(confPrefix))
+ )
resourceNames.map { rName => {
- val rInfo = getResourceInfoForType(sparkConf, prefix, rName)
+ val rInfo = getResourceInfo(sparkConf, confPrefix, rName)
(rName -> rInfo)
}}.toMap
}
- private def getResourceInfoForType(
+ private def getResourceInfo(
sparkConf: SparkConf,
- prefix: String,
- resourceType: String): ResourceInformation = {
- val discoveryConf = prefix + resourceType + SPARK_RESOURCE_DISCOVERY_SCRIPT_POSTFIX
+ confPrefix: String,
+ resourceName: String): ResourceInformation = {
+ val discoveryConf = confPrefix + resourceName + SPARK_RESOURCE_DISCOVERY_SCRIPT_SUFFIX
val script = sparkConf.getOption(discoveryConf)
val result = if (script.nonEmpty) {
val scriptFile = new File(script.get)
@@ -73,21 +86,50 @@ private[spark] object ResourceDiscoverer extends Logging {
val output = executeAndGetOutput(Seq(script.get), new File("."))
val parsedJson = parse(output)
val name = (parsedJson \ "name").extract[String]
- val addresses = (parsedJson \ "addresses").extract[Array[String]].toArray
+ val addresses = (parsedJson \ "addresses").extract[Array[String]]
+ if (name != resourceName) {
+ throw new SparkException(s"Discovery script: ${script.get} specified via " +
+ s"$discoveryConf returned a resource name: $name that doesn't match the " +
+ s"config name: $resourceName")
+ }
new ResourceInformation(name, addresses)
} catch {
case e @ (_: SparkException | _: MappingException | _: JsonParseException) =>
throw new SparkException(s"Error running the resource discovery script: $scriptFile" +
- s" for $resourceType", e)
+ s" for $resourceName", e)
}
} else {
- throw new SparkException(s"Resource script: $scriptFile to discover $resourceType" +
+ throw new SparkException(s"Resource script: $scriptFile to discover $resourceName" +
s" doesn't exist!")
}
} else {
- throw new SparkException(s"User is expecting to use $resourceType resources but " +
+ throw new SparkException(s"User is expecting to use $resourceName resources but " +
s"didn't specify a script via conf: $discoveryConf, to find them!")
}
result
}
+
+ /**
+ * Make sure the actual resources we have on startup are at least the number the user
+ * requested. Note that there is other code in SparkConf that makes sure we have executor configs
+ * for each task resource requirement and that they are large enough. This function
+ * is used by both driver and executors.
+ *
+ * @param requiredResources The resources that are required for us to run.
+ * @param actualResources The actual resources discovered.
+ */
+ def checkActualResourcesMeetRequirements(
+ requiredResources: Map[String, String],
+ actualResources: Map[String, ResourceInformation]): Unit = {
+ requiredResources.foreach { case (rName, reqCount) =>
+ val actualRInfo = actualResources.get(rName).getOrElse(
+ throw new SparkException(s"Resource: $rName required but wasn't discovered on startup"))
+
+ if (actualRInfo.addresses.size < reqCount.toLong) {
+ throw new SparkException(s"Resource: $rName, with addresses: " +
+ s"${actualRInfo.addresses.mkString(",")} " +
+ s"is less than what the user requested: $reqCount)")
+ }
+ }
+ }
}
diff --git a/core/src/main/scala/org/apache/spark/SparkConf.scala b/core/src/main/scala/org/apache/spark/SparkConf.scala
index bd2ef5b..15f1730 100644
--- a/core/src/main/scala/org/apache/spark/SparkConf.scala
+++ b/core/src/main/scala/org/apache/spark/SparkConf.scala
@@ -415,6 +415,13 @@ class SparkConf(loadDefaults: Boolean) extends Cloneable with Logging with Seria
.map { case (k, v) => (k.substring(prefix.length), v) }
}
+ /**
+ * Get all parameters that start with `prefix` and end with 'suffix'
+ */
+ def getAllWithPrefixAndSuffix(prefix: String, suffix: String): Array[(String, String)] = {
+ getAll.filter { case (k, v) => k.startsWith(prefix) && k.endsWith(suffix) }
+ .map { case (k, v) => (k.substring(prefix.length, (k.length - suffix.length)), v) }
+ }
/**
* Get a parameter as an integer, falling back to a default if not set
@@ -596,6 +603,30 @@ class SparkConf(loadDefaults: Boolean) extends Cloneable with Logging with Seria
require(executorTimeoutThresholdMs > executorHeartbeatIntervalMs, "The value of " +
s"${networkTimeout}=${executorTimeoutThresholdMs}ms must be no less than the value of " +
s"${EXECUTOR_HEARTBEAT_INTERVAL.key}=${executorHeartbeatIntervalMs}ms.")
+
+ // Make sure the executor resources were specified and are large enough if
+ // any task resources were specified.
+ val taskResourcesAndCount =
+ getAllWithPrefixAndSuffix(SPARK_TASK_RESOURCE_PREFIX, SPARK_RESOURCE_COUNT_SUFFIX).toMap
+ val executorResourcesAndCounts =
+ getAllWithPrefixAndSuffix(SPARK_EXECUTOR_RESOURCE_PREFIX, SPARK_RESOURCE_COUNT_SUFFIX).toMap
+
+ taskResourcesAndCount.foreach { case (rName, taskCount) =>
+ val execCount = executorResourcesAndCounts.get(rName).getOrElse(
+ throw new SparkException(
+ s"The executor resource config: " +
+ s"${SPARK_EXECUTOR_RESOURCE_PREFIX + rName + SPARK_RESOURCE_COUNT_SUFFIX} " +
+ "needs to be specified since a task requirement config: " +
+ s"${SPARK_TASK_RESOURCE_PREFIX + rName + SPARK_RESOURCE_COUNT_SUFFIX} was specified")
+ )
+ if (execCount.toLong < taskCount.toLong) {
+ throw new SparkException(
+ s"The executor resource config: " +
+ s"${SPARK_EXECUTOR_RESOURCE_PREFIX + rName + SPARK_RESOURCE_COUNT_SUFFIX} " +
+ s"= $execCount has to be >= the task config: " +
+ s"${SPARK_TASK_RESOURCE_PREFIX + rName + SPARK_RESOURCE_COUNT_SUFFIX} = $taskCount")
+ }
+ }
}
/**
@@ -789,6 +820,35 @@ private[spark] object SparkConf extends Logging {
}
/**
+ * A function to help parsing configs with multiple parts where the base and
+ * suffix could be one of many options. For instance configs like:
+ * spark.executor.resource.{resourceName}.{count/addresses}
+ * This function takes an Array of configs you got from the
+ * getAllWithPrefix function, selects only those that end with the suffix
+ * passed in and returns just the base part of the config before the first
+ * '.' and its value.
+ */
+ def getConfigsWithSuffix(
+ configs: Array[(String, String)],
+ suffix: String
+ ): Array[(String, String)] = {
+ configs.filter { case (rConf, _) => rConf.endsWith(suffix)}.
+ map { case (k, v) => (k.split('.').head, v) }
+ }
+
+ /**
+ * A function to help parsing configs with multiple parts where the base and
+ * suffix could be one of many options. For instance configs like:
+ * spark.executor.resource.{resourceName}.{count/addresses}
+ * This function takes an Array of configs you got from the
+ * getAllWithPrefix function and returns the base part of the config
+ * before the first '.'.
+ */
+ def getBaseOfConfigs(configs: Array[(String, String)]): Set[String] = {
+ configs.map { case (k, _) => k.split('.').head }.toSet
+ }
+
+ /**
* Holds information about keys that have been deprecated and do not have a replacement.
*
* @param key The deprecated key.
diff --git a/core/src/main/scala/org/apache/spark/SparkContext.scala b/core/src/main/scala/org/apache/spark/SparkContext.scala
index 9979410..878010d 100644
--- a/core/src/main/scala/org/apache/spark/SparkContext.scala
+++ b/core/src/main/scala/org/apache/spark/SparkContext.scala
@@ -213,6 +213,7 @@ class SparkContext(config: SparkConf) extends Logging {
private var _shutdownHookRef: AnyRef = _
private var _statusStore: AppStatusStore = _
private var _heartbeater: Heartbeater = _
+ private var _resources: scala.collection.immutable.Map[String, ResourceInformation] = _
/* ------------------------------------------------------------------------------------- *
| Accessors and public fields. These provide access to the internal state of the |
@@ -227,6 +228,8 @@ class SparkContext(config: SparkConf) extends Logging {
*/
def getConf: SparkConf = conf.clone()
+ def resources: Map[String, ResourceInformation] = _resources
+
def jars: Seq[String] = _jars
def files: Seq[String] = _files
def master: String = _conf.get("spark.master")
@@ -360,6 +363,48 @@ class SparkContext(config: SparkConf) extends Logging {
Utils.setLogLevel(org.apache.log4j.Level.toLevel(upperCased))
}
+ /**
+ * Checks to see if any resources (GPU/FPGA/etc) are available to the driver by looking
+ * at and processing the spark.driver.resource.resourceName.addresses and
+ * spark.driver.resource.resourceName.discoveryScript configs. The configs have to be
+ * present when the driver starts, setting them after startup does not work.
+ *
+ * If any resource addresses configs were specified then assume all resources will be specified
+ * in that way. Otherwise use the discovery scripts to find the resources. Users should
+ * not really be setting the addresses config directly and should not be mixing methods
+ * for different types of resources since the addresses config is meant for Standalone mode
+ * and other cluster managers should use the discovery scripts.
+ */
+ private def setupDriverResources(): Unit = {
+ // Only call getAllWithPrefix once and filter on those since there could be a lot of spark
+ // configs.
+ val allDriverResourceConfs = _conf.getAllWithPrefix(SPARK_DRIVER_RESOURCE_PREFIX)
+ val resourcesWithAddrsInConfs =
+ SparkConf.getConfigsWithSuffix(allDriverResourceConfs, SPARK_RESOURCE_ADDRESSES_SUFFIX)
+
+ _resources = if (resourcesWithAddrsInConfs.nonEmpty) {
+ resourcesWithAddrsInConfs.map { case (rName, addrString) =>
+ val addrsArray = addrString.split(",").map(_.trim())
+ (rName -> new ResourceInformation(rName, addrsArray))
+ }.toMap
+ } else {
+ // we already have the resources confs here so just pass in the unique resource names
+ // rather then having the resource discoverer reparse all the configs.
+ val uniqueResources = SparkConf.getBaseOfConfigs(allDriverResourceConfs)
+ ResourceDiscoverer.discoverResourcesInformation(_conf, SPARK_DRIVER_RESOURCE_PREFIX,
+ Some(uniqueResources))
+ }
+ // verify the resources we discovered are what the user requested
+ val driverReqResourcesAndCounts =
+ SparkConf.getConfigsWithSuffix(allDriverResourceConfs, SPARK_RESOURCE_COUNT_SUFFIX).toMap
+ ResourceDiscoverer.checkActualResourcesMeetRequirements(driverReqResourcesAndCounts, _resources)
+
+ logInfo("===============================================================================")
+ logInfo(s"Driver Resources:")
+ _resources.foreach { case (k, v) => logInfo(s"$k -> $v") }
+ logInfo("===============================================================================")
+ }
+
try {
_conf = config.clone()
_conf.validateSettings()
@@ -373,6 +418,8 @@ class SparkContext(config: SparkConf) extends Logging {
_driverLogger = DriverLogger(_conf)
+ setupDriverResources()
+
// log out spark.app.name in the Spark driver logs
logInfo(s"Submitted application: $appName")
diff --git a/core/src/main/scala/org/apache/spark/executor/CoarseGrainedExecutorBackend.scala b/core/src/main/scala/org/apache/spark/executor/CoarseGrainedExecutorBackend.scala
index af01e0b..fac4d40 100644
--- a/core/src/main/scala/org/apache/spark/executor/CoarseGrainedExecutorBackend.scala
+++ b/core/src/main/scala/org/apache/spark/executor/CoarseGrainedExecutorBackend.scala
@@ -83,51 +83,11 @@ private[spark] class CoarseGrainedExecutorBackend(
}(ThreadUtils.sameThread)
}
- // Check that the actual resources discovered will satisfy the user specified
- // requirements and that they match the configs specified by the user to catch
- // mismatches between what the user requested and what resource manager gave or
- // what the discovery script found.
- private def checkResourcesMeetRequirements(
- resourceConfigPrefix: String,
- reqResourcesAndCounts: Array[(String, String)],
- actualResources: Map[String, ResourceInformation]): Unit = {
-
- reqResourcesAndCounts.foreach { case (rName, reqCount) =>
- if (actualResources.contains(rName)) {
- val resourceInfo = actualResources(rName)
-
- if (resourceInfo.addresses.size < reqCount.toLong) {
- throw new SparkException(s"Resource: $rName with addresses: " +
- s"${resourceInfo.addresses.mkString(",")} doesn't meet the " +
- s"requirements of needing $reqCount of them")
- }
- // also make sure the resource count on start matches the
- // resource configs specified by user
- val userCountConfigName =
- resourceConfigPrefix + rName + SPARK_RESOURCE_COUNT_POSTFIX
- val userConfigCount = env.conf.getOption(userCountConfigName).
- getOrElse(throw new SparkException(s"Resource: $rName not specified " +
- s"via config: $userCountConfigName, but required, " +
- "please fix your configuration"))
-
- if (userConfigCount.toLong > resourceInfo.addresses.size) {
- throw new SparkException(s"Resource: $rName, with addresses: " +
- s"${resourceInfo.addresses.mkString(",")} " +
- s"is less than what the user requested for count: $userConfigCount, " +
- s"via $userCountConfigName")
- }
- } else {
- throw new SparkException(s"Executor resource config missing required task resource: $rName")
- }
- }
- }
-
// visible for testing
def parseOrFindResources(resourcesFile: Option[String]): Map[String, ResourceInformation] = {
// only parse the resources if a task requires them
- val taskResourceConfigs = env.conf.getAllWithPrefix(SPARK_TASK_RESOURCE_PREFIX)
- val resourceInfo = if (taskResourceConfigs.nonEmpty) {
- val execResources = resourcesFile.map { resourceFileStr => {
+ val resourceInfo = if (env.conf.getAllWithPrefix(SPARK_TASK_RESOURCE_PREFIX).nonEmpty) {
+ val actualExecResources = resourcesFile.map { resourceFileStr => {
val source = new BufferedInputStream(new FileInputStream(resourceFileStr))
val resourceMap = try {
val parsedJson = parse(source).asInstanceOf[JArray].arr
@@ -144,26 +104,26 @@ private[spark] class CoarseGrainedExecutorBackend(
source.close()
}
resourceMap
- }}.getOrElse(ResourceDiscoverer.findResources(env.conf, isDriver = false))
+ }}.getOrElse(ResourceDiscoverer.discoverResourcesInformation(env.conf,
+ SPARK_EXECUTOR_RESOURCE_PREFIX))
- if (execResources.isEmpty) {
+ if (actualExecResources.isEmpty) {
throw new SparkException("User specified resources per task via: " +
s"$SPARK_TASK_RESOURCE_PREFIX, but can't find any resources available on the executor.")
}
- // get just the map of resource name to count
- val resourcesAndCounts = taskResourceConfigs.
- withFilter { case (k, v) => k.endsWith(SPARK_RESOURCE_COUNT_POSTFIX)}.
- map { case (k, v) => (k.dropRight(SPARK_RESOURCE_COUNT_POSTFIX.size), v)}
+ val execReqResourcesAndCounts =
+ env.conf.getAllWithPrefixAndSuffix(SPARK_EXECUTOR_RESOURCE_PREFIX,
+ SPARK_RESOURCE_COUNT_SUFFIX).toMap
- checkResourcesMeetRequirements(SPARK_EXECUTOR_RESOURCE_PREFIX, resourcesAndCounts,
- execResources)
+ ResourceDiscoverer.checkActualResourcesMeetRequirements(execReqResourcesAndCounts,
+ actualExecResources)
logInfo("===============================================================================")
logInfo(s"Executor $executorId Resources:")
- execResources.foreach { case (k, v) => logInfo(s"$k -> $v") }
+ actualExecResources.foreach { case (k, v) => logInfo(s"$k -> $v") }
logInfo("===============================================================================")
- execResources
+ actualExecResources
} else {
if (resourcesFile.nonEmpty) {
logWarning(s"A resources file was specified but the application is not configured " +
diff --git a/core/src/main/scala/org/apache/spark/internal/config/package.scala b/core/src/main/scala/org/apache/spark/internal/config/package.scala
index 0aed1af..d9341bf 100644
--- a/core/src/main/scala/org/apache/spark/internal/config/package.scala
+++ b/core/src/main/scala/org/apache/spark/internal/config/package.scala
@@ -34,8 +34,9 @@ package object config {
private[spark] val SPARK_EXECUTOR_RESOURCE_PREFIX = "spark.executor.resource."
private[spark] val SPARK_TASK_RESOURCE_PREFIX = "spark.task.resource."
- private[spark] val SPARK_RESOURCE_COUNT_POSTFIX = ".count"
- private[spark] val SPARK_RESOURCE_DISCOVERY_SCRIPT_POSTFIX = ".discoveryScript"
+ private[spark] val SPARK_RESOURCE_COUNT_SUFFIX = ".count"
+ private[spark] val SPARK_RESOURCE_ADDRESSES_SUFFIX = ".addresses"
+ private[spark] val SPARK_RESOURCE_DISCOVERY_SCRIPT_SUFFIX = ".discoveryScript"
private[spark] val DRIVER_CLASS_PATH =
ConfigBuilder(SparkLauncher.DRIVER_EXTRA_CLASSPATH).stringConf.createOptional
diff --git a/core/src/test/scala/org/apache/spark/ResourceDiscovererSuite.scala b/core/src/test/scala/org/apache/spark/ResourceDiscovererSuite.scala
index 77f8a0b..623cd5c 100644
--- a/core/src/test/scala/org/apache/spark/ResourceDiscovererSuite.scala
+++ b/core/src/test/scala/org/apache/spark/ResourceDiscovererSuite.scala
@@ -40,7 +40,8 @@ class ResourceDiscovererSuite extends SparkFunSuite
test("Resource discoverer no resources") {
val sparkconf = new SparkConf
- val resources = ResourceDiscoverer.findResources(sparkconf, isDriver = false)
+ val resources =
+ ResourceDiscoverer.discoverResourcesInformation(sparkconf, SPARK_EXECUTOR_RESOURCE_PREFIX)
assert(resources.size === 0)
assert(resources.get("gpu").isEmpty,
"Should have a gpus entry that is empty")
@@ -54,8 +55,36 @@ class ResourceDiscovererSuite extends SparkFunSuite
val scriptPath = mockDiscoveryScript(gpuFile,
"""'{"name": "gpu","addresses":["0", "1"]}'""")
sparkconf.set(SPARK_EXECUTOR_RESOURCE_PREFIX + "gpu" +
- SPARK_RESOURCE_DISCOVERY_SCRIPT_POSTFIX, scriptPath)
- val resources = ResourceDiscoverer.findResources(sparkconf, isDriver = false)
+ SPARK_RESOURCE_DISCOVERY_SCRIPT_SUFFIX, scriptPath)
+ val resources =
+ ResourceDiscoverer.discoverResourcesInformation(sparkconf, SPARK_EXECUTOR_RESOURCE_PREFIX)
+ val gpuValue = resources.get("gpu")
+ assert(gpuValue.nonEmpty, "Should have a gpu entry")
+ assert(gpuValue.get.name == "gpu", "name should be gpu")
+ assert(gpuValue.get.addresses.size == 2, "Should have 2 indexes")
+ assert(gpuValue.get.addresses.deep == Array("0", "1").deep, "should have 0,1 entries")
+ }
+ }
+
+ test("Resource discoverer passed in resources") {
+ val sparkconf = new SparkConf
+ assume(!(Utils.isWindows))
+ withTempDir { dir =>
+ val gpuFile = new File(dir, "gpuDiscoverScript")
+ val gpuScript = mockDiscoveryScript(gpuFile,
+ """'{"name": "gpu","addresses":["0", "1"]}'""")
+ val fpgaFile = new File(dir, "fpgaDiscoverScript")
+ val fpgaScript = mockDiscoveryScript(fpgaFile,
+ """'{"name": "fpga","addresses":["f0", "f1"]}'""")
+ sparkconf.set(SPARK_EXECUTOR_RESOURCE_PREFIX + "gpu" +
+ SPARK_RESOURCE_DISCOVERY_SCRIPT_SUFFIX, gpuScript)
+ sparkconf.set(SPARK_EXECUTOR_RESOURCE_PREFIX + "fpga" +
+ SPARK_RESOURCE_DISCOVERY_SCRIPT_SUFFIX, fpgaScript)
+ // it should only look at the resources passed in and ignore fpga conf
+ val resources =
+ ResourceDiscoverer.discoverResourcesInformation(sparkconf,
+ SPARK_EXECUTOR_RESOURCE_PREFIX, Some(Set("gpu")))
+ assert(resources.size === 1, "should only have the gpu resource")
val gpuValue = resources.get("gpu")
assert(gpuValue.nonEmpty, "Should have a gpu entry")
assert(gpuValue.get.name == "gpu", "name should be gpu")
@@ -72,8 +101,9 @@ class ResourceDiscovererSuite extends SparkFunSuite
val scriptPath = mockDiscoveryScript(gpuFile,
"""'{"name": "gpu"}'""")
sparkconf.set(SPARK_EXECUTOR_RESOURCE_PREFIX + "gpu" +
- SPARK_RESOURCE_DISCOVERY_SCRIPT_POSTFIX, scriptPath)
- val resources = ResourceDiscoverer.findResources(sparkconf, isDriver = false)
+ SPARK_RESOURCE_DISCOVERY_SCRIPT_SUFFIX, scriptPath)
+ val resources =
+ ResourceDiscoverer.discoverResourcesInformation(sparkconf, SPARK_EXECUTOR_RESOURCE_PREFIX)
val gpuValue = resources.get("gpu")
assert(gpuValue.nonEmpty, "Should have a gpu entry")
assert(gpuValue.get.name == "gpu", "name should be gpu")
@@ -89,15 +119,16 @@ class ResourceDiscovererSuite extends SparkFunSuite
val gpuDiscovery = mockDiscoveryScript(gpuFile,
"""'{"name": "gpu", "addresses": ["0", "1"]}'""")
sparkconf.set(SPARK_EXECUTOR_RESOURCE_PREFIX + "gpu" +
- SPARK_RESOURCE_DISCOVERY_SCRIPT_POSTFIX, gpuDiscovery)
+ SPARK_RESOURCE_DISCOVERY_SCRIPT_SUFFIX, gpuDiscovery)
val fpgaFile = new File(dir, "fpgaDiscoverScript")
val fpgaDiscovery = mockDiscoveryScript(fpgaFile,
"""'{"name": "fpga", "addresses": ["f1", "f2", "f3"]}'""")
sparkconf.set(SPARK_EXECUTOR_RESOURCE_PREFIX + "fpga" +
- SPARK_RESOURCE_DISCOVERY_SCRIPT_POSTFIX, fpgaDiscovery)
+ SPARK_RESOURCE_DISCOVERY_SCRIPT_SUFFIX, fpgaDiscovery)
- val resources = ResourceDiscoverer.findResources(sparkconf, isDriver = false)
+ val resources =
+ ResourceDiscoverer.discoverResourcesInformation(sparkconf, SPARK_EXECUTOR_RESOURCE_PREFIX)
assert(resources.size === 2)
val gpuValue = resources.get("gpu")
assert(gpuValue.nonEmpty, "Should have a gpu entry")
@@ -122,11 +153,12 @@ class ResourceDiscovererSuite extends SparkFunSuite
val gpuDiscovery = mockDiscoveryScript(gpuFile,
"""'{"name": "gpu", "addresses": ["0", "1"]}'""")
sparkconf.set(SPARK_DRIVER_RESOURCE_PREFIX + "gpu" +
- SPARK_RESOURCE_DISCOVERY_SCRIPT_POSTFIX, gpuDiscovery)
+ SPARK_RESOURCE_DISCOVERY_SCRIPT_SUFFIX, gpuDiscovery)
sparkconf set(SPARK_EXECUTOR_RESOURCE_PREFIX + "gpu" +
- SPARK_RESOURCE_DISCOVERY_SCRIPT_POSTFIX, "boguspath")
+ SPARK_RESOURCE_DISCOVERY_SCRIPT_SUFFIX, "boguspath")
// make sure it reads from correct config, here it should use driver
- val resources = ResourceDiscoverer.findResources(sparkconf, isDriver = true)
+ val resources =
+ ResourceDiscoverer.discoverResourcesInformation(sparkconf, SPARK_DRIVER_RESOURCE_PREFIX)
val gpuValue = resources.get("gpu")
assert(gpuValue.nonEmpty, "Should have a gpu entry")
assert(gpuValue.get.name == "gpu", "name should be gpu")
@@ -135,6 +167,23 @@ class ResourceDiscovererSuite extends SparkFunSuite
}
}
+ test("Resource discoverer script returns mismatched name") {
+ val sparkconf = new SparkConf
+ assume(!(Utils.isWindows))
+ withTempDir { dir =>
+ val gpuFile = new File(dir, "gpuDiscoverScript")
+ val gpuDiscovery = mockDiscoveryScript(gpuFile,
+ """'{"name": "fpga", "addresses": ["0", "1"]}'""")
+ sparkconf.set(SPARK_DRIVER_RESOURCE_PREFIX + "gpu" +
+ SPARK_RESOURCE_DISCOVERY_SCRIPT_SUFFIX, gpuDiscovery)
+ val error = intercept[SparkException] {
+ ResourceDiscoverer.discoverResourcesInformation(sparkconf, SPARK_DRIVER_RESOURCE_PREFIX)
+ }.getMessage()
+
+ assert(error.contains("Error running the resource discovery script"))
+ }
+ }
+
test("Resource discoverer script returns invalid format") {
val sparkconf = new SparkConf
assume(!(Utils.isWindows))
@@ -143,9 +192,9 @@ class ResourceDiscovererSuite extends SparkFunSuite
val gpuDiscovery = mockDiscoveryScript(gpuFile,
"""'{"addresses": ["0", "1"]}'""")
sparkconf.set(SPARK_EXECUTOR_RESOURCE_PREFIX + "gpu" +
- SPARK_RESOURCE_DISCOVERY_SCRIPT_POSTFIX, gpuDiscovery)
+ SPARK_RESOURCE_DISCOVERY_SCRIPT_SUFFIX, gpuDiscovery)
val error = intercept[SparkException] {
- ResourceDiscoverer.findResources(sparkconf, isDriver = false)
+ ResourceDiscoverer.discoverResourcesInformation(sparkconf, SPARK_EXECUTOR_RESOURCE_PREFIX)
}.getMessage()
assert(error.contains("Error running the resource discovery"))
@@ -158,10 +207,10 @@ class ResourceDiscovererSuite extends SparkFunSuite
val file1 = new File(dir, "bogusfilepath")
try {
sparkconf.set(SPARK_EXECUTOR_RESOURCE_PREFIX + "gpu" +
- SPARK_RESOURCE_DISCOVERY_SCRIPT_POSTFIX, file1.getPath())
+ SPARK_RESOURCE_DISCOVERY_SCRIPT_SUFFIX, file1.getPath())
val error = intercept[SparkException] {
- ResourceDiscoverer.findResources(sparkconf, isDriver = false)
+ ResourceDiscoverer.discoverResourcesInformation(sparkconf, SPARK_EXECUTOR_RESOURCE_PREFIX)
}.getMessage()
assert(error.contains("doesn't exist"))
@@ -174,10 +223,10 @@ class ResourceDiscovererSuite extends SparkFunSuite
test("gpu's specified but not discovery script") {
val sparkconf = new SparkConf
sparkconf.set(SPARK_EXECUTOR_RESOURCE_PREFIX + "gpu" +
- SPARK_RESOURCE_COUNT_POSTFIX, "2")
+ SPARK_RESOURCE_COUNT_SUFFIX, "2")
val error = intercept[SparkException] {
- ResourceDiscoverer.findResources(sparkconf, isDriver = false)
+ ResourceDiscoverer.discoverResourcesInformation(sparkconf, SPARK_EXECUTOR_RESOURCE_PREFIX)
}.getMessage()
assert(error.contains("User is expecting to use"))
diff --git a/core/src/test/scala/org/apache/spark/SparkConfSuite.scala b/core/src/test/scala/org/apache/spark/SparkConfSuite.scala
index 83a9ea3..965c7df 100644
--- a/core/src/test/scala/org/apache/spark/SparkConfSuite.scala
+++ b/core/src/test/scala/org/apache/spark/SparkConfSuite.scala
@@ -110,6 +110,49 @@ class SparkConfSuite extends SparkFunSuite with LocalSparkContext with ResetSyst
assert(conf.getOption("k4") === None)
}
+ test("basic getAllWithPrefixAndPostfix") {
+ val conf = new SparkConf(false)
+ conf.set("spark.prefix.main.suffix", "v1")
+ val prefix = "spark.prefix."
+ val suffix = ".suffix"
+ assert(conf.getAllWithPrefixAndSuffix(prefix, suffix).toSet ===
+ Set(("main", "v1")))
+
+ conf.set("spark.prefix.main2.suffix", "v2")
+ conf.set("spark.prefix.main3.extra1.suffix", "v3")
+ conf.set("spark.prefix.main4.extra2.nonmatchingsuffix", "v4")
+ conf.set("spark.notmatchingprefix.main4.suffix", "v5")
+
+ assert(conf.getAllWithPrefixAndSuffix(prefix, suffix).toSet ===
+ Set(("main", "v1"), ("main2", "v2"), ("main3.extra1", "v3")))
+ }
+
+ test("test prefix config parsing utilities") {
+ val conf = new SparkConf(false)
+ conf.set("spark.prefix.main.suffix", "v1")
+ val prefix = "spark.prefix."
+ val suffix = ".suffix"
+ val configsWithPrefix = conf.getAllWithPrefix(prefix)
+ assert(configsWithPrefix.toSet === Set(("main.suffix", "v1")))
+ assert(SparkConf.getBaseOfConfigs(configsWithPrefix) === Set("main"))
+ assert(SparkConf.getConfigsWithSuffix(configsWithPrefix, suffix).toSet === Set(("main", "v1")))
+ }
+
+ test("basic getAllWithPrefix") {
+ val prefix = "spark.prefix."
+ val conf = new SparkConf(false)
+ conf.set("spark.prefix.main.suffix", "v1")
+ assert(conf.getAllWithPrefix(prefix).toSet ===
+ Set(("main.suffix", "v1")))
+
+ conf.set("spark.prefix.main2.suffix", "v2")
+ conf.set("spark.prefix.main3.extra1.suffix", "v3")
+ conf.set("spark.notMatching.main4", "v4")
+
+ assert(conf.getAllWithPrefix(prefix).toSet ===
+ Set(("main.suffix", "v1"), ("main2.suffix", "v2"), ("main3.extra1.suffix", "v3")))
+ }
+
test("creating SparkContext without master and app name") {
val conf = new SparkConf(false)
intercept[SparkException] { sc = new SparkContext(conf) }
diff --git a/core/src/test/scala/org/apache/spark/SparkContextSuite.scala b/core/src/test/scala/org/apache/spark/SparkContextSuite.scala
index 1dcb2f7..abd7d8a 100644
--- a/core/src/test/scala/org/apache/spark/SparkContextSuite.scala
+++ b/core/src/test/scala/org/apache/spark/SparkContextSuite.scala
@@ -20,6 +20,9 @@ package org.apache.spark
import java.io.File
import java.net.{MalformedURLException, URI}
import java.nio.charset.StandardCharsets
+import java.nio.file.{Files => JavaFiles}
+import java.nio.file.attribute.PosixFilePermission._
+import java.util.EnumSet
import java.util.concurrent.{CountDownLatch, Semaphore, TimeUnit}
import scala.concurrent.duration._
@@ -731,6 +734,101 @@ class SparkContextSuite extends SparkFunSuite with LocalSparkContext with Eventu
resetSparkContext()
}
}
+
+ test("test gpu driver discovery under local-cluster mode") {
+ withTempDir { dir =>
+ val gpuFile = new File(dir, "gpuDiscoverScript")
+ val scriptPath = mockDiscoveryScript(gpuFile,
+ """'{"name": "gpu","addresses":["5", "6"]}'""")
+
+ val conf = new SparkConf()
+ .set(SPARK_DRIVER_RESOURCE_PREFIX + "gpu" +
+ SPARK_RESOURCE_COUNT_SUFFIX, "1")
+ .set(SPARK_DRIVER_RESOURCE_PREFIX + "gpu" +
+ SPARK_RESOURCE_DISCOVERY_SCRIPT_SUFFIX, scriptPath)
+ .setMaster("local-cluster[1, 1, 1024]")
+ .setAppName("test-cluster")
+ sc = new SparkContext(conf)
+
+ // Ensure all executors has started
+ eventually(timeout(10.seconds)) {
+ assert(sc.statusTracker.getExecutorInfos.size == 1)
+ }
+ assert(sc.resources.size === 1)
+ assert(sc.resources.get("gpu").get.addresses === Array("5", "6"))
+ assert(sc.resources.get("gpu").get.name === "gpu")
+ }
+ }
+
+ test("test gpu driver resource address and discovery under local-cluster mode") {
+ withTempDir { dir =>
+ val gpuFile = new File(dir, "gpuDiscoverScript")
+ val scriptPath = mockDiscoveryScript(gpuFile,
+ """'{"name": "gpu","addresses":["5", "6"]}'""")
+
+ val conf = new SparkConf()
+ .set(SPARK_DRIVER_RESOURCE_PREFIX + "gpu" +
+ SPARK_RESOURCE_COUNT_SUFFIX, "1")
+ .set(SPARK_DRIVER_RESOURCE_PREFIX + "gpu" +
+ SPARK_RESOURCE_ADDRESSES_SUFFIX, "0, 1, 8")
+ .set(SPARK_DRIVER_RESOURCE_PREFIX + "gpu" +
+ SPARK_RESOURCE_DISCOVERY_SCRIPT_SUFFIX, scriptPath)
+ .setMaster("local-cluster[1, 1, 1024]")
+ .setAppName("test-cluster")
+ sc = new SparkContext(conf)
+
+ // Ensure all executors has started
+ eventually(timeout(10.seconds)) {
+ assert(sc.statusTracker.getExecutorInfos.size == 1)
+ }
+ // driver gpu addresses config should take precedence over the script
+ assert(sc.resources.size === 1)
+ assert(sc.resources.get("gpu").get.addresses === Array("0", "1", "8"))
+ assert(sc.resources.get("gpu").get.name === "gpu")
+ }
+ }
+
+ test("Test parsing resources task configs with missing executor config") {
+ val conf = new SparkConf()
+ .set(SPARK_TASK_RESOURCE_PREFIX + "gpu" +
+ SPARK_RESOURCE_COUNT_SUFFIX, "1")
+ .setMaster("local-cluster[1, 1, 1024]")
+ .setAppName("test-cluster")
+
+ var error = intercept[SparkException] {
+ sc = new SparkContext(conf)
+ }.getMessage()
+
+ assert(error.contains("The executor resource config: spark.executor.resource.gpu.count " +
+ "needs to be specified since a task requirement config: spark.task.resource.gpu.count " +
+ "was specified"))
+ }
+
+ test("Test parsing resources executor config < task requirements") {
+ val conf = new SparkConf()
+ .set(SPARK_TASK_RESOURCE_PREFIX + "gpu" +
+ SPARK_RESOURCE_COUNT_SUFFIX, "2")
+ .set(SPARK_EXECUTOR_RESOURCE_PREFIX + "gpu" +
+ SPARK_RESOURCE_COUNT_SUFFIX, "1")
+ .setMaster("local-cluster[1, 1, 1024]")
+ .setAppName("test-cluster")
+
+ var error = intercept[SparkException] {
+ sc = new SparkContext(conf)
+ }.getMessage()
+
+ assert(error.contains("The executor resource config: " +
+ "spark.executor.resource.gpu.count = 1 has to be >= the task config: " +
+ "spark.task.resource.gpu.count = 2"))
+ }
+
+ def mockDiscoveryScript(file: File, result: String): String = {
+ Files.write(s"echo $result", file, StandardCharsets.UTF_8)
+ JavaFiles.setPosixFilePermissions(file.toPath(),
+ EnumSet.of(OWNER_READ, OWNER_EXECUTE, OWNER_WRITE))
+ file.getPath()
+ }
+
}
object SparkContextSuite {
diff --git a/core/src/test/scala/org/apache/spark/executor/CoarseGrainedExecutorBackendSuite.scala b/core/src/test/scala/org/apache/spark/executor/CoarseGrainedExecutorBackendSuite.scala
index e59d1b0..43913d1 100644
--- a/core/src/test/scala/org/apache/spark/executor/CoarseGrainedExecutorBackendSuite.scala
+++ b/core/src/test/scala/org/apache/spark/executor/CoarseGrainedExecutorBackendSuite.scala
@@ -49,7 +49,7 @@ class CoarseGrainedExecutorBackendSuite extends SparkFunSuite
test("parsing no resources") {
val conf = new SparkConf
- conf.set(SPARK_TASK_RESOURCE_PREFIX + "gpu" + SPARK_RESOURCE_COUNT_POSTFIX, "2")
+ conf.set(SPARK_TASK_RESOURCE_PREFIX + "gpu" + SPARK_RESOURCE_COUNT_SUFFIX, "2")
val serializer = new JavaSerializer(conf)
val env = createMockEnv(conf, serializer)
@@ -71,8 +71,8 @@ class CoarseGrainedExecutorBackendSuite extends SparkFunSuite
test("parsing one resources") {
val conf = new SparkConf
- conf.set(SPARK_EXECUTOR_RESOURCE_PREFIX + "gpu" + SPARK_RESOURCE_COUNT_POSTFIX, "2")
- conf.set(SPARK_TASK_RESOURCE_PREFIX + "gpu" + SPARK_RESOURCE_COUNT_POSTFIX, "2")
+ conf.set(SPARK_EXECUTOR_RESOURCE_PREFIX + "gpu" + SPARK_RESOURCE_COUNT_SUFFIX, "2")
+ conf.set(SPARK_TASK_RESOURCE_PREFIX + "gpu" + SPARK_RESOURCE_COUNT_SUFFIX, "2")
val serializer = new JavaSerializer(conf)
val env = createMockEnv(conf, serializer)
// we don't really use this, just need it to get at the parser function
@@ -95,10 +95,10 @@ class CoarseGrainedExecutorBackendSuite extends SparkFunSuite
test("parsing multiple resources") {
val conf = new SparkConf
- conf.set(SPARK_EXECUTOR_RESOURCE_PREFIX + "fpga" + SPARK_RESOURCE_COUNT_POSTFIX, "3")
- conf.set(SPARK_TASK_RESOURCE_PREFIX + "fpga" + SPARK_RESOURCE_COUNT_POSTFIX, "3")
- conf.set(SPARK_EXECUTOR_RESOURCE_PREFIX + "gpu" + SPARK_RESOURCE_COUNT_POSTFIX, "2")
- conf.set(SPARK_TASK_RESOURCE_PREFIX + "gpu" + SPARK_RESOURCE_COUNT_POSTFIX, "2")
+ conf.set(SPARK_EXECUTOR_RESOURCE_PREFIX + "fpga" + SPARK_RESOURCE_COUNT_SUFFIX, "3")
+ conf.set(SPARK_TASK_RESOURCE_PREFIX + "fpga" + SPARK_RESOURCE_COUNT_SUFFIX, "3")
+ conf.set(SPARK_EXECUTOR_RESOURCE_PREFIX + "gpu" + SPARK_RESOURCE_COUNT_SUFFIX, "2")
+ conf.set(SPARK_TASK_RESOURCE_PREFIX + "gpu" + SPARK_RESOURCE_COUNT_SUFFIX, "2")
val serializer = new JavaSerializer(conf)
val env = createMockEnv(conf, serializer)
// we don't really use this, just need it to get at the parser function
@@ -128,8 +128,8 @@ class CoarseGrainedExecutorBackendSuite extends SparkFunSuite
test("error checking parsing resources and executor and task configs") {
val conf = new SparkConf
- conf.set(SPARK_EXECUTOR_RESOURCE_PREFIX + "gpu" + SPARK_RESOURCE_COUNT_POSTFIX, "2")
- conf.set(SPARK_TASK_RESOURCE_PREFIX + "gpu" + SPARK_RESOURCE_COUNT_POSTFIX, "2")
+ conf.set(SPARK_EXECUTOR_RESOURCE_PREFIX + "gpu" + SPARK_RESOURCE_COUNT_SUFFIX, "2")
+ conf.set(SPARK_TASK_RESOURCE_PREFIX + "gpu" + SPARK_RESOURCE_COUNT_SUFFIX, "2")
val serializer = new JavaSerializer(conf)
val env = createMockEnv(conf, serializer)
// we don't really use this, just need it to get at the parser function
@@ -148,7 +148,8 @@ class CoarseGrainedExecutorBackendSuite extends SparkFunSuite
val parsedResources = backend.parseOrFindResources(Some(f1))
}.getMessage()
- assert(error.contains("doesn't meet the requirements of needing"))
+ assert(error.contains("Resource: gpu, with addresses: 0 is less than what the " +
+ "user requested: 2"))
}
// missing resource on the executor
@@ -163,14 +164,14 @@ class CoarseGrainedExecutorBackendSuite extends SparkFunSuite
val parsedResources = backend.parseOrFindResources(Some(f1))
}.getMessage()
- assert(error.contains("Executor resource config missing required task resource"))
+ assert(error.contains("Resource: gpu required but wasn't discovered on startup"))
}
}
test("executor resource found less than required") {
val conf = new SparkConf
- conf.set(SPARK_EXECUTOR_RESOURCE_PREFIX + "gpu" + SPARK_RESOURCE_COUNT_POSTFIX, "4")
- conf.set(SPARK_TASK_RESOURCE_PREFIX + "gpu" + SPARK_RESOURCE_COUNT_POSTFIX, "1")
+ conf.set(SPARK_EXECUTOR_RESOURCE_PREFIX + "gpu" + SPARK_RESOURCE_COUNT_SUFFIX, "4")
+ conf.set(SPARK_TASK_RESOURCE_PREFIX + "gpu" + SPARK_RESOURCE_COUNT_SUFFIX, "1")
val serializer = new JavaSerializer(conf)
val env = createMockEnv(conf, serializer)
// we don't really use this, just need it to get at the parser function
@@ -189,40 +190,14 @@ class CoarseGrainedExecutorBackendSuite extends SparkFunSuite
val parsedResources = backend.parseOrFindResources(Some(f1))
}.getMessage()
- assert(error.contains("is less than what the user requested for count"))
- }
- }
-
- test("parsing resources task configs with missing executor config") {
- val conf = new SparkConf
- conf.set(SPARK_TASK_RESOURCE_PREFIX + "gpu" + SPARK_RESOURCE_COUNT_POSTFIX, "2")
- val serializer = new JavaSerializer(conf)
- val env = createMockEnv(conf, serializer)
- // we don't really use this, just need it to get at the parser function
- val backend = new CoarseGrainedExecutorBackend(env.rpcEnv, "driverurl", "1", "host1",
- 4, Seq.empty[URL], env, None)
-
- withTempDir { tmpDir =>
- val gpuArgs =
- ("name" -> "gpu") ~
- ("addresses" -> Seq("0", "1"))
- val ja = JArray(List(gpuArgs))
- val f1 = writeFileWithJson(tmpDir, ja)
-
- var error = intercept[SparkException] {
- val parsedResources = backend.parseOrFindResources(Some(f1))
- }.getMessage()
-
- assert(error.contains("Resource: gpu not specified via config: " +
- "spark.executor.resource.gpu.count, but required, please " +
- "fix your configuration"))
+ assert(error.contains("gpu, with addresses: 0,1 is less than what the user requested: 4"))
}
}
test("use discoverer") {
val conf = new SparkConf
- conf.set(SPARK_EXECUTOR_RESOURCE_PREFIX + "fpga" + SPARK_RESOURCE_COUNT_POSTFIX, "3")
- conf.set(SPARK_TASK_RESOURCE_PREFIX + "fpga" + SPARK_RESOURCE_COUNT_POSTFIX, "3")
+ conf.set(SPARK_EXECUTOR_RESOURCE_PREFIX + "fpga" + SPARK_RESOURCE_COUNT_SUFFIX, "3")
+ conf.set(SPARK_TASK_RESOURCE_PREFIX + "fpga" + SPARK_RESOURCE_COUNT_SUFFIX, "3")
assume(!(Utils.isWindows))
withTempDir { dir =>
val fpgaDiscovery = new File(dir, "resourceDiscoverScriptfpga")
@@ -231,7 +206,7 @@ class CoarseGrainedExecutorBackendSuite extends SparkFunSuite
JavaFiles.setPosixFilePermissions(fpgaDiscovery.toPath(),
EnumSet.of(OWNER_READ, OWNER_EXECUTE, OWNER_WRITE))
conf.set(SPARK_EXECUTOR_RESOURCE_PREFIX + "fpga" +
- SPARK_RESOURCE_DISCOVERY_SCRIPT_POSTFIX, fpgaDiscovery.getPath())
+ SPARK_RESOURCE_DISCOVERY_SCRIPT_SUFFIX, fpgaDiscovery.getPath())
val serializer = new JavaSerializer(conf)
val env = createMockEnv(conf, serializer)
diff --git a/docs/configuration.md b/docs/configuration.md
index d0b2699..d20b416 100644
--- a/docs/configuration.md
+++ b/docs/configuration.md
@@ -188,6 +188,25 @@ of the most common options to set are:
</td>
</tr>
<tr>
+ <td><code>spark.driver.resource.{resourceName}.count</code></td>
+ <td>0</td>
+ <td>
+ The number of a particular resource type to use on the driver.
+ If this is used, you must also specify the
+ <code>spark.driver.resource.{resourceName}.discoveryScript</code>
+ for the driver to find the resource on startup.
+ </td>
+</tr>
+<tr>
+ <td><code>spark.driver.resource.{resourceName}.discoveryScript</code></td>
+ <td>None</td>
+ <td>
+ A script for the driver to run to discover a particular resource type. This should
+ write to STDOUT a JSON string in the format of the ResourceInformation class. This has a
+ name and an array of addresses.
+ </td>
+</tr>
+<tr>
<td><code>spark.executor.memory</code></td>
<td>1g</td>
<td>
@@ -222,17 +241,17 @@ of the most common options to set are:
</td>
</tr>
<tr>
- <td><code>spark.executor.resource.{resourceType}.count</code></td>
+ <td><code>spark.executor.resource.{resourceName}.count</code></td>
<td>0</td>
<td>
The number of a particular resource type to use per executor process.
If this is used, you must also specify the
- <code>spark.executor.resource.{resourceType}.discoveryScript</code>
+ <code>spark.executor.resource.{resourceName}.discoveryScript</code>
for the executor to find the resource on startup.
</td>
</tr>
<tr>
- <td><code>spark.executor.resource.{resourceType}.discoveryScript</code></td>
+ <td><code>spark.executor.resource.{resourceName}.discoveryScript</code></td>
<td>None</td>
<td>
A script for the executor to run to discover a particular resource type. This should
@@ -1813,11 +1832,11 @@ Apart from these, the following properties are also available, and may be useful
</td>
</tr>
<tr>
- <td><code>spark.task.resource.{resourceType}.count</code></td>
+ <td><code>spark.task.resource.{resourceName}.count</code></td>
<td>1</td>
<td>
Number of a particular resource type to allocate for each task. If this is specified
- you must also provide the executor config <code>spark.executor.resource.{resourceType}.count</code>
+ you must also provide the executor config <code>spark.executor.resource.{resourceName}.count</code>
and any corresponding discovery configs so that your executors are created with that resource type.
</td>
</tr>
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org