[SPARK-27959][YARN] Change YARN resource configs to use .amount

docs/running-on-yarn.md

@@ -142,40 +142,40 @@ To use a custom metrics.properties for the application master and executors, upd
   </td>
 </tr>
 <tr>
-  <td><code>spark.yarn.am.resource.{resource-type}</code></td>
+  <td><code>spark.yarn.am.resource.{resource-type}.amount</code></td>
   <td><code>(none)</code></td>
   <td>
     Amount of resource to use for the YARN Application Master in client mode.
-    In cluster mode, use <code>spark.yarn.driver.resource.&lt;resource-type&gt;</code> instead.
+    In cluster mode, use <code>spark.yarn.driver.resource.&lt;resource-type&gt;.amount</code> instead.
     Please note that this feature can be used only with YARN 3.0+
     For reference, see YARN Resource Model documentation: https://hadoop.apache.org/docs/r3.0.1/hadoop-yarn/hadoop-yarn-site/ResourceModel.html
     <p/>
     Example: 
-    To request GPU resources from YARN, use: <code>spark.yarn.am.resource.yarn.io/gpu</code>
+    To request GPU resources from YARN, use: <code>spark.yarn.am.resource.yarn.io/gpu.amount</code>
   </td>
 </tr>
 <tr>
-  <td><code>spark.yarn.driver.resource.{resource-type}</code></td>
+  <td><code>spark.yarn.driver.resource.{resource-type}.amount</code></td>
   <td><code>(none)</code></td>
   <td>
     Amount of resource to use for the YARN Application Master in cluster mode.
     Please note that this feature can be used only with YARN 3.0+
     For reference, see YARN Resource Model documentation: https://hadoop.apache.org/docs/r3.0.1/hadoop-yarn/hadoop-yarn-site/ResourceModel.html
     <p/>
     Example: 
-    To request GPU resources from YARN, use: <code>spark.yarn.driver.resource.yarn.io/gpu</code>
+    To request GPU resources from YARN, use: <code>spark.yarn.driver.resource.yarn.io/gpu.amount</code>
   </td>
 </tr>
 <tr>
-  <td><code>spark.yarn.executor.resource.{resource-type}</code></td>
+  <td><code>spark.yarn.executor.resource.{resource-type}.amount</code></td>
   <td><code>(none)</code></td>
  <td>
      Amount of resource to use per executor process.
      Please note that this feature can be used only with YARN 3.0+
      For reference, see YARN Resource Model documentation: https://hadoop.apache.org/docs/r3.0.1/hadoop-yarn/hadoop-yarn-site/ResourceModel.html
      <p/>
      Example: 
-     To request GPU resources from YARN, use: <code>spark.yarn.executor.resource.yarn.io/gpu</code>
+     To request GPU resources from YARN, use: <code>spark.yarn.executor.resource.yarn.io/gpu.amount</code>
  </td>
 </tr>
 <tr>

resource-managers/yarn/src/main/scala/org/apache/spark/deploy/yarn/Client.scala

@@ -51,7 +51,7 @@ import org.apache.spark.{SecurityManager, SparkConf, SparkException}
 import org.apache.spark.api.python.PythonUtils
 import org.apache.spark.deploy.{SparkApplication, SparkHadoopUtil}
 import org.apache.spark.deploy.security.HadoopDelegationTokenManager
-import org.apache.spark.deploy.yarn.YarnSparkHadoopUtil._
+import org.apache.spark.deploy.yarn.ResourceRequestHelper._
 import org.apache.spark.deploy.yarn.config._
 import org.apache.spark.internal.Logging
 import org.apache.spark.internal.config._
@@ -241,12 +241,12 @@ private[spark] class Client(
       newApp: YarnClientApplication,
       containerContext: ContainerLaunchContext): ApplicationSubmissionContext = {
 
-    val yarnAMResources =
-      if (isClusterMode) {
-        sparkConf.getAllWithPrefix(config.YARN_DRIVER_RESOURCE_TYPES_PREFIX).toMap
-      } else {
-        sparkConf.getAllWithPrefix(config.YARN_AM_RESOURCE_TYPES_PREFIX).toMap
-      }
+    val componentName = if (isClusterMode) {
+      config.YARN_DRIVER_RESOURCE_TYPES_PREFIX
+    } else {
+      config.YARN_AM_RESOURCE_TYPES_PREFIX
+    }
+    val yarnAMResources = getYarnResourcesAndAmounts(sparkConf, componentName)
     val amResources = yarnAMResources ++
       getYarnResourcesFromSparkResources(SPARK_DRIVER_PREFIX, sparkConf)
     logDebug(s"AM resources: $amResources")

resource-managers/yarn/src/main/scala/org/apache/spark/deploy/yarn/ResourceRequestHelper.scala

@@ -26,11 +26,11 @@ import scala.util.Try
 import org.apache.hadoop.yarn.api.records.Resource
 
 import org.apache.spark.{SparkConf, SparkException}
-import org.apache.spark.deploy.yarn.YarnSparkHadoopUtil._
 import org.apache.spark.deploy.yarn.config._
 import org.apache.spark.internal.Logging
 import org.apache.spark.internal.config._
 import org.apache.spark.resource.ResourceID
+import org.apache.spark.resource.ResourceUtils.{AMOUNT, FPGA, GPU}
 import org.apache.spark.util.{CausedBy, Utils}
 
 /**
@@ -40,6 +40,45 @@ import org.apache.spark.util.{CausedBy, Utils}
 private object ResourceRequestHelper extends Logging {
   private val AMOUNT_AND_UNIT_REGEX = "([0-9]+)([A-Za-z]*)".r
   private val RESOURCE_INFO_CLASS = "org.apache.hadoop.yarn.api.records.ResourceInformation"
+  val YARN_GPU_RESOURCE_CONFIG = "yarn.io/gpu"
+  val YARN_FPGA_RESOURCE_CONFIG = "yarn.io/fpga"
+
+  private[yarn] def getYarnResourcesAndAmounts(
+      sparkConf: SparkConf,
+      componentName: String): Map[String, String] = {
+    sparkConf.getAllWithPrefix(s"$componentName").map { case (key, value) =>
+      val splitIndex = key.lastIndexOf('.')
+      if (splitIndex == -1) {
+        val errorMessage = s"Missing suffix for ${componentName}${key}, you must specify" +
+          s" a suffix - $AMOUNT is currently the only supported suffix."
+        throw new IllegalArgumentException(errorMessage.toString())
+      }
+      val resourceName = key.substring(0, splitIndex)
+      val resourceSuffix = key.substring(splitIndex + 1)
+      if (!AMOUNT.equals(resourceSuffix)) {
+        val errorMessage = s"Unsupported suffix: $resourceSuffix in: ${componentName}${key}, " +
+          s"only .$AMOUNT is supported."
+        throw new IllegalArgumentException(errorMessage.toString())
+      }
+      (resourceName, value)
+    }.toMap
+  }
+
+  /**
+   * Convert Spark resources into YARN resources.
+   * The only resources we know how to map from spark configs to yarn configs are
+   * gpus and fpgas, everything else the user has to specify them in both the
+   * spark.yarn.*.resource and the spark.*.resource configs.
+   */
+  private[yarn] def getYarnResourcesFromSparkResources(
+      confPrefix: String,
+      sparkConf: SparkConf
+  ): Map[String, String] = {
+    Map(GPU -> YARN_GPU_RESOURCE_CONFIG, FPGA -> YARN_FPGA_RESOURCE_CONFIG).map {
+      case (rName, yarnName) =>
+        (yarnName -> sparkConf.get(ResourceID(confPrefix, rName).amountConf, "0"))
+    }.filter { case (_, count) => count.toLong > 0 }
+  }
 
   /**
    * Validates sparkConf and throws a SparkException if any of standard resources (memory or cores)
@@ -81,8 +120,9 @@ private object ResourceRequestHelper extends Logging {
     val errorMessage = new mutable.StringBuilder()
 
     resourceDefinitions.foreach { case (sparkName, resourceRequest) =>
-      if (sparkConf.contains(resourceRequest)) {
-        errorMessage.append(s"Error: Do not use $resourceRequest, " +
+      val resourceRequestAmount = s"${resourceRequest}.${AMOUNT}"
+      if (sparkConf.contains(resourceRequestAmount)) {
+        errorMessage.append(s"Error: Do not use $resourceRequestAmount, " +
             s"please use $sparkName instead!\n")
       }
     }

resource-managers/yarn/src/main/scala/org/apache/spark/deploy/yarn/YarnAllocator.scala

@@ -32,6 +32,7 @@ import org.apache.hadoop.yarn.client.api.AMRMClient.ContainerRequest
 import org.apache.hadoop.yarn.conf.YarnConfiguration
 
 import org.apache.spark.{SecurityManager, SparkConf, SparkException}
+import org.apache.spark.deploy.yarn.ResourceRequestHelper._
 import org.apache.spark.deploy.yarn.YarnSparkHadoopUtil._
 import org.apache.spark.deploy.yarn.config._
 import org.apache.spark.internal.Logging
@@ -142,8 +143,8 @@ private[yarn] class YarnAllocator(
   protected val executorCores = sparkConf.get(EXECUTOR_CORES)
 
   private val executorResourceRequests =
-    sparkConf.getAllWithPrefix(config.YARN_EXECUTOR_RESOURCE_TYPES_PREFIX).toMap ++
-      getYarnResourcesFromSparkResources(SPARK_EXECUTOR_PREFIX, sparkConf)
+    getYarnResourcesAndAmounts(sparkConf, config.YARN_EXECUTOR_RESOURCE_TYPES_PREFIX) ++
+    getYarnResourcesFromSparkResources(SPARK_EXECUTOR_PREFIX, sparkConf)
 
   // Resource capability requested for each executor
   private[yarn] val resource: Resource = {

resource-managers/yarn/src/main/scala/org/apache/spark/deploy/yarn/YarnSparkHadoopUtil.scala

@@ -41,24 +41,6 @@ object YarnSparkHadoopUtil {
   val MEMORY_OVERHEAD_MIN = 384L
 
   val ANY_HOST = "*"
-  val YARN_GPU_RESOURCE_CONFIG = "yarn.io/gpu"
-  val YARN_FPGA_RESOURCE_CONFIG = "yarn.io/fpga"
-
-  /**
-   * Convert Spark resources into YARN resources.
-   * The only resources we know how to map from spark configs to yarn configs are
-   * gpus and fpgas, everything else the user has to specify them in both the
-   * spark.yarn.*.resource and the spark.*.resource configs.
-   */
-  private[yarn] def getYarnResourcesFromSparkResources(
-      confPrefix: String,
-      sparkConf: SparkConf
-      ): Map[String, String] = {
-    Map(GPU -> YARN_GPU_RESOURCE_CONFIG, FPGA -> YARN_FPGA_RESOURCE_CONFIG).map {
-      case (rName, yarnName) =>
-        (yarnName -> sparkConf.get(ResourceID(confPrefix, rName).amountConf, "0"))
-    }.filter { case (_, count) => count.toLong > 0 }
-  }
 
   // All RM requests are issued with same priority : we do not (yet) have any distinction between
   // request types (like map/reduce in hadoop for example)

resource-managers/yarn/src/test/scala/org/apache/spark/deploy/yarn/ClientSuite.scala

@@ -38,10 +38,11 @@ import org.mockito.Mockito.{spy, verify}
 import org.scalatest.Matchers
 
 import org.apache.spark.{SparkConf, SparkException, SparkFunSuite, TestUtils}
-import org.apache.spark.deploy.yarn.YarnSparkHadoopUtil._
+import org.apache.spark.deploy.yarn.ResourceRequestHelper._
 import org.apache.spark.deploy.yarn.config._
 import org.apache.spark.internal.config._
 import org.apache.spark.resource.ResourceID
+import org.apache.spark.resource.ResourceUtils.AMOUNT
 import org.apache.spark.util.{SparkConfWithEnv, Utils}
 
 class ClientSuite extends SparkFunSuite with Matchers {
@@ -372,7 +373,7 @@ class ClientSuite extends SparkFunSuite with Matchers {
 
       val conf = new SparkConf().set(SUBMIT_DEPLOY_MODE, deployMode)
       resources.foreach { case (name, v) =>
-        conf.set(prefix + name, v.toString)
+        conf.set(s"${prefix}${name}.${AMOUNT}", v.toString)
       }
 
       val appContext = Records.newRecord(classOf[ApplicationSubmissionContext])
@@ -397,7 +398,7 @@ class ClientSuite extends SparkFunSuite with Matchers {
 
     val conf = new SparkConf().set(SUBMIT_DEPLOY_MODE, "cluster")
     resources.keys.foreach { yarnName =>
-      conf.set(s"${YARN_DRIVER_RESOURCE_TYPES_PREFIX}${yarnName}", "2")
+      conf.set(s"${YARN_DRIVER_RESOURCE_TYPES_PREFIX}${yarnName}.${AMOUNT}", "2")
     }
     resources.values.foreach { rName =>
       conf.set(ResourceID(SPARK_DRIVER_PREFIX, rName).amountConf, "3")
@@ -407,9 +408,9 @@ class ClientSuite extends SparkFunSuite with Matchers {
       ResourceRequestHelper.validateResources(conf)
     }.getMessage()
 
-    assert(error.contains("Do not use spark.yarn.driver.resource.yarn.io/fpga," +
+    assert(error.contains("Do not use spark.yarn.driver.resource.yarn.io/fpga.amount," +
       " please use spark.driver.resource.fpga.amount"))
-    assert(error.contains("Do not use spark.yarn.driver.resource.yarn.io/gpu," +
+    assert(error.contains("Do not use spark.yarn.driver.resource.yarn.io/gpu.amount," +
       " please use spark.driver.resource.gpu.amount"))
   }
 
@@ -420,7 +421,7 @@ class ClientSuite extends SparkFunSuite with Matchers {
 
     val conf = new SparkConf().set(SUBMIT_DEPLOY_MODE, "cluster")
     resources.keys.foreach { yarnName =>
-      conf.set(s"${YARN_EXECUTOR_RESOURCE_TYPES_PREFIX}${yarnName}", "2")
+      conf.set(s"${YARN_EXECUTOR_RESOURCE_TYPES_PREFIX}${yarnName}.${AMOUNT}", "2")
     }
     resources.values.foreach { rName =>
       conf.set(ResourceID(SPARK_EXECUTOR_PREFIX, rName).amountConf, "3")
@@ -430,9 +431,9 @@ class ClientSuite extends SparkFunSuite with Matchers {
       ResourceRequestHelper.validateResources(conf)
     }.getMessage()
 
-    assert(error.contains("Do not use spark.yarn.executor.resource.yarn.io/fpga," +
+    assert(error.contains("Do not use spark.yarn.executor.resource.yarn.io/fpga.amount," +
       " please use spark.executor.resource.fpga.amount"))
-    assert(error.contains("Do not use spark.yarn.executor.resource.yarn.io/gpu," +
+    assert(error.contains("Do not use spark.yarn.executor.resource.yarn.io/gpu.amount," +
       " please use spark.executor.resource.gpu.amount"))
   }
 
@@ -450,7 +451,7 @@ class ClientSuite extends SparkFunSuite with Matchers {
       conf.set(ResourceID(SPARK_DRIVER_PREFIX, rName).amountConf, "3")
     }
     // also just set yarn one that we don't convert
-    conf.set(YARN_DRIVER_RESOURCE_TYPES_PREFIX + yarnMadeupResource, "5")
+    conf.set(s"${YARN_DRIVER_RESOURCE_TYPES_PREFIX}${yarnMadeupResource}.${AMOUNT}", "5")
     val appContext = Records.newRecord(classOf[ApplicationSubmissionContext])
     val getNewApplicationResponse = Records.newRecord(classOf[GetNewApplicationResponse])
     val containerLaunchContext = Records.newRecord(classOf[ContainerLaunchContext])