[GitHub] spark pull request #20761: [SPARK-20327][CORE][YARN] Add CLI support for YAR...
Github user felixcheung commented on a diff in the pull request:
https://github.com/apache/spark/pull/20761#discussion_r227462444
--- Diff: docs/running-on-yarn.md ---
@@ -121,6 +121,43 @@ To use a custom metrics.properties for the application
master and executors, upd
Use lower-case suffixes, e.g. k, m,
g, t, and p, for kibi-, mebi-, gibi-,
tebi-, and pebibytes, respectively.
+
+ spark.yarn.am.resource.{resource-type}
+ (none)
+
+Amount of resource to use for the YARN Application Master in client
mode.
+In cluster mode, use
spark.yarn.driver.resource.resource-type instead.
--- End diff --
nit: looks like `spark.yarn.driver.resource.resource-type` should
be `spark.yarn.driver.resource.{resource-type}`
(yes, I realize `resource-type` is to be replaced with)
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #20761: [SPARK-20327][CORE][YARN] Add CLI support for YAR...
Github user asfgit closed the pull request at: https://github.com/apache/spark/pull/20761 --- - To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #20761: [SPARK-20327][CORE][YARN] Add CLI support for YAR...
Github user szyszy commented on a diff in the pull request:
https://github.com/apache/spark/pull/20761#discussion_r224923768
--- Diff:
resource-managers/yarn/src/main/scala/org/apache/spark/deploy/yarn/ResourceRequestHelper.scala
---
@@ -0,0 +1,140 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.deploy.yarn
+
+import java.lang.{Long => JLong}
+import java.lang.reflect.InvocationTargetException
+
+import scala.collection.mutable
+import scala.util.Try
+
+import org.apache.hadoop.yarn.api.records.Resource
+
+import org.apache.spark.{SparkConf, SparkException}
+import org.apache.spark.deploy.yarn.config._
+import org.apache.spark.internal.Logging
+import org.apache.spark.internal.config._
+import org.apache.spark.util.Utils
+
+/**
+ * This helper class uses some of Hadoop 3 methods from the YARN API,
+ * so we need to use reflection to avoid compile error when building
against Hadoop 2.x
+ */
+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"
+
+ /**
+ * Validates sparkConf and throws a SparkException if any of standard
resources (memory or cores)
+ * is defined with the property spark.yarn.x.resource.y
+ */
+ def validateResources(sparkConf: SparkConf): Unit = {
+val resourceDefinitions = Seq[(String, String)](
+ (AM_MEMORY.key, YARN_AM_RESOURCE_TYPES_PREFIX + "memory"),
--- End diff --
I think the code is now complete, please check!
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #20761: [SPARK-20327][CORE][YARN] Add CLI support for YAR...
Github user szyszy commented on a diff in the pull request:
https://github.com/apache/spark/pull/20761#discussion_r224914397
--- Diff:
resource-managers/yarn/src/main/scala/org/apache/spark/deploy/yarn/ResourceRequestHelper.scala
---
@@ -0,0 +1,140 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.deploy.yarn
+
+import java.lang.{Long => JLong}
+import java.lang.reflect.InvocationTargetException
+
+import scala.collection.mutable
+import scala.util.Try
+
+import org.apache.hadoop.yarn.api.records.Resource
+
+import org.apache.spark.{SparkConf, SparkException}
+import org.apache.spark.deploy.yarn.config._
+import org.apache.spark.internal.Logging
+import org.apache.spark.internal.config._
+import org.apache.spark.util.Utils
+
+/**
+ * This helper class uses some of Hadoop 3 methods from the YARN API,
+ * so we need to use reflection to avoid compile error when building
against Hadoop 2.x
+ */
+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"
+
+ /**
+ * Validates sparkConf and throws a SparkException if any of standard
resources (memory or cores)
+ * is defined with the property spark.yarn.x.resource.y
+ */
+ def validateResources(sparkConf: SparkConf): Unit = {
+val resourceDefinitions = Seq[(String, String)](
+ (AM_MEMORY.key, YARN_AM_RESOURCE_TYPES_PREFIX + "memory"),
--- End diff --
Sure, adding some explanatory comments with my next commit.
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #20761: [SPARK-20327][CORE][YARN] Add CLI support for YAR...
Github user szyszy commented on a diff in the pull request:
https://github.com/apache/spark/pull/20761#discussion_r224913985
--- Diff:
resource-managers/yarn/src/main/scala/org/apache/spark/deploy/yarn/ResourceRequestHelper.scala
---
@@ -0,0 +1,140 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.deploy.yarn
+
+import java.lang.{Long => JLong}
+import java.lang.reflect.InvocationTargetException
+
+import scala.collection.mutable
+import scala.util.Try
+
+import org.apache.hadoop.yarn.api.records.Resource
+
+import org.apache.spark.{SparkConf, SparkException}
+import org.apache.spark.deploy.yarn.config._
+import org.apache.spark.internal.Logging
+import org.apache.spark.internal.config._
+import org.apache.spark.util.Utils
+
+/**
+ * This helper class uses some of Hadoop 3 methods from the YARN API,
+ * so we need to use reflection to avoid compile error when building
against Hadoop 2.x
+ */
+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"
+
+ /**
+ * Validates sparkConf and throws a SparkException if any of standard
resources (memory or cores)
+ * is defined with the property spark.yarn.x.resource.y
+ */
+ def validateResources(sparkConf: SparkConf): Unit = {
+val resourceDefinitions = Seq[(String, String)](
+ (AM_MEMORY.key, YARN_AM_RESOURCE_TYPES_PREFIX + "memory"),
--- End diff --
Please see my last commit with the updates.
I only added some tests, so they are not extensive for every combination of
spark resources and YARN standard resources. If you think I can add more
testcases but I think this is fine as it is.
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #20761: [SPARK-20327][CORE][YARN] Add CLI support for YAR...
Github user vanzin commented on a diff in the pull request:
https://github.com/apache/spark/pull/20761#discussion_r224913824
--- Diff:
resource-managers/yarn/src/main/scala/org/apache/spark/deploy/yarn/ResourceRequestHelper.scala
---
@@ -0,0 +1,140 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.deploy.yarn
+
+import java.lang.{Long => JLong}
+import java.lang.reflect.InvocationTargetException
+
+import scala.collection.mutable
+import scala.util.Try
+
+import org.apache.hadoop.yarn.api.records.Resource
+
+import org.apache.spark.{SparkConf, SparkException}
+import org.apache.spark.deploy.yarn.config._
+import org.apache.spark.internal.Logging
+import org.apache.spark.internal.config._
+import org.apache.spark.util.Utils
+
+/**
+ * This helper class uses some of Hadoop 3 methods from the YARN API,
+ * so we need to use reflection to avoid compile error when building
against Hadoop 2.x
+ */
+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"
+
+ /**
+ * Validates sparkConf and throws a SparkException if any of standard
resources (memory or cores)
+ * is defined with the property spark.yarn.x.resource.y
+ */
+ def validateResources(sparkConf: SparkConf): Unit = {
+val resourceDefinitions = Seq[(String, String)](
+ (AM_MEMORY.key, YARN_AM_RESOURCE_TYPES_PREFIX + "memory"),
--- End diff --
> What did you mean with this?
I meant you were initializing `memory-mb` in tests but checking only
`memory` here. That smells like you should be checking `memory-mb` here.
There kinds of things should have comments in the code so in the future we
know why they are that way.
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #20761: [SPARK-20327][CORE][YARN] Add CLI support for YAR...
Github user szyszy commented on a diff in the pull request:
https://github.com/apache/spark/pull/20761#discussion_r224909836
--- Diff:
resource-managers/yarn/src/main/scala/org/apache/spark/deploy/yarn/ResourceRequestHelper.scala
---
@@ -0,0 +1,140 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.deploy.yarn
+
+import java.lang.{Long => JLong}
+import java.lang.reflect.InvocationTargetException
+
+import scala.collection.mutable
+import scala.util.Try
+
+import org.apache.hadoop.yarn.api.records.Resource
+
+import org.apache.spark.{SparkConf, SparkException}
+import org.apache.spark.deploy.yarn.config._
+import org.apache.spark.internal.Logging
+import org.apache.spark.internal.config._
+import org.apache.spark.util.Utils
+
+/**
+ * This helper class uses some of Hadoop 3 methods from the YARN API,
+ * so we need to use reflection to avoid compile error when building
against Hadoop 2.x
+ */
+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"
+
+ /**
+ * Validates sparkConf and throws a SparkException if any of standard
resources (memory or cores)
+ * is defined with the property spark.yarn.x.resource.y
+ */
+ def validateResources(sparkConf: SparkConf): Unit = {
+val resourceDefinitions = Seq[(String, String)](
+ (AM_MEMORY.key, YARN_AM_RESOURCE_TYPES_PREFIX + "memory"),
--- End diff --
These are two separate things:
1. One is that I don't reject all the deprecated standard resources has
been known to YARN (explained in previous comment) which I will address soon.
2. Using `memory-mb` is the only way to initialize the memory resource with
the YARN client, with the method `ResourceUtils.reinitializeResources`.
I played around with this a bit, if I omit the standard resources and try
to specify custom resources and then call
`ResourceUtils.reinitializeResources`, an internal YARN exception will be
thrown.
Unfortunately, invoking this method is the most simple way to build tests
upon custom resource types, to my best knowledge, so I can't really do much
about this.
> and also the inconsistency in your code (using both memory and memory-mb).
What did you mean with this? The only use of `"memory"` all around the
change is to prevent it from being used with the new resource configs.
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #20761: [SPARK-20327][CORE][YARN] Add CLI support for YAR...
Github user vanzin commented on a diff in the pull request:
https://github.com/apache/spark/pull/20761#discussion_r224851274
--- Diff:
resource-managers/yarn/src/main/scala/org/apache/spark/deploy/yarn/ResourceRequestHelper.scala
---
@@ -0,0 +1,140 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.deploy.yarn
+
+import java.lang.{Long => JLong}
+import java.lang.reflect.InvocationTargetException
+
+import scala.collection.mutable
+import scala.util.Try
+
+import org.apache.hadoop.yarn.api.records.Resource
+
+import org.apache.spark.{SparkConf, SparkException}
+import org.apache.spark.deploy.yarn.config._
+import org.apache.spark.internal.Logging
+import org.apache.spark.internal.config._
+import org.apache.spark.util.Utils
+
+/**
+ * This helper class uses some of Hadoop 3 methods from the YARN API,
+ * so we need to use reflection to avoid compile error when building
against Hadoop 2.x
+ */
+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"
+
+ /**
+ * Validates sparkConf and throws a SparkException if any of standard
resources (memory or cores)
+ * is defined with the property spark.yarn.x.resource.y
+ */
+ def validateResources(sparkConf: SparkConf): Unit = {
+val resourceDefinitions = Seq[(String, String)](
+ (AM_MEMORY.key, YARN_AM_RESOURCE_TYPES_PREFIX + "memory"),
--- End diff --
I'm not familiar with the YARN code or what it does here.
I'm just worried about users setting cpu/memory resources outside of the
proper Spark settings, and also the inconsistency in your code (using both
memory and memory-mb).
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #20761: [SPARK-20327][CORE][YARN] Add CLI support for YAR...
Github user szyszy commented on a diff in the pull request:
https://github.com/apache/spark/pull/20761#discussion_r224850514
--- Diff:
resource-managers/yarn/src/main/scala/org/apache/spark/deploy/yarn/ResourceRequestHelper.scala
---
@@ -0,0 +1,140 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.deploy.yarn
+
+import java.lang.{Long => JLong}
+import java.lang.reflect.InvocationTargetException
+
+import scala.collection.mutable
+import scala.util.Try
+
+import org.apache.hadoop.yarn.api.records.Resource
+
+import org.apache.spark.{SparkConf, SparkException}
+import org.apache.spark.deploy.yarn.config._
+import org.apache.spark.internal.Logging
+import org.apache.spark.internal.config._
+import org.apache.spark.util.Utils
+
+/**
+ * This helper class uses some of Hadoop 3 methods from the YARN API,
+ * so we need to use reflection to avoid compile error when building
against Hadoop 2.x
+ */
+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"
+
+ /**
+ * Validates sparkConf and throws a SparkException if any of standard
resources (memory or cores)
+ * is defined with the property spark.yarn.x.resource.y
+ */
+ def validateResources(sparkConf: SparkConf): Unit = {
+val resourceDefinitions = Seq[(String, String)](
+ (AM_MEMORY.key, YARN_AM_RESOURCE_TYPES_PREFIX + "memory"),
--- End diff --
Sure!
Did you mean this documentation?
https://hadoop.apache.org/docs/r3.0.1/hadoop-yarn/hadoop-yarn-site/ResourceModel.html
I think it's required to check all the keys for memory / vcore that YARN
deprecates, as those will flow trough Spark and eventually reach YARN's
`ResourceInformation` and it will just blow up as only `memory-mb` and `vcores`
are the ones that are not deprecated. The reason why it haven't caused a
problem with current Spark code as it is using the `Resource` object and not
using `ResourceInformation` at all.
So we need to disallow these:
- cpu-vcores
- memory
- mb
What do you think?
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #20761: [SPARK-20327][CORE][YARN] Add CLI support for YAR...
Github user vanzin commented on a diff in the pull request:
https://github.com/apache/spark/pull/20761#discussion_r224830769
--- Diff:
resource-managers/yarn/src/main/scala/org/apache/spark/deploy/yarn/ResourceRequestHelper.scala
---
@@ -0,0 +1,140 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.deploy.yarn
+
+import java.lang.{Long => JLong}
+import java.lang.reflect.InvocationTargetException
+
+import scala.collection.mutable
+import scala.util.Try
+
+import org.apache.hadoop.yarn.api.records.Resource
+
+import org.apache.spark.{SparkConf, SparkException}
+import org.apache.spark.deploy.yarn.config._
+import org.apache.spark.internal.Logging
+import org.apache.spark.internal.config._
+import org.apache.spark.util.Utils
+
+/**
+ * This helper class uses some of Hadoop 3 methods from the YARN API,
+ * so we need to use reflection to avoid compile error when building
against Hadoop 2.x
+ */
+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"
+
+ /**
+ * Validates sparkConf and throws a SparkException if any of standard
resources (memory or cores)
+ * is defined with the property spark.yarn.x.resource.y
+ */
+ def validateResources(sparkConf: SparkConf): Unit = {
+val resourceDefinitions = Seq[(String, String)](
+ (AM_MEMORY.key, YARN_AM_RESOURCE_TYPES_PREFIX + "memory"),
--- End diff --
Still waiting for a word on this.
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #20761: [SPARK-20327][CORE][YARN] Add CLI support for YAR...
Github user szyszy commented on a diff in the pull request:
https://github.com/apache/spark/pull/20761#discussion_r224595724
--- Diff:
resource-managers/yarn/src/test/scala/org/apache/spark/deploy/yarn/ResourceRequestHelperSuite.scala
---
@@ -0,0 +1,200 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.deploy.yarn
+
+import org.apache.hadoop.yarn.api.records.Resource
+import org.apache.hadoop.yarn.util.Records
+import org.scalatest.Matchers
+
+import org.apache.spark.{SparkConf, SparkException, SparkFunSuite}
+import
org.apache.spark.deploy.yarn.ResourceRequestTestHelper.ResourceInformation
+import org.apache.spark.deploy.yarn.config._
+import org.apache.spark.internal.config.{DRIVER_MEMORY, EXECUTOR_MEMORY}
+
+class ResourceRequestHelperSuite extends SparkFunSuite with Matchers {
+
+ private val CUSTOM_RES_1 = "custom-resource-type-1"
+ private val CUSTOM_RES_2 = "custom-resource-type-2"
+ private val MEMORY = "memory"
+ private val CORES = "cores"
+ private val NEW_CONFIG_EXECUTOR_MEMORY =
YARN_EXECUTOR_RESOURCE_TYPES_PREFIX + MEMORY
+ private val NEW_CONFIG_EXECUTOR_CORES =
YARN_EXECUTOR_RESOURCE_TYPES_PREFIX + CORES
+ private val NEW_CONFIG_AM_MEMORY = YARN_AM_RESOURCE_TYPES_PREFIX + MEMORY
+ private val NEW_CONFIG_AM_CORES = YARN_AM_RESOURCE_TYPES_PREFIX + CORES
+ private val NEW_CONFIG_DRIVER_MEMORY = YARN_DRIVER_RESOURCE_TYPES_PREFIX
+ MEMORY
+ private val NEW_CONFIG_DRIVER_CORES = YARN_DRIVER_RESOURCE_TYPES_PREFIX
+ CORES
+
+ test("resource request value does not match pattern") {
+verifySetResourceRequestsException(List(CUSTOM_RES_1),
+ Map(CUSTOM_RES_1 -> "**@#"), CUSTOM_RES_1)
+ }
+
+ test("resource request just unit defined") {
+verifySetResourceRequestsException(List(), Map(CUSTOM_RES_1 -> "m"),
CUSTOM_RES_1)
+ }
+
+ test("resource request with null value should not be allowed") {
+verifySetResourceRequestsException(List(), null, Map(CUSTOM_RES_1 ->
"123"),
+ "requirement failed: Resource parameter should not be null!")
+ }
+
+ test("resource request with valid value and invalid unit") {
+verifySetResourceRequestsException(List(CUSTOM_RES_1), createResource,
+ Map(CUSTOM_RES_1 -> "123ppp"), "")
+ }
+
+ test("resource request with valid value and without unit") {
+verifySetResourceRequestsSuccessful(List(CUSTOM_RES_1),
Map(CUSTOM_RES_1 -> "123"),
+ Map(CUSTOM_RES_1 -> ResourceInformation(CUSTOM_RES_1, 123, "")))
+ }
+
+ test("resource request with valid value and unit") {
+verifySetResourceRequestsSuccessful(List(CUSTOM_RES_1),
Map(CUSTOM_RES_1 -> "2g"),
+ Map(CUSTOM_RES_1 -> ResourceInformation(CUSTOM_RES_1, 2, "G")))
+ }
+
+ test("two resource requests with valid values and units") {
+verifySetResourceRequestsSuccessful(List(CUSTOM_RES_1, CUSTOM_RES_2),
+ Map(CUSTOM_RES_1 -> "123m", CUSTOM_RES_2 -> "10G"),
+ Map(CUSTOM_RES_1 -> ResourceInformation(CUSTOM_RES_1, 123, "m"),
+CUSTOM_RES_2 -> ResourceInformation(CUSTOM_RES_2, 10, "G")))
+ }
+
+ test("empty SparkConf should be valid") {
+val sparkConf = new SparkConf()
+ResourceRequestHelper.validateResources(sparkConf)
+ }
+
+ test("just normal resources are defined") {
+val sparkConf = new SparkConf()
+sparkConf.set(DRIVER_MEMORY.key, "3G")
+sparkConf.set(DRIVER_CORES.key, "4")
+sparkConf.set(EXECUTOR_MEMORY.key, "4G")
+sparkConf.set(EXECUTOR_CORES.key, "2")
+ResourceRequestHelper.validateResources(sparkConf)
+ }
+
+ test("memory defined with new config for executor") {
+val sparkConf = new SparkConf()
+sparkConf.set(NEW_CONFIG_EXECUTOR_MEMORY, "30G")
+verifyValidateResourcesException(sparkConf, NEW_CONFIG_EXECUTOR_MEMORY)
+ }
+
+ test("cores defined with new config for executor") {
+val sparkConf =
[GitHub] spark pull request #20761: [SPARK-20327][CORE][YARN] Add CLI support for YAR...
Github user szyszy commented on a diff in the pull request:
https://github.com/apache/spark/pull/20761#discussion_r224595178
--- Diff:
resource-managers/yarn/src/test/scala/org/apache/spark/deploy/yarn/ResourceRequestTestHelper.scala
---
@@ -0,0 +1,86 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.deploy.yarn
+
+import scala.collection.JavaConverters._
+import scala.collection.mutable.ListBuffer
+
+import org.apache.hadoop.yarn.api.records.Resource
+
+import org.apache.spark.util.Utils
+
+object ResourceRequestTestHelper {
+ def initializeResourceTypes(resourceTypes: List[String]): Unit = {
+if (!ResourceRequestHelper.isYarnResourceTypesAvailable()) {
+ throw new IllegalStateException("This method should not be invoked "
+
+"since YARN resource types is not available because of old Hadoop
version!" )
+}
+
+val allResourceTypes = new ListBuffer[AnyRef]
+val defaultResourceTypes = List(
+ createResourceTypeInfo("memory-mb"),
+ createResourceTypeInfo("vcores"))
+val customResourceTypes = resourceTypes.map(rt =>
createResourceTypeInfo(rt))
--- End diff --
fixed
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #20761: [SPARK-20327][CORE][YARN] Add CLI support for YAR...
Github user szyszy commented on a diff in the pull request:
https://github.com/apache/spark/pull/20761#discussion_r224594842
--- Diff:
resource-managers/yarn/src/test/scala/org/apache/spark/deploy/yarn/ResourceRequestHelperSuite.scala
---
@@ -0,0 +1,200 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.deploy.yarn
+
+import org.apache.hadoop.yarn.api.records.Resource
+import org.apache.hadoop.yarn.util.Records
+import org.scalatest.Matchers
+
+import org.apache.spark.{SparkConf, SparkException, SparkFunSuite}
+import
org.apache.spark.deploy.yarn.ResourceRequestTestHelper.ResourceInformation
+import org.apache.spark.deploy.yarn.config._
+import org.apache.spark.internal.config.{DRIVER_MEMORY, EXECUTOR_MEMORY}
+
+class ResourceRequestHelperSuite extends SparkFunSuite with Matchers {
+
+ private val CUSTOM_RES_1 = "custom-resource-type-1"
+ private val CUSTOM_RES_2 = "custom-resource-type-2"
+ private val MEMORY = "memory"
+ private val CORES = "cores"
+ private val NEW_CONFIG_EXECUTOR_MEMORY =
YARN_EXECUTOR_RESOURCE_TYPES_PREFIX + MEMORY
+ private val NEW_CONFIG_EXECUTOR_CORES =
YARN_EXECUTOR_RESOURCE_TYPES_PREFIX + CORES
+ private val NEW_CONFIG_AM_MEMORY = YARN_AM_RESOURCE_TYPES_PREFIX + MEMORY
+ private val NEW_CONFIG_AM_CORES = YARN_AM_RESOURCE_TYPES_PREFIX + CORES
+ private val NEW_CONFIG_DRIVER_MEMORY = YARN_DRIVER_RESOURCE_TYPES_PREFIX
+ MEMORY
+ private val NEW_CONFIG_DRIVER_CORES = YARN_DRIVER_RESOURCE_TYPES_PREFIX
+ CORES
+
+ test("resource request value does not match pattern") {
+verifySetResourceRequestsException(List(CUSTOM_RES_1),
+ Map(CUSTOM_RES_1 -> "**@#"), CUSTOM_RES_1)
+ }
+
+ test("resource request just unit defined") {
+verifySetResourceRequestsException(List(), Map(CUSTOM_RES_1 -> "m"),
CUSTOM_RES_1)
+ }
+
+ test("resource request with null value should not be allowed") {
+verifySetResourceRequestsException(List(), null, Map(CUSTOM_RES_1 ->
"123"),
+ "requirement failed: Resource parameter should not be null!")
+ }
+
+ test("resource request with valid value and invalid unit") {
+verifySetResourceRequestsException(List(CUSTOM_RES_1), createResource,
+ Map(CUSTOM_RES_1 -> "123ppp"), "")
+ }
+
+ test("resource request with valid value and without unit") {
+verifySetResourceRequestsSuccessful(List(CUSTOM_RES_1),
Map(CUSTOM_RES_1 -> "123"),
+ Map(CUSTOM_RES_1 -> ResourceInformation(CUSTOM_RES_1, 123, "")))
+ }
+
+ test("resource request with valid value and unit") {
+verifySetResourceRequestsSuccessful(List(CUSTOM_RES_1),
Map(CUSTOM_RES_1 -> "2g"),
+ Map(CUSTOM_RES_1 -> ResourceInformation(CUSTOM_RES_1, 2, "G")))
+ }
+
+ test("two resource requests with valid values and units") {
+verifySetResourceRequestsSuccessful(List(CUSTOM_RES_1, CUSTOM_RES_2),
+ Map(CUSTOM_RES_1 -> "123m", CUSTOM_RES_2 -> "10G"),
+ Map(CUSTOM_RES_1 -> ResourceInformation(CUSTOM_RES_1, 123, "m"),
+CUSTOM_RES_2 -> ResourceInformation(CUSTOM_RES_2, 10, "G")))
+ }
+
+ test("empty SparkConf should be valid") {
+val sparkConf = new SparkConf()
+ResourceRequestHelper.validateResources(sparkConf)
+ }
+
+ test("just normal resources are defined") {
+val sparkConf = new SparkConf()
+sparkConf.set(DRIVER_MEMORY.key, "3G")
+sparkConf.set(DRIVER_CORES.key, "4")
+sparkConf.set(EXECUTOR_MEMORY.key, "4G")
+sparkConf.set(EXECUTOR_CORES.key, "2")
+ResourceRequestHelper.validateResources(sparkConf)
+ }
+
+ test("memory defined with new config for executor") {
+val sparkConf = new SparkConf()
+sparkConf.set(NEW_CONFIG_EXECUTOR_MEMORY, "30G")
+verifyValidateResourcesException(sparkConf, NEW_CONFIG_EXECUTOR_MEMORY)
+ }
+
+ test("cores defined with new config for executor") {
+val sparkConf =
[GitHub] spark pull request #20761: [SPARK-20327][CORE][YARN] Add CLI support for YAR...
Github user szyszy commented on a diff in the pull request:
https://github.com/apache/spark/pull/20761#discussion_r224594296
--- Diff:
resource-managers/yarn/src/test/scala/org/apache/spark/deploy/yarn/ResourceRequestHelperSuite.scala
---
@@ -0,0 +1,200 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.deploy.yarn
+
+import org.apache.hadoop.yarn.api.records.Resource
+import org.apache.hadoop.yarn.util.Records
+import org.scalatest.Matchers
+
+import org.apache.spark.{SparkConf, SparkException, SparkFunSuite}
+import
org.apache.spark.deploy.yarn.ResourceRequestTestHelper.ResourceInformation
+import org.apache.spark.deploy.yarn.config._
+import org.apache.spark.internal.config.{DRIVER_MEMORY, EXECUTOR_MEMORY}
+
+class ResourceRequestHelperSuite extends SparkFunSuite with Matchers {
+
+ private val CUSTOM_RES_1 = "custom-resource-type-1"
+ private val CUSTOM_RES_2 = "custom-resource-type-2"
+ private val MEMORY = "memory"
+ private val CORES = "cores"
+ private val NEW_CONFIG_EXECUTOR_MEMORY =
YARN_EXECUTOR_RESOURCE_TYPES_PREFIX + MEMORY
+ private val NEW_CONFIG_EXECUTOR_CORES =
YARN_EXECUTOR_RESOURCE_TYPES_PREFIX + CORES
+ private val NEW_CONFIG_AM_MEMORY = YARN_AM_RESOURCE_TYPES_PREFIX + MEMORY
+ private val NEW_CONFIG_AM_CORES = YARN_AM_RESOURCE_TYPES_PREFIX + CORES
+ private val NEW_CONFIG_DRIVER_MEMORY = YARN_DRIVER_RESOURCE_TYPES_PREFIX
+ MEMORY
+ private val NEW_CONFIG_DRIVER_CORES = YARN_DRIVER_RESOURCE_TYPES_PREFIX
+ CORES
+
+ test("resource request value does not match pattern") {
+verifySetResourceRequestsException(List(CUSTOM_RES_1),
+ Map(CUSTOM_RES_1 -> "**@#"), CUSTOM_RES_1)
+ }
+
+ test("resource request just unit defined") {
+verifySetResourceRequestsException(List(), Map(CUSTOM_RES_1 -> "m"),
CUSTOM_RES_1)
+ }
+
+ test("resource request with null value should not be allowed") {
+verifySetResourceRequestsException(List(), null, Map(CUSTOM_RES_1 ->
"123"),
+ "requirement failed: Resource parameter should not be null!")
+ }
+
+ test("resource request with valid value and invalid unit") {
+verifySetResourceRequestsException(List(CUSTOM_RES_1), createResource,
+ Map(CUSTOM_RES_1 -> "123ppp"), "")
+ }
+
+ test("resource request with valid value and without unit") {
+verifySetResourceRequestsSuccessful(List(CUSTOM_RES_1),
Map(CUSTOM_RES_1 -> "123"),
+ Map(CUSTOM_RES_1 -> ResourceInformation(CUSTOM_RES_1, 123, "")))
+ }
+
+ test("resource request with valid value and unit") {
+verifySetResourceRequestsSuccessful(List(CUSTOM_RES_1),
Map(CUSTOM_RES_1 -> "2g"),
+ Map(CUSTOM_RES_1 -> ResourceInformation(CUSTOM_RES_1, 2, "G")))
+ }
+
+ test("two resource requests with valid values and units") {
+verifySetResourceRequestsSuccessful(List(CUSTOM_RES_1, CUSTOM_RES_2),
+ Map(CUSTOM_RES_1 -> "123m", CUSTOM_RES_2 -> "10G"),
+ Map(CUSTOM_RES_1 -> ResourceInformation(CUSTOM_RES_1, 123, "m"),
+CUSTOM_RES_2 -> ResourceInformation(CUSTOM_RES_2, 10, "G")))
+ }
+
+ test("empty SparkConf should be valid") {
+val sparkConf = new SparkConf()
+ResourceRequestHelper.validateResources(sparkConf)
+ }
+
+ test("just normal resources are defined") {
+val sparkConf = new SparkConf()
+sparkConf.set(DRIVER_MEMORY.key, "3G")
+sparkConf.set(DRIVER_CORES.key, "4")
+sparkConf.set(EXECUTOR_MEMORY.key, "4G")
+sparkConf.set(EXECUTOR_CORES.key, "2")
+ResourceRequestHelper.validateResources(sparkConf)
+ }
+
+ test("memory defined with new config for executor") {
+val sparkConf = new SparkConf()
+sparkConf.set(NEW_CONFIG_EXECUTOR_MEMORY, "30G")
+verifyValidateResourcesException(sparkConf, NEW_CONFIG_EXECUTOR_MEMORY)
+ }
+
+ test("cores defined with new config for executor") {
+val sparkConf =
[GitHub] spark pull request #20761: [SPARK-20327][CORE][YARN] Add CLI support for YAR...
Github user szyszy commented on a diff in the pull request:
https://github.com/apache/spark/pull/20761#discussion_r224594217
--- Diff:
resource-managers/yarn/src/test/scala/org/apache/spark/deploy/yarn/ResourceRequestHelperSuite.scala
---
@@ -0,0 +1,200 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.deploy.yarn
+
+import org.apache.hadoop.yarn.api.records.Resource
+import org.apache.hadoop.yarn.util.Records
+import org.scalatest.Matchers
+
+import org.apache.spark.{SparkConf, SparkException, SparkFunSuite}
+import
org.apache.spark.deploy.yarn.ResourceRequestTestHelper.ResourceInformation
+import org.apache.spark.deploy.yarn.config._
+import org.apache.spark.internal.config.{DRIVER_MEMORY, EXECUTOR_MEMORY}
+
+class ResourceRequestHelperSuite extends SparkFunSuite with Matchers {
+
+ private val CUSTOM_RES_1 = "custom-resource-type-1"
+ private val CUSTOM_RES_2 = "custom-resource-type-2"
+ private val MEMORY = "memory"
+ private val CORES = "cores"
+ private val NEW_CONFIG_EXECUTOR_MEMORY =
YARN_EXECUTOR_RESOURCE_TYPES_PREFIX + MEMORY
+ private val NEW_CONFIG_EXECUTOR_CORES =
YARN_EXECUTOR_RESOURCE_TYPES_PREFIX + CORES
+ private val NEW_CONFIG_AM_MEMORY = YARN_AM_RESOURCE_TYPES_PREFIX + MEMORY
+ private val NEW_CONFIG_AM_CORES = YARN_AM_RESOURCE_TYPES_PREFIX + CORES
+ private val NEW_CONFIG_DRIVER_MEMORY = YARN_DRIVER_RESOURCE_TYPES_PREFIX
+ MEMORY
+ private val NEW_CONFIG_DRIVER_CORES = YARN_DRIVER_RESOURCE_TYPES_PREFIX
+ CORES
+
+ test("resource request value does not match pattern") {
+verifySetResourceRequestsException(List(CUSTOM_RES_1),
+ Map(CUSTOM_RES_1 -> "**@#"), CUSTOM_RES_1)
+ }
+
+ test("resource request just unit defined") {
+verifySetResourceRequestsException(List(), Map(CUSTOM_RES_1 -> "m"),
CUSTOM_RES_1)
+ }
+
+ test("resource request with null value should not be allowed") {
+verifySetResourceRequestsException(List(), null, Map(CUSTOM_RES_1 ->
"123"),
+ "requirement failed: Resource parameter should not be null!")
+ }
+
+ test("resource request with valid value and invalid unit") {
+verifySetResourceRequestsException(List(CUSTOM_RES_1), createResource,
+ Map(CUSTOM_RES_1 -> "123ppp"), "")
+ }
+
+ test("resource request with valid value and without unit") {
+verifySetResourceRequestsSuccessful(List(CUSTOM_RES_1),
Map(CUSTOM_RES_1 -> "123"),
+ Map(CUSTOM_RES_1 -> ResourceInformation(CUSTOM_RES_1, 123, "")))
+ }
+
+ test("resource request with valid value and unit") {
+verifySetResourceRequestsSuccessful(List(CUSTOM_RES_1),
Map(CUSTOM_RES_1 -> "2g"),
+ Map(CUSTOM_RES_1 -> ResourceInformation(CUSTOM_RES_1, 2, "G")))
+ }
+
+ test("two resource requests with valid values and units") {
+verifySetResourceRequestsSuccessful(List(CUSTOM_RES_1, CUSTOM_RES_2),
+ Map(CUSTOM_RES_1 -> "123m", CUSTOM_RES_2 -> "10G"),
+ Map(CUSTOM_RES_1 -> ResourceInformation(CUSTOM_RES_1, 123, "m"),
+CUSTOM_RES_2 -> ResourceInformation(CUSTOM_RES_2, 10, "G")))
+ }
+
+ test("empty SparkConf should be valid") {
+val sparkConf = new SparkConf()
+ResourceRequestHelper.validateResources(sparkConf)
+ }
+
+ test("just normal resources are defined") {
+val sparkConf = new SparkConf()
+sparkConf.set(DRIVER_MEMORY.key, "3G")
+sparkConf.set(DRIVER_CORES.key, "4")
+sparkConf.set(EXECUTOR_MEMORY.key, "4G")
+sparkConf.set(EXECUTOR_CORES.key, "2")
+ResourceRequestHelper.validateResources(sparkConf)
+ }
+
+ test("memory defined with new config for executor") {
+val sparkConf = new SparkConf()
+sparkConf.set(NEW_CONFIG_EXECUTOR_MEMORY, "30G")
+verifyValidateResourcesException(sparkConf, NEW_CONFIG_EXECUTOR_MEMORY)
+ }
+
+ test("cores defined with new config for executor") {
+val sparkConf =
[GitHub] spark pull request #20761: [SPARK-20327][CORE][YARN] Add CLI support for YAR...
Github user szyszy commented on a diff in the pull request:
https://github.com/apache/spark/pull/20761#discussion_r224594094
--- Diff:
resource-managers/yarn/src/main/scala/org/apache/spark/deploy/yarn/ResourceRequestHelper.scala
---
@@ -0,0 +1,140 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.deploy.yarn
+
+import java.lang.{Long => JLong}
+import java.lang.reflect.InvocationTargetException
+
+import scala.collection.mutable
+import scala.util.Try
+
+import org.apache.hadoop.yarn.api.records.Resource
+
+import org.apache.spark.{SparkConf, SparkException}
+import org.apache.spark.deploy.yarn.config._
+import org.apache.spark.internal.Logging
+import org.apache.spark.internal.config._
+import org.apache.spark.util.Utils
+
+/**
+ * This helper class uses some of Hadoop 3 methods from the YARN API,
+ * so we need to use reflection to avoid compile error when building
against Hadoop 2.x
+ */
+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"
+
+ /**
+ * Validates sparkConf and throws a SparkException if any of standard
resources (memory or cores)
+ * is defined with the property spark.yarn.x.resource.y
+ */
+ def validateResources(sparkConf: SparkConf): Unit = {
+val resourceDefinitions = Seq[(String, String)](
+ (AM_MEMORY.key, YARN_AM_RESOURCE_TYPES_PREFIX + "memory"),
+ (AM_CORES.key, YARN_AM_RESOURCE_TYPES_PREFIX + "cores"),
+ (DRIVER_MEMORY.key, YARN_DRIVER_RESOURCE_TYPES_PREFIX + "memory"),
+ (DRIVER_CORES.key, YARN_DRIVER_RESOURCE_TYPES_PREFIX + "cores"),
+ (EXECUTOR_MEMORY.key, YARN_EXECUTOR_RESOURCE_TYPES_PREFIX +
"memory"),
+ (EXECUTOR_CORES.key, YARN_EXECUTOR_RESOURCE_TYPES_PREFIX + "cores"))
+val errorMessage = new mutable.StringBuilder()
+
+resourceDefinitions.foreach { case (sparkName, resourceRequest) =>
+ if (sparkConf.contains(resourceRequest)) {
+errorMessage.append(s"Error: Do not use $resourceRequest, " +
+s"please use $sparkName instead!\n")
+ }
+}
+
+if (errorMessage.nonEmpty) {
+ throw new SparkException(errorMessage.toString())
+}
+ }
+
+ /**
+ * Sets resource amount with the corresponding unit to the passed
resource object.
+ * @param resources resource values to set
+ * @param resource resource object to update
+ */
+ def setResourceRequests(
+ resources: Map[String, String],
+ resource: Resource): Unit = {
+require(resource != null, "Resource parameter should not be null!")
+
+logDebug(s"Custom resources requested: $resources")
+if (!isYarnResourceTypesAvailable()) {
+ if (resources.nonEmpty) {
+logWarning("Ignoring custom resource requests because " +
+"the version of YARN does not support it!")
+ }
+ return
+}
+
+val resInfoClass = Utils.classForName(RESOURCE_INFO_CLASS)
+val setResourceInformationMethod =
+ resource.getClass.getMethod("setResourceInformation",
classOf[String], resInfoClass)
+resources.foreach { case (name, rawAmount) =>
+ try {
+val AMOUNT_AND_UNIT_REGEX(amountPart, unitPart) = rawAmount
+val amount = amountPart.toLong
+val unit = unitPart match {
+ case "g" => "G"
+ case "t" => "T"
+ case "p" => "P"
+ case _ => unitPart
+}
+logDebug(s"Registering resource with name: $name, amount: $amount,
unit: $unit")
+val resourceInformation = createResourceInformation(
+ name, amount, unit, resInfoClass)
--- End diff --
fixed
---
-
To
[GitHub] spark pull request #20761: [SPARK-20327][CORE][YARN] Add CLI support for YAR...
Github user vanzin commented on a diff in the pull request:
https://github.com/apache/spark/pull/20761#discussion_r224272133
--- Diff:
resource-managers/yarn/src/test/scala/org/apache/spark/deploy/yarn/ResourceRequestTestHelper.scala
---
@@ -0,0 +1,86 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.deploy.yarn
+
+import scala.collection.JavaConverters._
+import scala.collection.mutable.ListBuffer
+
+import org.apache.hadoop.yarn.api.records.Resource
+
+import org.apache.spark.util.Utils
+
+object ResourceRequestTestHelper {
+ def initializeResourceTypes(resourceTypes: List[String]): Unit = {
+if (!ResourceRequestHelper.isYarnResourceTypesAvailable()) {
+ throw new IllegalStateException("This method should not be invoked "
+
+"since YARN resource types is not available because of old Hadoop
version!" )
+}
+
+val allResourceTypes = new ListBuffer[AnyRef]
+val defaultResourceTypes = List(
+ createResourceTypeInfo("memory-mb"),
+ createResourceTypeInfo("vcores"))
+val customResourceTypes = resourceTypes.map(rt =>
createResourceTypeInfo(rt))
--- End diff --
`.map { rt => ... }`, or `.map(createResourceTypeInfo)`
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #20761: [SPARK-20327][CORE][YARN] Add CLI support for YAR...
Github user vanzin commented on a diff in the pull request:
https://github.com/apache/spark/pull/20761#discussion_r224271908
--- Diff:
resource-managers/yarn/src/test/scala/org/apache/spark/deploy/yarn/ResourceRequestHelperSuite.scala
---
@@ -0,0 +1,200 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.deploy.yarn
+
+import org.apache.hadoop.yarn.api.records.Resource
+import org.apache.hadoop.yarn.util.Records
+import org.scalatest.Matchers
+
+import org.apache.spark.{SparkConf, SparkException, SparkFunSuite}
+import
org.apache.spark.deploy.yarn.ResourceRequestTestHelper.ResourceInformation
+import org.apache.spark.deploy.yarn.config._
+import org.apache.spark.internal.config.{DRIVER_MEMORY, EXECUTOR_MEMORY}
+
+class ResourceRequestHelperSuite extends SparkFunSuite with Matchers {
+
+ private val CUSTOM_RES_1 = "custom-resource-type-1"
+ private val CUSTOM_RES_2 = "custom-resource-type-2"
+ private val MEMORY = "memory"
+ private val CORES = "cores"
+ private val NEW_CONFIG_EXECUTOR_MEMORY =
YARN_EXECUTOR_RESOURCE_TYPES_PREFIX + MEMORY
+ private val NEW_CONFIG_EXECUTOR_CORES =
YARN_EXECUTOR_RESOURCE_TYPES_PREFIX + CORES
+ private val NEW_CONFIG_AM_MEMORY = YARN_AM_RESOURCE_TYPES_PREFIX + MEMORY
+ private val NEW_CONFIG_AM_CORES = YARN_AM_RESOURCE_TYPES_PREFIX + CORES
+ private val NEW_CONFIG_DRIVER_MEMORY = YARN_DRIVER_RESOURCE_TYPES_PREFIX
+ MEMORY
+ private val NEW_CONFIG_DRIVER_CORES = YARN_DRIVER_RESOURCE_TYPES_PREFIX
+ CORES
+
+ test("resource request value does not match pattern") {
+verifySetResourceRequestsException(List(CUSTOM_RES_1),
+ Map(CUSTOM_RES_1 -> "**@#"), CUSTOM_RES_1)
+ }
+
+ test("resource request just unit defined") {
+verifySetResourceRequestsException(List(), Map(CUSTOM_RES_1 -> "m"),
CUSTOM_RES_1)
+ }
+
+ test("resource request with null value should not be allowed") {
+verifySetResourceRequestsException(List(), null, Map(CUSTOM_RES_1 ->
"123"),
+ "requirement failed: Resource parameter should not be null!")
+ }
+
+ test("resource request with valid value and invalid unit") {
+verifySetResourceRequestsException(List(CUSTOM_RES_1), createResource,
+ Map(CUSTOM_RES_1 -> "123ppp"), "")
+ }
+
+ test("resource request with valid value and without unit") {
+verifySetResourceRequestsSuccessful(List(CUSTOM_RES_1),
Map(CUSTOM_RES_1 -> "123"),
+ Map(CUSTOM_RES_1 -> ResourceInformation(CUSTOM_RES_1, 123, "")))
+ }
+
+ test("resource request with valid value and unit") {
+verifySetResourceRequestsSuccessful(List(CUSTOM_RES_1),
Map(CUSTOM_RES_1 -> "2g"),
+ Map(CUSTOM_RES_1 -> ResourceInformation(CUSTOM_RES_1, 2, "G")))
+ }
+
+ test("two resource requests with valid values and units") {
+verifySetResourceRequestsSuccessful(List(CUSTOM_RES_1, CUSTOM_RES_2),
+ Map(CUSTOM_RES_1 -> "123m", CUSTOM_RES_2 -> "10G"),
+ Map(CUSTOM_RES_1 -> ResourceInformation(CUSTOM_RES_1, 123, "m"),
+CUSTOM_RES_2 -> ResourceInformation(CUSTOM_RES_2, 10, "G")))
+ }
+
+ test("empty SparkConf should be valid") {
+val sparkConf = new SparkConf()
+ResourceRequestHelper.validateResources(sparkConf)
+ }
+
+ test("just normal resources are defined") {
+val sparkConf = new SparkConf()
+sparkConf.set(DRIVER_MEMORY.key, "3G")
+sparkConf.set(DRIVER_CORES.key, "4")
+sparkConf.set(EXECUTOR_MEMORY.key, "4G")
+sparkConf.set(EXECUTOR_CORES.key, "2")
+ResourceRequestHelper.validateResources(sparkConf)
+ }
+
+ test("memory defined with new config for executor") {
+val sparkConf = new SparkConf()
+sparkConf.set(NEW_CONFIG_EXECUTOR_MEMORY, "30G")
+verifyValidateResourcesException(sparkConf, NEW_CONFIG_EXECUTOR_MEMORY)
+ }
+
+ test("cores defined with new config for executor") {
+val sparkConf =
[GitHub] spark pull request #20761: [SPARK-20327][CORE][YARN] Add CLI support for YAR...
Github user vanzin commented on a diff in the pull request:
https://github.com/apache/spark/pull/20761#discussion_r224270997
--- Diff:
resource-managers/yarn/src/main/scala/org/apache/spark/deploy/yarn/ResourceRequestHelper.scala
---
@@ -0,0 +1,140 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.deploy.yarn
+
+import java.lang.{Long => JLong}
+import java.lang.reflect.InvocationTargetException
+
+import scala.collection.mutable
+import scala.util.Try
+
+import org.apache.hadoop.yarn.api.records.Resource
+
+import org.apache.spark.{SparkConf, SparkException}
+import org.apache.spark.deploy.yarn.config._
+import org.apache.spark.internal.Logging
+import org.apache.spark.internal.config._
+import org.apache.spark.util.Utils
+
+/**
+ * This helper class uses some of Hadoop 3 methods from the YARN API,
+ * so we need to use reflection to avoid compile error when building
against Hadoop 2.x
+ */
+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"
+
+ /**
+ * Validates sparkConf and throws a SparkException if any of standard
resources (memory or cores)
+ * is defined with the property spark.yarn.x.resource.y
+ */
+ def validateResources(sparkConf: SparkConf): Unit = {
+val resourceDefinitions = Seq[(String, String)](
+ (AM_MEMORY.key, YARN_AM_RESOURCE_TYPES_PREFIX + "memory"),
+ (AM_CORES.key, YARN_AM_RESOURCE_TYPES_PREFIX + "cores"),
+ (DRIVER_MEMORY.key, YARN_DRIVER_RESOURCE_TYPES_PREFIX + "memory"),
+ (DRIVER_CORES.key, YARN_DRIVER_RESOURCE_TYPES_PREFIX + "cores"),
+ (EXECUTOR_MEMORY.key, YARN_EXECUTOR_RESOURCE_TYPES_PREFIX +
"memory"),
+ (EXECUTOR_CORES.key, YARN_EXECUTOR_RESOURCE_TYPES_PREFIX + "cores"))
+val errorMessage = new mutable.StringBuilder()
+
+resourceDefinitions.foreach { case (sparkName, resourceRequest) =>
+ if (sparkConf.contains(resourceRequest)) {
+errorMessage.append(s"Error: Do not use $resourceRequest, " +
+s"please use $sparkName instead!\n")
+ }
+}
+
+if (errorMessage.nonEmpty) {
+ throw new SparkException(errorMessage.toString())
+}
+ }
+
+ /**
+ * Sets resource amount with the corresponding unit to the passed
resource object.
+ * @param resources resource values to set
+ * @param resource resource object to update
+ */
+ def setResourceRequests(
+ resources: Map[String, String],
+ resource: Resource): Unit = {
+require(resource != null, "Resource parameter should not be null!")
+
+logDebug(s"Custom resources requested: $resources")
+if (!isYarnResourceTypesAvailable()) {
+ if (resources.nonEmpty) {
+logWarning("Ignoring custom resource requests because " +
+"the version of YARN does not support it!")
+ }
+ return
+}
+
+val resInfoClass = Utils.classForName(RESOURCE_INFO_CLASS)
+val setResourceInformationMethod =
+ resource.getClass.getMethod("setResourceInformation",
classOf[String], resInfoClass)
+resources.foreach { case (name, rawAmount) =>
+ try {
+val AMOUNT_AND_UNIT_REGEX(amountPart, unitPart) = rawAmount
+val amount = amountPart.toLong
+val unit = unitPart match {
+ case "g" => "G"
+ case "t" => "T"
+ case "p" => "P"
+ case _ => unitPart
+}
+logDebug(s"Registering resource with name: $name, amount: $amount,
unit: $unit")
+val resourceInformation = createResourceInformation(
+ name, amount, unit, resInfoClass)
--- End diff --
Fits in previous line.
---
[GitHub] spark pull request #20761: [SPARK-20327][CORE][YARN] Add CLI support for YAR...
Github user vanzin commented on a diff in the pull request:
https://github.com/apache/spark/pull/20761#discussion_r224271778
--- Diff:
resource-managers/yarn/src/test/scala/org/apache/spark/deploy/yarn/ResourceRequestHelperSuite.scala
---
@@ -0,0 +1,200 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.deploy.yarn
+
+import org.apache.hadoop.yarn.api.records.Resource
+import org.apache.hadoop.yarn.util.Records
+import org.scalatest.Matchers
+
+import org.apache.spark.{SparkConf, SparkException, SparkFunSuite}
+import
org.apache.spark.deploy.yarn.ResourceRequestTestHelper.ResourceInformation
+import org.apache.spark.deploy.yarn.config._
+import org.apache.spark.internal.config.{DRIVER_MEMORY, EXECUTOR_MEMORY}
+
+class ResourceRequestHelperSuite extends SparkFunSuite with Matchers {
+
+ private val CUSTOM_RES_1 = "custom-resource-type-1"
+ private val CUSTOM_RES_2 = "custom-resource-type-2"
+ private val MEMORY = "memory"
+ private val CORES = "cores"
+ private val NEW_CONFIG_EXECUTOR_MEMORY =
YARN_EXECUTOR_RESOURCE_TYPES_PREFIX + MEMORY
+ private val NEW_CONFIG_EXECUTOR_CORES =
YARN_EXECUTOR_RESOURCE_TYPES_PREFIX + CORES
+ private val NEW_CONFIG_AM_MEMORY = YARN_AM_RESOURCE_TYPES_PREFIX + MEMORY
+ private val NEW_CONFIG_AM_CORES = YARN_AM_RESOURCE_TYPES_PREFIX + CORES
+ private val NEW_CONFIG_DRIVER_MEMORY = YARN_DRIVER_RESOURCE_TYPES_PREFIX
+ MEMORY
+ private val NEW_CONFIG_DRIVER_CORES = YARN_DRIVER_RESOURCE_TYPES_PREFIX
+ CORES
+
+ test("resource request value does not match pattern") {
+verifySetResourceRequestsException(List(CUSTOM_RES_1),
+ Map(CUSTOM_RES_1 -> "**@#"), CUSTOM_RES_1)
+ }
+
+ test("resource request just unit defined") {
+verifySetResourceRequestsException(List(), Map(CUSTOM_RES_1 -> "m"),
CUSTOM_RES_1)
+ }
+
+ test("resource request with null value should not be allowed") {
+verifySetResourceRequestsException(List(), null, Map(CUSTOM_RES_1 ->
"123"),
+ "requirement failed: Resource parameter should not be null!")
+ }
+
+ test("resource request with valid value and invalid unit") {
+verifySetResourceRequestsException(List(CUSTOM_RES_1), createResource,
+ Map(CUSTOM_RES_1 -> "123ppp"), "")
+ }
+
+ test("resource request with valid value and without unit") {
+verifySetResourceRequestsSuccessful(List(CUSTOM_RES_1),
Map(CUSTOM_RES_1 -> "123"),
+ Map(CUSTOM_RES_1 -> ResourceInformation(CUSTOM_RES_1, 123, "")))
+ }
+
+ test("resource request with valid value and unit") {
+verifySetResourceRequestsSuccessful(List(CUSTOM_RES_1),
Map(CUSTOM_RES_1 -> "2g"),
+ Map(CUSTOM_RES_1 -> ResourceInformation(CUSTOM_RES_1, 2, "G")))
+ }
+
+ test("two resource requests with valid values and units") {
+verifySetResourceRequestsSuccessful(List(CUSTOM_RES_1, CUSTOM_RES_2),
+ Map(CUSTOM_RES_1 -> "123m", CUSTOM_RES_2 -> "10G"),
+ Map(CUSTOM_RES_1 -> ResourceInformation(CUSTOM_RES_1, 123, "m"),
+CUSTOM_RES_2 -> ResourceInformation(CUSTOM_RES_2, 10, "G")))
+ }
+
+ test("empty SparkConf should be valid") {
+val sparkConf = new SparkConf()
+ResourceRequestHelper.validateResources(sparkConf)
+ }
+
+ test("just normal resources are defined") {
+val sparkConf = new SparkConf()
+sparkConf.set(DRIVER_MEMORY.key, "3G")
+sparkConf.set(DRIVER_CORES.key, "4")
+sparkConf.set(EXECUTOR_MEMORY.key, "4G")
+sparkConf.set(EXECUTOR_CORES.key, "2")
+ResourceRequestHelper.validateResources(sparkConf)
+ }
+
+ test("memory defined with new config for executor") {
+val sparkConf = new SparkConf()
+sparkConf.set(NEW_CONFIG_EXECUTOR_MEMORY, "30G")
+verifyValidateResourcesException(sparkConf, NEW_CONFIG_EXECUTOR_MEMORY)
+ }
+
+ test("cores defined with new config for executor") {
+val sparkConf =
[GitHub] spark pull request #20761: [SPARK-20327][CORE][YARN] Add CLI support for YAR...
Github user vanzin commented on a diff in the pull request:
https://github.com/apache/spark/pull/20761#discussion_r224270816
--- Diff:
resource-managers/yarn/src/main/scala/org/apache/spark/deploy/yarn/ResourceRequestHelper.scala
---
@@ -0,0 +1,140 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.deploy.yarn
+
+import java.lang.{Long => JLong}
+import java.lang.reflect.InvocationTargetException
+
+import scala.collection.mutable
+import scala.util.Try
+
+import org.apache.hadoop.yarn.api.records.Resource
+
+import org.apache.spark.{SparkConf, SparkException}
+import org.apache.spark.deploy.yarn.config._
+import org.apache.spark.internal.Logging
+import org.apache.spark.internal.config._
+import org.apache.spark.util.Utils
+
+/**
+ * This helper class uses some of Hadoop 3 methods from the YARN API,
+ * so we need to use reflection to avoid compile error when building
against Hadoop 2.x
+ */
+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"
+
+ /**
+ * Validates sparkConf and throws a SparkException if any of standard
resources (memory or cores)
+ * is defined with the property spark.yarn.x.resource.y
+ */
+ def validateResources(sparkConf: SparkConf): Unit = {
+val resourceDefinitions = Seq[(String, String)](
+ (AM_MEMORY.key, YARN_AM_RESOURCE_TYPES_PREFIX + "memory"),
--- End diff --
I went and looked at the documentation because I remember this being
confusing. The documentation mentions both `memory` and `memory-mb` as being
valid, with the latter being preferred. So it sounds to me like you can use
either, and that this code should disallow both.
You even initialize `memory-mb` in your tests, instead of `memory`.
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #20761: [SPARK-20327][CORE][YARN] Add CLI support for YAR...
Github user vanzin commented on a diff in the pull request:
https://github.com/apache/spark/pull/20761#discussion_r224271821
--- Diff:
resource-managers/yarn/src/test/scala/org/apache/spark/deploy/yarn/ResourceRequestHelperSuite.scala
---
@@ -0,0 +1,200 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.deploy.yarn
+
+import org.apache.hadoop.yarn.api.records.Resource
+import org.apache.hadoop.yarn.util.Records
+import org.scalatest.Matchers
+
+import org.apache.spark.{SparkConf, SparkException, SparkFunSuite}
+import
org.apache.spark.deploy.yarn.ResourceRequestTestHelper.ResourceInformation
+import org.apache.spark.deploy.yarn.config._
+import org.apache.spark.internal.config.{DRIVER_MEMORY, EXECUTOR_MEMORY}
+
+class ResourceRequestHelperSuite extends SparkFunSuite with Matchers {
+
+ private val CUSTOM_RES_1 = "custom-resource-type-1"
+ private val CUSTOM_RES_2 = "custom-resource-type-2"
+ private val MEMORY = "memory"
+ private val CORES = "cores"
+ private val NEW_CONFIG_EXECUTOR_MEMORY =
YARN_EXECUTOR_RESOURCE_TYPES_PREFIX + MEMORY
+ private val NEW_CONFIG_EXECUTOR_CORES =
YARN_EXECUTOR_RESOURCE_TYPES_PREFIX + CORES
+ private val NEW_CONFIG_AM_MEMORY = YARN_AM_RESOURCE_TYPES_PREFIX + MEMORY
+ private val NEW_CONFIG_AM_CORES = YARN_AM_RESOURCE_TYPES_PREFIX + CORES
+ private val NEW_CONFIG_DRIVER_MEMORY = YARN_DRIVER_RESOURCE_TYPES_PREFIX
+ MEMORY
+ private val NEW_CONFIG_DRIVER_CORES = YARN_DRIVER_RESOURCE_TYPES_PREFIX
+ CORES
+
+ test("resource request value does not match pattern") {
+verifySetResourceRequestsException(List(CUSTOM_RES_1),
+ Map(CUSTOM_RES_1 -> "**@#"), CUSTOM_RES_1)
+ }
+
+ test("resource request just unit defined") {
+verifySetResourceRequestsException(List(), Map(CUSTOM_RES_1 -> "m"),
CUSTOM_RES_1)
+ }
+
+ test("resource request with null value should not be allowed") {
+verifySetResourceRequestsException(List(), null, Map(CUSTOM_RES_1 ->
"123"),
+ "requirement failed: Resource parameter should not be null!")
+ }
+
+ test("resource request with valid value and invalid unit") {
+verifySetResourceRequestsException(List(CUSTOM_RES_1), createResource,
+ Map(CUSTOM_RES_1 -> "123ppp"), "")
+ }
+
+ test("resource request with valid value and without unit") {
+verifySetResourceRequestsSuccessful(List(CUSTOM_RES_1),
Map(CUSTOM_RES_1 -> "123"),
+ Map(CUSTOM_RES_1 -> ResourceInformation(CUSTOM_RES_1, 123, "")))
+ }
+
+ test("resource request with valid value and unit") {
+verifySetResourceRequestsSuccessful(List(CUSTOM_RES_1),
Map(CUSTOM_RES_1 -> "2g"),
+ Map(CUSTOM_RES_1 -> ResourceInformation(CUSTOM_RES_1, 2, "G")))
+ }
+
+ test("two resource requests with valid values and units") {
+verifySetResourceRequestsSuccessful(List(CUSTOM_RES_1, CUSTOM_RES_2),
+ Map(CUSTOM_RES_1 -> "123m", CUSTOM_RES_2 -> "10G"),
+ Map(CUSTOM_RES_1 -> ResourceInformation(CUSTOM_RES_1, 123, "m"),
+CUSTOM_RES_2 -> ResourceInformation(CUSTOM_RES_2, 10, "G")))
+ }
+
+ test("empty SparkConf should be valid") {
+val sparkConf = new SparkConf()
+ResourceRequestHelper.validateResources(sparkConf)
+ }
+
+ test("just normal resources are defined") {
+val sparkConf = new SparkConf()
+sparkConf.set(DRIVER_MEMORY.key, "3G")
+sparkConf.set(DRIVER_CORES.key, "4")
+sparkConf.set(EXECUTOR_MEMORY.key, "4G")
+sparkConf.set(EXECUTOR_CORES.key, "2")
+ResourceRequestHelper.validateResources(sparkConf)
+ }
+
+ test("memory defined with new config for executor") {
+val sparkConf = new SparkConf()
+sparkConf.set(NEW_CONFIG_EXECUTOR_MEMORY, "30G")
+verifyValidateResourcesException(sparkConf, NEW_CONFIG_EXECUTOR_MEMORY)
+ }
+
+ test("cores defined with new config for executor") {
+val sparkConf =
[GitHub] spark pull request #20761: [SPARK-20327][CORE][YARN] Add CLI support for YAR...
Github user vanzin commented on a diff in the pull request:
https://github.com/apache/spark/pull/20761#discussion_r224271845
--- Diff:
resource-managers/yarn/src/test/scala/org/apache/spark/deploy/yarn/ResourceRequestHelperSuite.scala
---
@@ -0,0 +1,200 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.deploy.yarn
+
+import org.apache.hadoop.yarn.api.records.Resource
+import org.apache.hadoop.yarn.util.Records
+import org.scalatest.Matchers
+
+import org.apache.spark.{SparkConf, SparkException, SparkFunSuite}
+import
org.apache.spark.deploy.yarn.ResourceRequestTestHelper.ResourceInformation
+import org.apache.spark.deploy.yarn.config._
+import org.apache.spark.internal.config.{DRIVER_MEMORY, EXECUTOR_MEMORY}
+
+class ResourceRequestHelperSuite extends SparkFunSuite with Matchers {
+
+ private val CUSTOM_RES_1 = "custom-resource-type-1"
+ private val CUSTOM_RES_2 = "custom-resource-type-2"
+ private val MEMORY = "memory"
+ private val CORES = "cores"
+ private val NEW_CONFIG_EXECUTOR_MEMORY =
YARN_EXECUTOR_RESOURCE_TYPES_PREFIX + MEMORY
+ private val NEW_CONFIG_EXECUTOR_CORES =
YARN_EXECUTOR_RESOURCE_TYPES_PREFIX + CORES
+ private val NEW_CONFIG_AM_MEMORY = YARN_AM_RESOURCE_TYPES_PREFIX + MEMORY
+ private val NEW_CONFIG_AM_CORES = YARN_AM_RESOURCE_TYPES_PREFIX + CORES
+ private val NEW_CONFIG_DRIVER_MEMORY = YARN_DRIVER_RESOURCE_TYPES_PREFIX
+ MEMORY
+ private val NEW_CONFIG_DRIVER_CORES = YARN_DRIVER_RESOURCE_TYPES_PREFIX
+ CORES
+
+ test("resource request value does not match pattern") {
+verifySetResourceRequestsException(List(CUSTOM_RES_1),
+ Map(CUSTOM_RES_1 -> "**@#"), CUSTOM_RES_1)
+ }
+
+ test("resource request just unit defined") {
+verifySetResourceRequestsException(List(), Map(CUSTOM_RES_1 -> "m"),
CUSTOM_RES_1)
+ }
+
+ test("resource request with null value should not be allowed") {
+verifySetResourceRequestsException(List(), null, Map(CUSTOM_RES_1 ->
"123"),
+ "requirement failed: Resource parameter should not be null!")
+ }
+
+ test("resource request with valid value and invalid unit") {
+verifySetResourceRequestsException(List(CUSTOM_RES_1), createResource,
+ Map(CUSTOM_RES_1 -> "123ppp"), "")
+ }
+
+ test("resource request with valid value and without unit") {
+verifySetResourceRequestsSuccessful(List(CUSTOM_RES_1),
Map(CUSTOM_RES_1 -> "123"),
+ Map(CUSTOM_RES_1 -> ResourceInformation(CUSTOM_RES_1, 123, "")))
+ }
+
+ test("resource request with valid value and unit") {
+verifySetResourceRequestsSuccessful(List(CUSTOM_RES_1),
Map(CUSTOM_RES_1 -> "2g"),
+ Map(CUSTOM_RES_1 -> ResourceInformation(CUSTOM_RES_1, 2, "G")))
+ }
+
+ test("two resource requests with valid values and units") {
+verifySetResourceRequestsSuccessful(List(CUSTOM_RES_1, CUSTOM_RES_2),
+ Map(CUSTOM_RES_1 -> "123m", CUSTOM_RES_2 -> "10G"),
+ Map(CUSTOM_RES_1 -> ResourceInformation(CUSTOM_RES_1, 123, "m"),
+CUSTOM_RES_2 -> ResourceInformation(CUSTOM_RES_2, 10, "G")))
+ }
+
+ test("empty SparkConf should be valid") {
+val sparkConf = new SparkConf()
+ResourceRequestHelper.validateResources(sparkConf)
+ }
+
+ test("just normal resources are defined") {
+val sparkConf = new SparkConf()
+sparkConf.set(DRIVER_MEMORY.key, "3G")
+sparkConf.set(DRIVER_CORES.key, "4")
+sparkConf.set(EXECUTOR_MEMORY.key, "4G")
+sparkConf.set(EXECUTOR_CORES.key, "2")
+ResourceRequestHelper.validateResources(sparkConf)
+ }
+
+ test("memory defined with new config for executor") {
+val sparkConf = new SparkConf()
+sparkConf.set(NEW_CONFIG_EXECUTOR_MEMORY, "30G")
+verifyValidateResourcesException(sparkConf, NEW_CONFIG_EXECUTOR_MEMORY)
+ }
+
+ test("cores defined with new config for executor") {
+val sparkConf =
[GitHub] spark pull request #20761: [SPARK-20327][CORE][YARN] Add CLI support for YAR...
Github user szyszy commented on a diff in the pull request:
https://github.com/apache/spark/pull/20761#discussion_r224245968
--- Diff:
resource-managers/yarn/src/test/scala/org/apache/spark/deploy/yarn/ClientSuite.scala
---
@@ -433,4 +442,39 @@ class ClientSuite extends SparkFunSuite with Matchers {
classpath(env)
}
+ private def testResourceRequest(expectedResources: Seq[(String, Long)],
--- End diff --
Fair enough. I decided to pass in the sparkConf, the list of known
resources and the expected resources and their values, so the testcase is more
clear now.
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #20761: [SPARK-20327][CORE][YARN] Add CLI support for YAR...
Github user szyszy commented on a diff in the pull request:
https://github.com/apache/spark/pull/20761#discussion_r224243624
--- Diff:
resource-managers/yarn/src/test/scala/org/apache/spark/deploy/yarn/YarnAllocatorSuite.scala
---
@@ -95,6 +97,12 @@ class YarnAllocatorSuite extends SparkFunSuite with
Matchers with BeforeAndAfter
.set("spark.executor.instances", maxExecutors.toString)
.set("spark.executor.cores", "5")
.set("spark.executor.memory", "2048")
+
+// add additional configs from map
--- End diff --
removed
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #20761: [SPARK-20327][CORE][YARN] Add CLI support for YAR...
Github user szyszy commented on a diff in the pull request:
https://github.com/apache/spark/pull/20761#discussion_r224243499
--- Diff:
resource-managers/yarn/src/test/scala/org/apache/spark/deploy/yarn/ResourceRequestHelperSuite.scala
---
@@ -0,0 +1,207 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.deploy.yarn
+
+import org.apache.hadoop.yarn.api.records.Resource
+import org.apache.hadoop.yarn.util.Records
+import org.scalatest.Matchers
+
+import org.apache.spark.{SparkConf, SparkException, SparkFunSuite}
+import
org.apache.spark.deploy.yarn.ResourceRequestTestHelper.ResourceInformation
+import org.apache.spark.deploy.yarn.config._
+import org.apache.spark.internal.config.{DRIVER_MEMORY, EXECUTOR_MEMORY}
+
+class ResourceRequestHelperSuite extends SparkFunSuite with Matchers {
+
+ private val CUSTOM_RES_1 = "custom-resource-type-1"
+ private val CUSTOM_RES_2 = "custom-resource-type-2"
+ private val MEMORY = "memory"
+ private val CORES = "cores"
+ private val NEW_CONFIG_EXECUTOR_MEMORY =
YARN_EXECUTOR_RESOURCE_TYPES_PREFIX + MEMORY
+ private val NEW_CONFIG_EXECUTOR_CORES =
YARN_EXECUTOR_RESOURCE_TYPES_PREFIX + CORES
+ private val NEW_CONFIG_AM_MEMORY = YARN_AM_RESOURCE_TYPES_PREFIX + MEMORY
+ private val NEW_CONFIG_AM_CORES = YARN_AM_RESOURCE_TYPES_PREFIX + CORES
+ private val NEW_CONFIG_DRIVER_MEMORY = YARN_DRIVER_RESOURCE_TYPES_PREFIX
+ MEMORY
+ private val NEW_CONFIG_DRIVER_CORES = YARN_DRIVER_RESOURCE_TYPES_PREFIX
+ CORES
+
+ test("resource request value does not match pattern") {
+verifySetResourceRequestsException(List(CUSTOM_RES_1),
+ Map(CUSTOM_RES_1 -> "**@#"), CUSTOM_RES_1)
+ }
+
+ test("resource request just unit defined") {
+verifySetResourceRequestsException(List(), Map(CUSTOM_RES_1 -> "m"),
CUSTOM_RES_1)
+ }
+
+ test("resource request with null value should not be allowed") {
+verifySetResourceRequestsExceptionWithNullRequest(List(),
Map(CUSTOM_RES_1 -> "123"),
+ "requirement failed: Resource parameter should not be null!")
+ }
+
+ test("resource request with valid value and invalid unit") {
+verifySetResourceRequestsException(List(CUSTOM_RES_1),
Map(CUSTOM_RES_1 -> "123ppp"))
+ }
+
+ test("resource request with valid value and without unit") {
+verifySetResourceRequestsSuccessful(List(CUSTOM_RES_1),
Map(CUSTOM_RES_1 -> "123"),
+ Map(CUSTOM_RES_1 -> ResourceInformation(CUSTOM_RES_1, 123, "")))
+ }
+
+ test("resource request with valid value and unit") {
+verifySetResourceRequestsSuccessful(List(CUSTOM_RES_1),
Map(CUSTOM_RES_1 -> "2g"),
+ Map(CUSTOM_RES_1 -> ResourceInformation(CUSTOM_RES_1, 2, "G")))
+ }
+
+ test("two resource requests with valid values and units") {
+verifySetResourceRequestsSuccessful(List(CUSTOM_RES_1, CUSTOM_RES_2),
+ Map(CUSTOM_RES_1 -> "123m", CUSTOM_RES_2 -> "10G"),
+ Map(CUSTOM_RES_1 -> ResourceInformation(CUSTOM_RES_1, 123, "m"),
+CUSTOM_RES_2 -> ResourceInformation(CUSTOM_RES_2, 10, "G")))
+ }
+
+ test("empty SparkConf should be valid") {
+val sparkConf = new SparkConf()
+ResourceRequestHelper.validateResources(sparkConf)
+ }
+
+ test("just normal resources are defined") {
+val sparkConf = new SparkConf()
+sparkConf.set(DRIVER_MEMORY.key, "3G")
+sparkConf.set(DRIVER_CORES.key, "4")
+sparkConf.set(EXECUTOR_MEMORY.key, "4G")
+sparkConf.set(EXECUTOR_CORES.key, "2")
+ResourceRequestHelper.validateResources(sparkConf)
+ }
+
+ test("memory defined with new config for executor") {
+val sparkConf = new SparkConf()
+sparkConf.set(NEW_CONFIG_EXECUTOR_MEMORY, "30G")
+verifyValidateResourcesException(sparkConf, NEW_CONFIG_EXECUTOR_MEMORY)
+ }
+
+ test("cores defined with new config for executor") {
+val sparkConf = new SparkConf()
+
[GitHub] spark pull request #20761: [SPARK-20327][CORE][YARN] Add CLI support for YAR...
Github user szyszy commented on a diff in the pull request:
https://github.com/apache/spark/pull/20761#discussion_r224242115
--- Diff:
resource-managers/yarn/src/test/scala/org/apache/spark/deploy/yarn/ResourceRequestHelperSuite.scala
---
@@ -0,0 +1,207 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.deploy.yarn
+
+import org.apache.hadoop.yarn.api.records.Resource
+import org.apache.hadoop.yarn.util.Records
+import org.scalatest.Matchers
+
+import org.apache.spark.{SparkConf, SparkException, SparkFunSuite}
+import
org.apache.spark.deploy.yarn.ResourceRequestTestHelper.ResourceInformation
+import org.apache.spark.deploy.yarn.config._
+import org.apache.spark.internal.config.{DRIVER_MEMORY, EXECUTOR_MEMORY}
+
+class ResourceRequestHelperSuite extends SparkFunSuite with Matchers {
+
+ private val CUSTOM_RES_1 = "custom-resource-type-1"
+ private val CUSTOM_RES_2 = "custom-resource-type-2"
+ private val MEMORY = "memory"
+ private val CORES = "cores"
+ private val NEW_CONFIG_EXECUTOR_MEMORY =
YARN_EXECUTOR_RESOURCE_TYPES_PREFIX + MEMORY
+ private val NEW_CONFIG_EXECUTOR_CORES =
YARN_EXECUTOR_RESOURCE_TYPES_PREFIX + CORES
+ private val NEW_CONFIG_AM_MEMORY = YARN_AM_RESOURCE_TYPES_PREFIX + MEMORY
+ private val NEW_CONFIG_AM_CORES = YARN_AM_RESOURCE_TYPES_PREFIX + CORES
+ private val NEW_CONFIG_DRIVER_MEMORY = YARN_DRIVER_RESOURCE_TYPES_PREFIX
+ MEMORY
+ private val NEW_CONFIG_DRIVER_CORES = YARN_DRIVER_RESOURCE_TYPES_PREFIX
+ CORES
+
+ test("resource request value does not match pattern") {
+verifySetResourceRequestsException(List(CUSTOM_RES_1),
+ Map(CUSTOM_RES_1 -> "**@#"), CUSTOM_RES_1)
+ }
+
+ test("resource request just unit defined") {
+verifySetResourceRequestsException(List(), Map(CUSTOM_RES_1 -> "m"),
CUSTOM_RES_1)
+ }
+
+ test("resource request with null value should not be allowed") {
+verifySetResourceRequestsExceptionWithNullRequest(List(),
Map(CUSTOM_RES_1 -> "123"),
+ "requirement failed: Resource parameter should not be null!")
+ }
+
+ test("resource request with valid value and invalid unit") {
+verifySetResourceRequestsException(List(CUSTOM_RES_1),
Map(CUSTOM_RES_1 -> "123ppp"))
+ }
+
+ test("resource request with valid value and without unit") {
+verifySetResourceRequestsSuccessful(List(CUSTOM_RES_1),
Map(CUSTOM_RES_1 -> "123"),
+ Map(CUSTOM_RES_1 -> ResourceInformation(CUSTOM_RES_1, 123, "")))
+ }
+
+ test("resource request with valid value and unit") {
+verifySetResourceRequestsSuccessful(List(CUSTOM_RES_1),
Map(CUSTOM_RES_1 -> "2g"),
+ Map(CUSTOM_RES_1 -> ResourceInformation(CUSTOM_RES_1, 2, "G")))
+ }
+
+ test("two resource requests with valid values and units") {
+verifySetResourceRequestsSuccessful(List(CUSTOM_RES_1, CUSTOM_RES_2),
+ Map(CUSTOM_RES_1 -> "123m", CUSTOM_RES_2 -> "10G"),
+ Map(CUSTOM_RES_1 -> ResourceInformation(CUSTOM_RES_1, 123, "m"),
+CUSTOM_RES_2 -> ResourceInformation(CUSTOM_RES_2, 10, "G")))
+ }
+
+ test("empty SparkConf should be valid") {
+val sparkConf = new SparkConf()
+ResourceRequestHelper.validateResources(sparkConf)
+ }
+
+ test("just normal resources are defined") {
+val sparkConf = new SparkConf()
+sparkConf.set(DRIVER_MEMORY.key, "3G")
+sparkConf.set(DRIVER_CORES.key, "4")
+sparkConf.set(EXECUTOR_MEMORY.key, "4G")
+sparkConf.set(EXECUTOR_CORES.key, "2")
+ResourceRequestHelper.validateResources(sparkConf)
+ }
+
+ test("memory defined with new config for executor") {
+val sparkConf = new SparkConf()
+sparkConf.set(NEW_CONFIG_EXECUTOR_MEMORY, "30G")
+verifyValidateResourcesException(sparkConf, NEW_CONFIG_EXECUTOR_MEMORY)
+ }
+
+ test("cores defined with new config for executor") {
+val sparkConf = new SparkConf()
+
[GitHub] spark pull request #20761: [SPARK-20327][CORE][YARN] Add CLI support for YAR...
Github user szyszy commented on a diff in the pull request:
https://github.com/apache/spark/pull/20761#discussion_r224241554
--- Diff:
resource-managers/yarn/src/main/scala/org/apache/spark/deploy/yarn/ResourceRequestHelper.scala
---
@@ -0,0 +1,141 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.deploy.yarn
+
+import java.lang.{Integer => JInteger, Long => JLong}
+import java.lang.reflect.InvocationTargetException
+
+import scala.collection.mutable
+import scala.util.Try
+
+import org.apache.hadoop.yarn.api.records.Resource
+
+import org.apache.spark.{SparkConf, SparkException}
+import org.apache.spark.deploy.yarn.config.{AM_CORES, AM_MEMORY,
DRIVER_CORES, EXECUTOR_CORES, YARN_AM_RESOURCE_TYPES_PREFIX,
YARN_DRIVER_RESOURCE_TYPES_PREFIX, YARN_EXECUTOR_RESOURCE_TYPES_PREFIX}
--- End diff --
fixed
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #20761: [SPARK-20327][CORE][YARN] Add CLI support for YAR...
Github user szyszy commented on a diff in the pull request:
https://github.com/apache/spark/pull/20761#discussion_r224241720
--- Diff:
resource-managers/yarn/src/main/scala/org/apache/spark/deploy/yarn/ResourceRequestHelper.scala
---
@@ -0,0 +1,141 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.deploy.yarn
+
+import java.lang.{Integer => JInteger, Long => JLong}
+import java.lang.reflect.InvocationTargetException
+
+import scala.collection.mutable
+import scala.util.Try
+
+import org.apache.hadoop.yarn.api.records.Resource
+
+import org.apache.spark.{SparkConf, SparkException}
+import org.apache.spark.deploy.yarn.config.{AM_CORES, AM_MEMORY,
DRIVER_CORES, EXECUTOR_CORES, YARN_AM_RESOURCE_TYPES_PREFIX,
YARN_DRIVER_RESOURCE_TYPES_PREFIX, YARN_EXECUTOR_RESOURCE_TYPES_PREFIX}
+import org.apache.spark.internal.Logging
+import org.apache.spark.internal.config.{DRIVER_MEMORY, EXECUTOR_MEMORY}
+import org.apache.spark.util.Utils
+
+/**
+ * This helper class uses some of Hadoop 3 methods from the YARN API,
+ * so we need to use reflection to avoid compile error when building
against Hadoop 2.x
+ */
+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"
+ private val ERROR_PREFIX = "Error:"
--- End diff --
Fixed
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #20761: [SPARK-20327][CORE][YARN] Add CLI support for YAR...
Github user vanzin commented on a diff in the pull request:
https://github.com/apache/spark/pull/20761#discussion_r224230450
--- Diff:
resource-managers/yarn/src/test/scala/org/apache/spark/deploy/yarn/ClientSuite.scala
---
@@ -433,4 +442,39 @@ class ClientSuite extends SparkFunSuite with Matchers {
classpath(env)
}
+ private def testResourceRequest(expectedResources: Seq[(String, Long)],
--- End diff --
Repeating the comment since github doesn't re-open the original thread:
This isn't fixed yet. You're still passing in the expected resources, which
are expected to match the hardcoded configs that are set in this method
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #20761: [SPARK-20327][CORE][YARN] Add CLI support for YAR...
Github user vanzin commented on a diff in the pull request:
https://github.com/apache/spark/pull/20761#discussion_r224228487
--- Diff:
resource-managers/yarn/src/test/scala/org/apache/spark/deploy/yarn/ClientSuite.scala
---
@@ -433,4 +465,36 @@ class ClientSuite extends SparkFunSuite with Matchers {
classpath(env)
}
+ private def testResourceRequest(expectedResources: Seq[(String, Long)],
--- End diff --
This isn't fixed yet. You're still passing in the expected resources, which
are expected to match the hardcoded configs that are set in this method.
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #20761: [SPARK-20327][CORE][YARN] Add CLI support for YAR...
Github user vanzin commented on a diff in the pull request:
https://github.com/apache/spark/pull/20761#discussion_r224228882
--- Diff:
resource-managers/yarn/src/test/scala/org/apache/spark/deploy/yarn/ResourceRequestHelperSuite.scala
---
@@ -0,0 +1,207 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.deploy.yarn
+
+import org.apache.hadoop.yarn.api.records.Resource
+import org.apache.hadoop.yarn.util.Records
+import org.scalatest.Matchers
+
+import org.apache.spark.{SparkConf, SparkException, SparkFunSuite}
+import
org.apache.spark.deploy.yarn.ResourceRequestTestHelper.ResourceInformation
+import org.apache.spark.deploy.yarn.config._
+import org.apache.spark.internal.config.{DRIVER_MEMORY, EXECUTOR_MEMORY}
+
+class ResourceRequestHelperSuite extends SparkFunSuite with Matchers {
+
+ private val CUSTOM_RES_1 = "custom-resource-type-1"
+ private val CUSTOM_RES_2 = "custom-resource-type-2"
+ private val MEMORY = "memory"
+ private val CORES = "cores"
+ private val NEW_CONFIG_EXECUTOR_MEMORY =
YARN_EXECUTOR_RESOURCE_TYPES_PREFIX + MEMORY
+ private val NEW_CONFIG_EXECUTOR_CORES =
YARN_EXECUTOR_RESOURCE_TYPES_PREFIX + CORES
+ private val NEW_CONFIG_AM_MEMORY = YARN_AM_RESOURCE_TYPES_PREFIX + MEMORY
+ private val NEW_CONFIG_AM_CORES = YARN_AM_RESOURCE_TYPES_PREFIX + CORES
+ private val NEW_CONFIG_DRIVER_MEMORY = YARN_DRIVER_RESOURCE_TYPES_PREFIX
+ MEMORY
+ private val NEW_CONFIG_DRIVER_CORES = YARN_DRIVER_RESOURCE_TYPES_PREFIX
+ CORES
+
+ test("resource request value does not match pattern") {
+verifySetResourceRequestsException(List(CUSTOM_RES_1),
+ Map(CUSTOM_RES_1 -> "**@#"), CUSTOM_RES_1)
+ }
+
+ test("resource request just unit defined") {
+verifySetResourceRequestsException(List(), Map(CUSTOM_RES_1 -> "m"),
CUSTOM_RES_1)
+ }
+
+ test("resource request with null value should not be allowed") {
+verifySetResourceRequestsExceptionWithNullRequest(List(),
Map(CUSTOM_RES_1 -> "123"),
+ "requirement failed: Resource parameter should not be null!")
+ }
+
+ test("resource request with valid value and invalid unit") {
+verifySetResourceRequestsException(List(CUSTOM_RES_1),
Map(CUSTOM_RES_1 -> "123ppp"))
+ }
+
+ test("resource request with valid value and without unit") {
+verifySetResourceRequestsSuccessful(List(CUSTOM_RES_1),
Map(CUSTOM_RES_1 -> "123"),
+ Map(CUSTOM_RES_1 -> ResourceInformation(CUSTOM_RES_1, 123, "")))
+ }
+
+ test("resource request with valid value and unit") {
+verifySetResourceRequestsSuccessful(List(CUSTOM_RES_1),
Map(CUSTOM_RES_1 -> "2g"),
+ Map(CUSTOM_RES_1 -> ResourceInformation(CUSTOM_RES_1, 2, "G")))
+ }
+
+ test("two resource requests with valid values and units") {
+verifySetResourceRequestsSuccessful(List(CUSTOM_RES_1, CUSTOM_RES_2),
+ Map(CUSTOM_RES_1 -> "123m", CUSTOM_RES_2 -> "10G"),
+ Map(CUSTOM_RES_1 -> ResourceInformation(CUSTOM_RES_1, 123, "m"),
+CUSTOM_RES_2 -> ResourceInformation(CUSTOM_RES_2, 10, "G")))
+ }
+
+ test("empty SparkConf should be valid") {
+val sparkConf = new SparkConf()
+ResourceRequestHelper.validateResources(sparkConf)
+ }
+
+ test("just normal resources are defined") {
+val sparkConf = new SparkConf()
+sparkConf.set(DRIVER_MEMORY.key, "3G")
+sparkConf.set(DRIVER_CORES.key, "4")
+sparkConf.set(EXECUTOR_MEMORY.key, "4G")
+sparkConf.set(EXECUTOR_CORES.key, "2")
+ResourceRequestHelper.validateResources(sparkConf)
+ }
+
+ test("memory defined with new config for executor") {
+val sparkConf = new SparkConf()
+sparkConf.set(NEW_CONFIG_EXECUTOR_MEMORY, "30G")
+verifyValidateResourcesException(sparkConf, NEW_CONFIG_EXECUTOR_MEMORY)
+ }
+
+ test("cores defined with new config for executor") {
+val sparkConf = new SparkConf()
+
[GitHub] spark pull request #20761: [SPARK-20327][CORE][YARN] Add CLI support for YAR...
Github user vanzin commented on a diff in the pull request:
https://github.com/apache/spark/pull/20761#discussion_r224229718
--- Diff:
resource-managers/yarn/src/test/scala/org/apache/spark/deploy/yarn/YarnAllocatorSuite.scala
---
@@ -95,6 +97,12 @@ class YarnAllocatorSuite extends SparkFunSuite with
Matchers with BeforeAndAfter
.set("spark.executor.instances", maxExecutors.toString)
.set("spark.executor.cores", "5")
.set("spark.executor.memory", "2048")
+
+// add additional configs from map
--- End diff --
Unnecessary comment.
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #20761: [SPARK-20327][CORE][YARN] Add CLI support for YAR...
Github user vanzin commented on a diff in the pull request:
https://github.com/apache/spark/pull/20761#discussion_r224226970
--- Diff:
resource-managers/yarn/src/main/scala/org/apache/spark/deploy/yarn/ResourceRequestHelper.scala
---
@@ -0,0 +1,141 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.deploy.yarn
+
+import java.lang.{Integer => JInteger, Long => JLong}
+import java.lang.reflect.InvocationTargetException
+
+import scala.collection.mutable
+import scala.util.Try
+
+import org.apache.hadoop.yarn.api.records.Resource
+
+import org.apache.spark.{SparkConf, SparkException}
+import org.apache.spark.deploy.yarn.config.{AM_CORES, AM_MEMORY,
DRIVER_CORES, EXECUTOR_CORES, YARN_AM_RESOURCE_TYPES_PREFIX,
YARN_DRIVER_RESOURCE_TYPES_PREFIX, YARN_EXECUTOR_RESOURCE_TYPES_PREFIX}
+import org.apache.spark.internal.Logging
+import org.apache.spark.internal.config.{DRIVER_MEMORY, EXECUTOR_MEMORY}
+import org.apache.spark.util.Utils
+
+/**
+ * This helper class uses some of Hadoop 3 methods from the YARN API,
+ * so we need to use reflection to avoid compile error when building
against Hadoop 2.x
+ */
+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"
+ private val ERROR_PREFIX = "Error:"
--- End diff --
To be frank this is used in a single place so just put the string there...
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #20761: [SPARK-20327][CORE][YARN] Add CLI support for YAR...
Github user vanzin commented on a diff in the pull request:
https://github.com/apache/spark/pull/20761#discussion_r224229029
--- Diff:
resource-managers/yarn/src/test/scala/org/apache/spark/deploy/yarn/ResourceRequestHelperSuite.scala
---
@@ -0,0 +1,207 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.deploy.yarn
+
+import org.apache.hadoop.yarn.api.records.Resource
+import org.apache.hadoop.yarn.util.Records
+import org.scalatest.Matchers
+
+import org.apache.spark.{SparkConf, SparkException, SparkFunSuite}
+import
org.apache.spark.deploy.yarn.ResourceRequestTestHelper.ResourceInformation
+import org.apache.spark.deploy.yarn.config._
+import org.apache.spark.internal.config.{DRIVER_MEMORY, EXECUTOR_MEMORY}
+
+class ResourceRequestHelperSuite extends SparkFunSuite with Matchers {
+
+ private val CUSTOM_RES_1 = "custom-resource-type-1"
+ private val CUSTOM_RES_2 = "custom-resource-type-2"
+ private val MEMORY = "memory"
+ private val CORES = "cores"
+ private val NEW_CONFIG_EXECUTOR_MEMORY =
YARN_EXECUTOR_RESOURCE_TYPES_PREFIX + MEMORY
+ private val NEW_CONFIG_EXECUTOR_CORES =
YARN_EXECUTOR_RESOURCE_TYPES_PREFIX + CORES
+ private val NEW_CONFIG_AM_MEMORY = YARN_AM_RESOURCE_TYPES_PREFIX + MEMORY
+ private val NEW_CONFIG_AM_CORES = YARN_AM_RESOURCE_TYPES_PREFIX + CORES
+ private val NEW_CONFIG_DRIVER_MEMORY = YARN_DRIVER_RESOURCE_TYPES_PREFIX
+ MEMORY
+ private val NEW_CONFIG_DRIVER_CORES = YARN_DRIVER_RESOURCE_TYPES_PREFIX
+ CORES
+
+ test("resource request value does not match pattern") {
+verifySetResourceRequestsException(List(CUSTOM_RES_1),
+ Map(CUSTOM_RES_1 -> "**@#"), CUSTOM_RES_1)
+ }
+
+ test("resource request just unit defined") {
+verifySetResourceRequestsException(List(), Map(CUSTOM_RES_1 -> "m"),
CUSTOM_RES_1)
+ }
+
+ test("resource request with null value should not be allowed") {
+verifySetResourceRequestsExceptionWithNullRequest(List(),
Map(CUSTOM_RES_1 -> "123"),
+ "requirement failed: Resource parameter should not be null!")
+ }
+
+ test("resource request with valid value and invalid unit") {
+verifySetResourceRequestsException(List(CUSTOM_RES_1),
Map(CUSTOM_RES_1 -> "123ppp"))
+ }
+
+ test("resource request with valid value and without unit") {
+verifySetResourceRequestsSuccessful(List(CUSTOM_RES_1),
Map(CUSTOM_RES_1 -> "123"),
+ Map(CUSTOM_RES_1 -> ResourceInformation(CUSTOM_RES_1, 123, "")))
+ }
+
+ test("resource request with valid value and unit") {
+verifySetResourceRequestsSuccessful(List(CUSTOM_RES_1),
Map(CUSTOM_RES_1 -> "2g"),
+ Map(CUSTOM_RES_1 -> ResourceInformation(CUSTOM_RES_1, 2, "G")))
+ }
+
+ test("two resource requests with valid values and units") {
+verifySetResourceRequestsSuccessful(List(CUSTOM_RES_1, CUSTOM_RES_2),
+ Map(CUSTOM_RES_1 -> "123m", CUSTOM_RES_2 -> "10G"),
+ Map(CUSTOM_RES_1 -> ResourceInformation(CUSTOM_RES_1, 123, "m"),
+CUSTOM_RES_2 -> ResourceInformation(CUSTOM_RES_2, 10, "G")))
+ }
+
+ test("empty SparkConf should be valid") {
+val sparkConf = new SparkConf()
+ResourceRequestHelper.validateResources(sparkConf)
+ }
+
+ test("just normal resources are defined") {
+val sparkConf = new SparkConf()
+sparkConf.set(DRIVER_MEMORY.key, "3G")
+sparkConf.set(DRIVER_CORES.key, "4")
+sparkConf.set(EXECUTOR_MEMORY.key, "4G")
+sparkConf.set(EXECUTOR_CORES.key, "2")
+ResourceRequestHelper.validateResources(sparkConf)
+ }
+
+ test("memory defined with new config for executor") {
+val sparkConf = new SparkConf()
+sparkConf.set(NEW_CONFIG_EXECUTOR_MEMORY, "30G")
+verifyValidateResourcesException(sparkConf, NEW_CONFIG_EXECUTOR_MEMORY)
+ }
+
+ test("cores defined with new config for executor") {
+val sparkConf = new SparkConf()
+
[GitHub] spark pull request #20761: [SPARK-20327][CORE][YARN] Add CLI support for YAR...
Github user vanzin commented on a diff in the pull request:
https://github.com/apache/spark/pull/20761#discussion_r224226745
--- Diff:
resource-managers/yarn/src/main/scala/org/apache/spark/deploy/yarn/ResourceRequestHelper.scala
---
@@ -0,0 +1,141 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.deploy.yarn
+
+import java.lang.{Integer => JInteger, Long => JLong}
+import java.lang.reflect.InvocationTargetException
+
+import scala.collection.mutable
+import scala.util.Try
+
+import org.apache.hadoop.yarn.api.records.Resource
+
+import org.apache.spark.{SparkConf, SparkException}
+import org.apache.spark.deploy.yarn.config.{AM_CORES, AM_MEMORY,
DRIVER_CORES, EXECUTOR_CORES, YARN_AM_RESOURCE_TYPES_PREFIX,
YARN_DRIVER_RESOURCE_TYPES_PREFIX, YARN_EXECUTOR_RESOURCE_TYPES_PREFIX}
--- End diff --
Wildcard import.
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #20761: [SPARK-20327][CORE][YARN] Add CLI support for YAR...
Github user szyszy commented on a diff in the pull request:
https://github.com/apache/spark/pull/20761#discussion_r224195358
--- Diff:
resource-managers/yarn/src/main/scala/org/apache/spark/deploy/yarn/YarnAllocator.scala
---
@@ -288,9 +296,14 @@ private[yarn] class YarnAllocator(
s"executorsStarting: ${numExecutorsStarting.get}")
if (missing > 0) {
- logInfo(s"Will request $missing executor container(s), each with " +
-s"${resource.getVirtualCores} core(s) and " +
-s"${resource.getMemory} MB memory (including $memoryOverhead MB of
overhead)")
+ var requestContainerMessage = s"Will request $missing executor
container(s), each with " +
--- End diff --
Fair enough, added a condition to check whether it's enabled.
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #20761: [SPARK-20327][CORE][YARN] Add CLI support for YAR...
Github user szyszy commented on a diff in the pull request:
https://github.com/apache/spark/pull/20761#discussion_r224194497
--- Diff:
resource-managers/yarn/src/test/scala/org/apache/spark/deploy/yarn/ClientSuite.scala
---
@@ -199,6 +200,37 @@ class ClientSuite extends SparkFunSuite with Matchers {
appContext.getMaxAppAttempts should be (42)
}
+ test("resource request for invalid resource") {
--- End diff --
Okay, I agree with that. Then I think just removal is fine here.
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #20761: [SPARK-20327][CORE][YARN] Add CLI support for YAR...
Github user szyszy commented on a diff in the pull request:
https://github.com/apache/spark/pull/20761#discussion_r224193837
--- Diff:
resource-managers/yarn/src/test/scala/org/apache/spark/deploy/yarn/YarnAllocatorSuite.scala
---
@@ -87,6 +88,20 @@ class YarnAllocatorSuite extends SparkFunSuite with
Matchers with BeforeAndAfter
def createAllocator(
maxExecutors: Int = 5,
rmClient: AMRMClient[ContainerRequest] = rmClient): YarnAllocator = {
+createAllocatorInternal(maxExecutors, rmClient, Map())
+ }
+
+ def createAllocatorWithAdditionalConfigs(
--- End diff --
Fair enough, fixed.
Did similarly with the `createResource` method.
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #20761: [SPARK-20327][CORE][YARN] Add CLI support for YAR...
Github user szyszy commented on a diff in the pull request:
https://github.com/apache/spark/pull/20761#discussion_r224190144
--- Diff:
resource-managers/yarn/src/test/scala/org/apache/spark/deploy/yarn/ClientSuite.scala
---
@@ -433,4 +465,36 @@ class ClientSuite extends SparkFunSuite with Matchers {
classpath(env)
}
+ private def testResourceRequest(expectedResources: Seq[(String, Long)],
--- End diff --
Good point, fixed!
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #20761: [SPARK-20327][CORE][YARN] Add CLI support for YAR...
Github user vanzin commented on a diff in the pull request:
https://github.com/apache/spark/pull/20761#discussion_r224189326
--- Diff:
resource-managers/yarn/src/test/scala/org/apache/spark/deploy/yarn/ClientSuite.scala
---
@@ -199,6 +200,37 @@ class ClientSuite extends SparkFunSuite with Matchers {
appContext.getMaxAppAttempts should be (42)
}
+ test("resource request for invalid resource") {
--- End diff --
You already sort of do that in the tests that call
`initializeResourceTypes`, don't you?
Otherwise that's a YARN test, it's not interesting to test that
functionality in Spark because if it's broken, Spark cannot fix it.
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #20761: [SPARK-20327][CORE][YARN] Add CLI support for YAR...
Github user szyszy commented on a diff in the pull request:
https://github.com/apache/spark/pull/20761#discussion_r224187532
--- Diff:
resource-managers/yarn/src/test/scala/org/apache/spark/deploy/yarn/ClientSuite.scala
---
@@ -199,6 +200,37 @@ class ClientSuite extends SparkFunSuite with Matchers {
appContext.getMaxAppAttempts should be (42)
}
+ test("resource request for invalid resource") {
--- End diff --
Yes, this is very YARN-specific unlike the other testcases that verifies
the resources in the app context.
Removed this testcase.
Do you think of any other way to test the scenario when a resource is
specified in sparkConf which is not known for YARN?
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #20761: [SPARK-20327][CORE][YARN] Add CLI support for YAR...
Github user szyszy commented on a diff in the pull request:
https://github.com/apache/spark/pull/20761#discussion_r224185444
--- Diff:
resource-managers/yarn/src/main/scala/org/apache/spark/deploy/yarn/ResourceRequestHelper.scala
---
@@ -0,0 +1,148 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.deploy.yarn
+
+import java.lang.{Integer => JInteger, Long => JLong}
+import java.lang.reflect.InvocationTargetException
+
+import scala.collection.mutable
+import scala.util.Try
+
+import org.apache.hadoop.yarn.api.records.Resource
+
+import org.apache.spark.{SparkConf, SparkException}
+import org.apache.spark.deploy.yarn.config.{AM_CORES, AM_MEMORY,
DRIVER_CORES, EXECUTOR_CORES, YARN_AM_RESOURCE_TYPES_PREFIX,
YARN_DRIVER_RESOURCE_TYPES_PREFIX, YARN_EXECUTOR_RESOURCE_TYPES_PREFIX}
+import org.apache.spark.internal.Logging
+import org.apache.spark.internal.config.{DRIVER_MEMORY, EXECUTOR_MEMORY}
+import org.apache.spark.util.Utils
+
+/**
+ * This helper class uses some of Hadoop 3 methods from the YARN API,
+ * so we need to use reflection to avoid compile error when building
against Hadoop 2.x
+ */
+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"
+ private val ERROR_PREFIX: String = "Error:"
+
+ /**
+ * Validates sparkConf and throws a SparkException if any of standard
resources (memory or cores)
+ * is defined with the property spark.yarn.x.resource.y
+ * @param sparkConf
+ */
+ def validateResources(sparkConf: SparkConf): Unit = {
+val resourceDefinitions = Seq[(String, String)](
+ (AM_MEMORY.key, YARN_AM_RESOURCE_TYPES_PREFIX + "memory"),
+ (AM_CORES.key, YARN_AM_RESOURCE_TYPES_PREFIX + "cores"),
+ (DRIVER_MEMORY.key, YARN_DRIVER_RESOURCE_TYPES_PREFIX + "memory"),
+ (DRIVER_CORES.key, YARN_DRIVER_RESOURCE_TYPES_PREFIX + "cores"),
+ (EXECUTOR_MEMORY.key, YARN_EXECUTOR_RESOURCE_TYPES_PREFIX +
"memory"),
+ (EXECUTOR_CORES.key, YARN_EXECUTOR_RESOURCE_TYPES_PREFIX + "cores"))
+val errorMessage = new mutable.StringBuilder()
+
+resourceDefinitions.foreach { case (sparkName, resourceRequest) =>
+ if (sparkConf.contains(resourceRequest)) {
+errorMessage.append(s"$ERROR_PREFIX Do not use $resourceRequest, "
+
+s"please use $sparkName instead!\n")
+ }
+}
+
+if (errorMessage.nonEmpty) {
+ throw new SparkException(errorMessage.toString())
+}
+ }
+
+ /**
+ * Sets resource amount with the corresponding unit to the passed
resource object.
+ * @param resources resource values to set
+ * @param resource resource object to update
+ */
+ def setResourceRequests(
+ resources: Map[String, String],
+ resource: Resource): Unit = {
+require(resource != null, "Resource parameter should not be null!")
+require(!resources.contains("memory"),
--- End diff --
That's true, removed these require checks.
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #20761: [SPARK-20327][CORE][YARN] Add CLI support for YAR...
Github user szyszy commented on a diff in the pull request:
https://github.com/apache/spark/pull/20761#discussion_r224179095
--- Diff:
resource-managers/yarn/src/test/scala/org/apache/spark/deploy/yarn/ResourceRequestHelperSuite.scala
---
@@ -0,0 +1,255 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.deploy.yarn
+
+import org.apache.hadoop.yarn.api.records.Resource
+import org.apache.hadoop.yarn.util.Records
+import org.scalatest.{BeforeAndAfterAll, Matchers}
+
+import org.apache.spark.{SparkConf, SparkException, SparkFunSuite}
+import
org.apache.spark.deploy.yarn.ResourceRequestTestHelper.ResourceInformation
+import org.apache.spark.deploy.yarn.config.{AM_CORES, AM_MEMORY,
DRIVER_CORES, EXECUTOR_CORES, YARN_AM_RESOURCE_TYPES_PREFIX,
YARN_DRIVER_RESOURCE_TYPES_PREFIX, YARN_EXECUTOR_RESOURCE_TYPES_PREFIX}
+import org.apache.spark.internal.config.{DRIVER_MEMORY, EXECUTOR_MEMORY}
+
+class ResourceRequestHelperSuite extends SparkFunSuite with Matchers with
BeforeAndAfterAll {
+
+ private val CUSTOM_RES_1 = "custom-resource-type-1"
+ private val CUSTOM_RES_2 = "custom-resource-type-2"
+ private val MEMORY = "memory"
+ private val CORES = "cores"
+ private val NEW_CONFIG_EXECUTOR_MEMORY =
YARN_EXECUTOR_RESOURCE_TYPES_PREFIX + MEMORY
+ private val NEW_CONFIG_EXECUTOR_CORES =
YARN_EXECUTOR_RESOURCE_TYPES_PREFIX + CORES
+ private val NEW_CONFIG_AM_MEMORY = YARN_AM_RESOURCE_TYPES_PREFIX + MEMORY
+ private val NEW_CONFIG_AM_CORES = YARN_AM_RESOURCE_TYPES_PREFIX + CORES
+ private val NEW_CONFIG_DRIVER_MEMORY = YARN_DRIVER_RESOURCE_TYPES_PREFIX
+ MEMORY
+ private val NEW_CONFIG_DRIVER_CORES = YARN_DRIVER_RESOURCE_TYPES_PREFIX
+ CORES
+
+ override def beforeAll(): Unit = {
+super.beforeAll()
+ }
+
+ private def getExpectedUnmatchedErrorMessage(name: String, value:
String): String = {
+s"Resource request for '$name' ('$value') does not match pattern
([0-9]+)([A-Za-z]*)."
+ }
+
+ test("resource type value does not match pattern") {
+assume(ResourceRequestHelper.isYarnResourceTypesAvailable())
+ResourceRequestTestHelper.initializeResourceTypes(List(CUSTOM_RES_1))
+
+val resourceTypes = Map(CUSTOM_RES_1 -> "**@#")
+
+val thrown = intercept[IllegalArgumentException] {
+ ResourceRequestHelper.setResourceRequests(resourceTypes,
createAResource)
+}
+thrown.getMessage should equal
(getExpectedUnmatchedErrorMessage(CUSTOM_RES_1, "**@#"))
+ }
+
+ test("resource type just unit defined") {
+assume(ResourceRequestHelper.isYarnResourceTypesAvailable())
+ResourceRequestTestHelper.initializeResourceTypes(List())
+
+val resourceTypes = Map(CUSTOM_RES_1 -> "m")
+
+val thrown = intercept[IllegalArgumentException] {
+ ResourceRequestHelper.setResourceRequests(resourceTypes,
createAResource)
+}
+thrown.getMessage should equal
(getExpectedUnmatchedErrorMessage(CUSTOM_RES_1, "m"))
+ }
+
+ test("resource type with null value should not be allowed") {
+assume(ResourceRequestHelper.isYarnResourceTypesAvailable())
+ResourceRequestTestHelper.initializeResourceTypes(List())
+
+val resourceTypes = Map(CUSTOM_RES_1 -> "123")
+
+val thrown = intercept[IllegalArgumentException] {
+ ResourceRequestHelper.setResourceRequests(resourceTypes, null)
+}
+thrown.getMessage should equal ("requirement failed: Resource
parameter should not be null!")
+ }
+
+ test("resource type with valid value and invalid unit") {
+assume(ResourceRequestHelper.isYarnResourceTypesAvailable())
+ResourceRequestTestHelper.initializeResourceTypes(List(CUSTOM_RES_1))
+
+val resourceTypes = Map(CUSTOM_RES_1 -> "123ppp")
+val resource = createAResource
+
+val thrown = intercept[IllegalArgumentException] {
+ ResourceRequestHelper.setResourceRequests(resourceTypes,
[GitHub] spark pull request #20761: [SPARK-20327][CORE][YARN] Add CLI support for YAR...
Github user szyszy commented on a diff in the pull request:
https://github.com/apache/spark/pull/20761#discussion_r224178817
--- Diff:
resource-managers/yarn/src/test/scala/org/apache/spark/deploy/yarn/ResourceRequestHelperSuite.scala
---
@@ -0,0 +1,255 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.deploy.yarn
+
+import org.apache.hadoop.yarn.api.records.Resource
+import org.apache.hadoop.yarn.util.Records
+import org.scalatest.{BeforeAndAfterAll, Matchers}
+
+import org.apache.spark.{SparkConf, SparkException, SparkFunSuite}
+import
org.apache.spark.deploy.yarn.ResourceRequestTestHelper.ResourceInformation
+import org.apache.spark.deploy.yarn.config.{AM_CORES, AM_MEMORY,
DRIVER_CORES, EXECUTOR_CORES, YARN_AM_RESOURCE_TYPES_PREFIX,
YARN_DRIVER_RESOURCE_TYPES_PREFIX, YARN_EXECUTOR_RESOURCE_TYPES_PREFIX}
+import org.apache.spark.internal.config.{DRIVER_MEMORY, EXECUTOR_MEMORY}
+
+class ResourceRequestHelperSuite extends SparkFunSuite with Matchers with
BeforeAndAfterAll {
+
+ private val CUSTOM_RES_1 = "custom-resource-type-1"
+ private val CUSTOM_RES_2 = "custom-resource-type-2"
+ private val MEMORY = "memory"
+ private val CORES = "cores"
+ private val NEW_CONFIG_EXECUTOR_MEMORY =
YARN_EXECUTOR_RESOURCE_TYPES_PREFIX + MEMORY
+ private val NEW_CONFIG_EXECUTOR_CORES =
YARN_EXECUTOR_RESOURCE_TYPES_PREFIX + CORES
+ private val NEW_CONFIG_AM_MEMORY = YARN_AM_RESOURCE_TYPES_PREFIX + MEMORY
+ private val NEW_CONFIG_AM_CORES = YARN_AM_RESOURCE_TYPES_PREFIX + CORES
+ private val NEW_CONFIG_DRIVER_MEMORY = YARN_DRIVER_RESOURCE_TYPES_PREFIX
+ MEMORY
+ private val NEW_CONFIG_DRIVER_CORES = YARN_DRIVER_RESOURCE_TYPES_PREFIX
+ CORES
+
+ override def beforeAll(): Unit = {
+super.beforeAll()
+ }
+
+ private def getExpectedUnmatchedErrorMessage(name: String, value:
String): String = {
+s"Resource request for '$name' ('$value') does not match pattern
([0-9]+)([A-Za-z]*)."
+ }
+
+ test("resource type value does not match pattern") {
+assume(ResourceRequestHelper.isYarnResourceTypesAvailable())
+ResourceRequestTestHelper.initializeResourceTypes(List(CUSTOM_RES_1))
+
+val resourceTypes = Map(CUSTOM_RES_1 -> "**@#")
+
+val thrown = intercept[IllegalArgumentException] {
+ ResourceRequestHelper.setResourceRequests(resourceTypes,
createAResource)
+}
+thrown.getMessage should equal
(getExpectedUnmatchedErrorMessage(CUSTOM_RES_1, "**@#"))
+ }
+
+ test("resource type just unit defined") {
+assume(ResourceRequestHelper.isYarnResourceTypesAvailable())
+ResourceRequestTestHelper.initializeResourceTypes(List())
+
+val resourceTypes = Map(CUSTOM_RES_1 -> "m")
+
+val thrown = intercept[IllegalArgumentException] {
+ ResourceRequestHelper.setResourceRequests(resourceTypes,
createAResource)
+}
+thrown.getMessage should equal
(getExpectedUnmatchedErrorMessage(CUSTOM_RES_1, "m"))
+ }
+
+ test("resource type with null value should not be allowed") {
+assume(ResourceRequestHelper.isYarnResourceTypesAvailable())
+ResourceRequestTestHelper.initializeResourceTypes(List())
+
+val resourceTypes = Map(CUSTOM_RES_1 -> "123")
+
+val thrown = intercept[IllegalArgumentException] {
+ ResourceRequestHelper.setResourceRequests(resourceTypes, null)
+}
+thrown.getMessage should equal ("requirement failed: Resource
parameter should not be null!")
+ }
+
+ test("resource type with valid value and invalid unit") {
+assume(ResourceRequestHelper.isYarnResourceTypesAvailable())
+ResourceRequestTestHelper.initializeResourceTypes(List(CUSTOM_RES_1))
+
+val resourceTypes = Map(CUSTOM_RES_1 -> "123ppp")
+val resource = createAResource
+
+val thrown = intercept[IllegalArgumentException] {
+ ResourceRequestHelper.setResourceRequests(resourceTypes,
[GitHub] spark pull request #20761: [SPARK-20327][CORE][YARN] Add CLI support for YAR...
Github user szyszy commented on a diff in the pull request:
https://github.com/apache/spark/pull/20761#discussion_r224178252
--- Diff:
resource-managers/yarn/src/test/scala/org/apache/spark/deploy/yarn/ResourceRequestHelperSuite.scala
---
@@ -0,0 +1,255 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.deploy.yarn
+
+import org.apache.hadoop.yarn.api.records.Resource
+import org.apache.hadoop.yarn.util.Records
+import org.scalatest.{BeforeAndAfterAll, Matchers}
+
+import org.apache.spark.{SparkConf, SparkException, SparkFunSuite}
+import
org.apache.spark.deploy.yarn.ResourceRequestTestHelper.ResourceInformation
+import org.apache.spark.deploy.yarn.config.{AM_CORES, AM_MEMORY,
DRIVER_CORES, EXECUTOR_CORES, YARN_AM_RESOURCE_TYPES_PREFIX,
YARN_DRIVER_RESOURCE_TYPES_PREFIX, YARN_EXECUTOR_RESOURCE_TYPES_PREFIX}
+import org.apache.spark.internal.config.{DRIVER_MEMORY, EXECUTOR_MEMORY}
+
+class ResourceRequestHelperSuite extends SparkFunSuite with Matchers with
BeforeAndAfterAll {
+
+ private val CUSTOM_RES_1 = "custom-resource-type-1"
+ private val CUSTOM_RES_2 = "custom-resource-type-2"
+ private val MEMORY = "memory"
+ private val CORES = "cores"
+ private val NEW_CONFIG_EXECUTOR_MEMORY =
YARN_EXECUTOR_RESOURCE_TYPES_PREFIX + MEMORY
+ private val NEW_CONFIG_EXECUTOR_CORES =
YARN_EXECUTOR_RESOURCE_TYPES_PREFIX + CORES
+ private val NEW_CONFIG_AM_MEMORY = YARN_AM_RESOURCE_TYPES_PREFIX + MEMORY
+ private val NEW_CONFIG_AM_CORES = YARN_AM_RESOURCE_TYPES_PREFIX + CORES
+ private val NEW_CONFIG_DRIVER_MEMORY = YARN_DRIVER_RESOURCE_TYPES_PREFIX
+ MEMORY
+ private val NEW_CONFIG_DRIVER_CORES = YARN_DRIVER_RESOURCE_TYPES_PREFIX
+ CORES
+
+ override def beforeAll(): Unit = {
+super.beforeAll()
+ }
+
+ private def getExpectedUnmatchedErrorMessage(name: String, value:
String): String = {
+s"Resource request for '$name' ('$value') does not match pattern
([0-9]+)([A-Za-z]*)."
+ }
+
+ test("resource type value does not match pattern") {
+assume(ResourceRequestHelper.isYarnResourceTypesAvailable())
+ResourceRequestTestHelper.initializeResourceTypes(List(CUSTOM_RES_1))
+
+val resourceTypes = Map(CUSTOM_RES_1 -> "**@#")
+
+val thrown = intercept[IllegalArgumentException] {
+ ResourceRequestHelper.setResourceRequests(resourceTypes,
createAResource)
+}
+thrown.getMessage should equal
(getExpectedUnmatchedErrorMessage(CUSTOM_RES_1, "**@#"))
+ }
+
+ test("resource type just unit defined") {
+assume(ResourceRequestHelper.isYarnResourceTypesAvailable())
+ResourceRequestTestHelper.initializeResourceTypes(List())
+
+val resourceTypes = Map(CUSTOM_RES_1 -> "m")
+
+val thrown = intercept[IllegalArgumentException] {
+ ResourceRequestHelper.setResourceRequests(resourceTypes,
createAResource)
+}
+thrown.getMessage should equal
(getExpectedUnmatchedErrorMessage(CUSTOM_RES_1, "m"))
+ }
+
+ test("resource type with null value should not be allowed") {
+assume(ResourceRequestHelper.isYarnResourceTypesAvailable())
+ResourceRequestTestHelper.initializeResourceTypes(List())
+
+val resourceTypes = Map(CUSTOM_RES_1 -> "123")
+
+val thrown = intercept[IllegalArgumentException] {
+ ResourceRequestHelper.setResourceRequests(resourceTypes, null)
+}
+thrown.getMessage should equal ("requirement failed: Resource
parameter should not be null!")
+ }
+
+ test("resource type with valid value and invalid unit") {
+assume(ResourceRequestHelper.isYarnResourceTypesAvailable())
+ResourceRequestTestHelper.initializeResourceTypes(List(CUSTOM_RES_1))
+
+val resourceTypes = Map(CUSTOM_RES_1 -> "123ppp")
+val resource = createAResource
+
+val thrown = intercept[IllegalArgumentException] {
+ ResourceRequestHelper.setResourceRequests(resourceTypes,
[GitHub] spark pull request #20761: [SPARK-20327][CORE][YARN] Add CLI support for YAR...
Github user szyszy commented on a diff in the pull request:
https://github.com/apache/spark/pull/20761#discussion_r224177875
--- Diff:
resource-managers/yarn/src/test/scala/org/apache/spark/deploy/yarn/ResourceRequestHelperSuite.scala
---
@@ -0,0 +1,255 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.deploy.yarn
+
+import org.apache.hadoop.yarn.api.records.Resource
+import org.apache.hadoop.yarn.util.Records
+import org.scalatest.{BeforeAndAfterAll, Matchers}
+
+import org.apache.spark.{SparkConf, SparkException, SparkFunSuite}
+import
org.apache.spark.deploy.yarn.ResourceRequestTestHelper.ResourceInformation
+import org.apache.spark.deploy.yarn.config.{AM_CORES, AM_MEMORY,
DRIVER_CORES, EXECUTOR_CORES, YARN_AM_RESOURCE_TYPES_PREFIX,
YARN_DRIVER_RESOURCE_TYPES_PREFIX, YARN_EXECUTOR_RESOURCE_TYPES_PREFIX}
+import org.apache.spark.internal.config.{DRIVER_MEMORY, EXECUTOR_MEMORY}
+
+class ResourceRequestHelperSuite extends SparkFunSuite with Matchers with
BeforeAndAfterAll {
+
+ private val CUSTOM_RES_1 = "custom-resource-type-1"
+ private val CUSTOM_RES_2 = "custom-resource-type-2"
+ private val MEMORY = "memory"
+ private val CORES = "cores"
+ private val NEW_CONFIG_EXECUTOR_MEMORY =
YARN_EXECUTOR_RESOURCE_TYPES_PREFIX + MEMORY
+ private val NEW_CONFIG_EXECUTOR_CORES =
YARN_EXECUTOR_RESOURCE_TYPES_PREFIX + CORES
+ private val NEW_CONFIG_AM_MEMORY = YARN_AM_RESOURCE_TYPES_PREFIX + MEMORY
+ private val NEW_CONFIG_AM_CORES = YARN_AM_RESOURCE_TYPES_PREFIX + CORES
+ private val NEW_CONFIG_DRIVER_MEMORY = YARN_DRIVER_RESOURCE_TYPES_PREFIX
+ MEMORY
+ private val NEW_CONFIG_DRIVER_CORES = YARN_DRIVER_RESOURCE_TYPES_PREFIX
+ CORES
+
+ override def beforeAll(): Unit = {
+super.beforeAll()
+ }
+
+ private def getExpectedUnmatchedErrorMessage(name: String, value:
String): String = {
+s"Resource request for '$name' ('$value') does not match pattern
([0-9]+)([A-Za-z]*)."
+ }
+
+ test("resource type value does not match pattern") {
+assume(ResourceRequestHelper.isYarnResourceTypesAvailable())
+ResourceRequestTestHelper.initializeResourceTypes(List(CUSTOM_RES_1))
+
+val resourceTypes = Map(CUSTOM_RES_1 -> "**@#")
+
+val thrown = intercept[IllegalArgumentException] {
+ ResourceRequestHelper.setResourceRequests(resourceTypes,
createAResource)
+}
+thrown.getMessage should equal
(getExpectedUnmatchedErrorMessage(CUSTOM_RES_1, "**@#"))
+ }
+
+ test("resource type just unit defined") {
--- End diff --
Good idea. Extracted the common parts of all test methods in this class.
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #20761: [SPARK-20327][CORE][YARN] Add CLI support for YAR...
Github user szyszy commented on a diff in the pull request:
https://github.com/apache/spark/pull/20761#discussion_r223929329
--- Diff:
resource-managers/yarn/src/test/scala/org/apache/spark/deploy/yarn/ResourceRequestHelperSuite.scala
---
@@ -0,0 +1,255 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.deploy.yarn
+
+import org.apache.hadoop.yarn.api.records.Resource
+import org.apache.hadoop.yarn.util.Records
+import org.scalatest.{BeforeAndAfterAll, Matchers}
+
+import org.apache.spark.{SparkConf, SparkException, SparkFunSuite}
+import
org.apache.spark.deploy.yarn.ResourceRequestTestHelper.ResourceInformation
+import org.apache.spark.deploy.yarn.config.{AM_CORES, AM_MEMORY,
DRIVER_CORES, EXECUTOR_CORES, YARN_AM_RESOURCE_TYPES_PREFIX,
YARN_DRIVER_RESOURCE_TYPES_PREFIX, YARN_EXECUTOR_RESOURCE_TYPES_PREFIX}
+import org.apache.spark.internal.config.{DRIVER_MEMORY, EXECUTOR_MEMORY}
+
+class ResourceRequestHelperSuite extends SparkFunSuite with Matchers with
BeforeAndAfterAll {
+
+ private val CUSTOM_RES_1 = "custom-resource-type-1"
+ private val CUSTOM_RES_2 = "custom-resource-type-2"
+ private val MEMORY = "memory"
+ private val CORES = "cores"
+ private val NEW_CONFIG_EXECUTOR_MEMORY =
YARN_EXECUTOR_RESOURCE_TYPES_PREFIX + MEMORY
+ private val NEW_CONFIG_EXECUTOR_CORES =
YARN_EXECUTOR_RESOURCE_TYPES_PREFIX + CORES
+ private val NEW_CONFIG_AM_MEMORY = YARN_AM_RESOURCE_TYPES_PREFIX + MEMORY
+ private val NEW_CONFIG_AM_CORES = YARN_AM_RESOURCE_TYPES_PREFIX + CORES
+ private val NEW_CONFIG_DRIVER_MEMORY = YARN_DRIVER_RESOURCE_TYPES_PREFIX
+ MEMORY
+ private val NEW_CONFIG_DRIVER_CORES = YARN_DRIVER_RESOURCE_TYPES_PREFIX
+ CORES
+
+ override def beforeAll(): Unit = {
+super.beforeAll()
+ }
+
+ private def getExpectedUnmatchedErrorMessage(name: String, value:
String): String = {
--- End diff --
Removed the method. The testcases only check whether the resource name is
in the message from now.
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #20761: [SPARK-20327][CORE][YARN] Add CLI support for YAR...
Github user szyszy commented on a diff in the pull request:
https://github.com/apache/spark/pull/20761#discussion_r223928827
--- Diff:
resource-managers/yarn/src/test/scala/org/apache/spark/deploy/yarn/YarnAllocatorSuite.scala
---
@@ -134,6 +163,29 @@ class YarnAllocatorSuite extends SparkFunSuite with
Matchers with BeforeAndAfter
size should be (0)
}
+ test("custom resource requested from yarn") {
+assume(ResourceRequestHelper.isYarnResourceTypesAvailable())
+ResourceRequestTestHelper.initializeResourceTypes(List("gpu"))
+
+val mockAmClient = mock(classOf[AMRMClient[ContainerRequest]])
+val handler = createAllocatorWithAdditionalConfigs(1, Map(
+ YARN_EXECUTOR_RESOURCE_TYPES_PREFIX + "gpu" -> "2G"), mockAmClient)
+
+handler.updateResourceRequests()
+val container = createContainerWithResource("host1", handler.resource)
+handler.handleAllocatedContainers(Array(container))
+
+// get amount of memory and vcores from resource, so effectively
skipping their validation
+val expectedResources =
Resource.newInstance(handler.resource.getMemory(),
+ handler.resource.getVirtualCores)
+ResourceRequestHelper.setResourceRequests(Map("gpu" -> "2G"),
expectedResources)
+val captor = ArgumentCaptor.forClass(classOf[ContainerRequest])
+
+verify(mockAmClient).addContainerRequest(captor.capture())
+val containerRequest: ContainerRequest = captor.getValue
+assert(containerRequest.getCapability == expectedResources)
--- End diff --
fixed
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #20761: [SPARK-20327][CORE][YARN] Add CLI support for YAR...
Github user vanzin commented on a diff in the pull request:
https://github.com/apache/spark/pull/20761#discussion_r223919358
--- Diff:
resource-managers/yarn/src/main/scala/org/apache/spark/deploy/yarn/YarnAllocator.scala
---
@@ -288,9 +296,14 @@ private[yarn] class YarnAllocator(
s"executorsStarting: ${numExecutorsStarting.get}")
if (missing > 0) {
- logInfo(s"Will request $missing executor container(s), each with " +
-s"${resource.getVirtualCores} core(s) and " +
-s"${resource.getMemory} MB memory (including $memoryOverhead MB of
overhead)")
+ var requestContainerMessage = s"Will request $missing executor
container(s), each with " +
--- End diff --
But here you're doing a bunch of things regardless of whether info logs are
enabled.
You should check if they're enabled so you don't have to spend time doing
stuff that will just be thrown away.
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #20761: [SPARK-20327][CORE][YARN] Add CLI support for YAR...
Github user szyszy commented on a diff in the pull request:
https://github.com/apache/spark/pull/20761#discussion_r223918822
--- Diff:
resource-managers/yarn/src/main/scala/org/apache/spark/deploy/yarn/YarnAllocator.scala
---
@@ -288,9 +296,14 @@ private[yarn] class YarnAllocator(
s"executorsStarting: ${numExecutorsStarting.get}")
if (missing > 0) {
- logInfo(s"Will request $missing executor container(s), each with " +
-s"${resource.getVirtualCores} core(s) and " +
-s"${resource.getMemory} MB memory (including $memoryOverhead MB of
overhead)")
+ var requestContainerMessage = s"Will request $missing executor
container(s), each with " +
--- End diff --
`lofInfo()` already calls `if (log.isInfoEnabled())`:
`protected def logInfo(msg: => String) {
if (log.isInfoEnabled) log.info(msg)
}`
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #20761: [SPARK-20327][CORE][YARN] Add CLI support for YAR...
Github user szyszy commented on a diff in the pull request:
https://github.com/apache/spark/pull/20761#discussion_r223918256
--- Diff:
resource-managers/yarn/src/test/scala/org/apache/spark/deploy/yarn/ResourceRequestTestHelper.scala
---
@@ -0,0 +1,90 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.deploy.yarn
+
+import scala.collection.JavaConverters._
+import scala.collection.mutable.ListBuffer
+
+import org.apache.hadoop.yarn.api.records.Resource
+
+import org.apache.spark.util.Utils
+
+object ResourceRequestTestHelper {
+ def initializeResourceTypes(resourceTypes: List[String]): Unit = {
+if (!ResourceRequestHelper.isYarnResourceTypesAvailable()) {
+ throw new IllegalStateException("initializeResourceTypes() should
not be invoked " +
+"since YARN resource types is not available because of old Hadoop
version!" )
+}
+
+val allResourceTypes = new ListBuffer[AnyRef]
+val defaultResourceTypes = List(
+ createResourceTypeInfo("memory-mb"),
+ createResourceTypeInfo("vcores"))
+val customResourceTypes = resourceTypes.map(rt =>
createResourceTypeInfo(rt))
+allResourceTypes ++= defaultResourceTypes
+allResourceTypes ++= customResourceTypes
+
+reinitializeResources(allResourceTypes)
+ }
+
+ private def createResourceTypeInfo(resourceName: String): AnyRef = {
+val resTypeInfoClass =
Utils.classForName("org.apache.hadoop.yarn.api.records.ResourceTypeInfo")
+val resTypeInfoNewInstanceMethod =
resTypeInfoClass.getMethod("newInstance", classOf[String])
+resTypeInfoNewInstanceMethod.invoke(null, resourceName)
+ }
+
+ private def reinitializeResources(resourceTypes: ListBuffer[AnyRef]):
Unit = {
+val resourceUtilsClass =
+
Utils.classForName("org.apache.hadoop.yarn.util.resource.ResourceUtils")
+val reinitializeResourcesMethod =
resourceUtilsClass.getMethod("reinitializeResources",
+ classOf[java.util.List[AnyRef]])
+reinitializeResourcesMethod.invoke(null, resourceTypes.asJava)
+ }
+
+ def getResourceTypeValue(res: Resource, name: String): AnyRef = {
+val resourceInformation: AnyRef = getResourceInformation(res, name)
--- End diff --
removed type
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #20761: [SPARK-20327][CORE][YARN] Add CLI support for YAR...
Github user szyszy commented on a diff in the pull request:
https://github.com/apache/spark/pull/20761#discussion_r223918024
--- Diff:
resource-managers/yarn/src/test/scala/org/apache/spark/deploy/yarn/ResourceRequestTestHelper.scala
---
@@ -0,0 +1,90 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.deploy.yarn
+
+import scala.collection.JavaConverters._
+import scala.collection.mutable.ListBuffer
+
+import org.apache.hadoop.yarn.api.records.Resource
+
+import org.apache.spark.util.Utils
+
+object ResourceRequestTestHelper {
+ def initializeResourceTypes(resourceTypes: List[String]): Unit = {
+if (!ResourceRequestHelper.isYarnResourceTypesAvailable()) {
+ throw new IllegalStateException("initializeResourceTypes() should
not be invoked " +
+"since YARN resource types is not available because of old Hadoop
version!" )
+}
+
+val allResourceTypes = new ListBuffer[AnyRef]
+val defaultResourceTypes = List(
+ createResourceTypeInfo("memory-mb"),
+ createResourceTypeInfo("vcores"))
+val customResourceTypes = resourceTypes.map(rt =>
createResourceTypeInfo(rt))
+allResourceTypes ++= defaultResourceTypes
+allResourceTypes ++= customResourceTypes
+
+reinitializeResources(allResourceTypes)
+ }
+
+ private def createResourceTypeInfo(resourceName: String): AnyRef = {
+val resTypeInfoClass =
Utils.classForName("org.apache.hadoop.yarn.api.records.ResourceTypeInfo")
+val resTypeInfoNewInstanceMethod =
resTypeInfoClass.getMethod("newInstance", classOf[String])
+resTypeInfoNewInstanceMethod.invoke(null, resourceName)
+ }
+
+ private def reinitializeResources(resourceTypes: ListBuffer[AnyRef]):
Unit = {
--- End diff --
Inlined the method
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #20761: [SPARK-20327][CORE][YARN] Add CLI support for YAR...
Github user szyszy commented on a diff in the pull request:
https://github.com/apache/spark/pull/20761#discussion_r223918077
--- Diff:
resource-managers/yarn/src/test/scala/org/apache/spark/deploy/yarn/ResourceRequestHelperSuite.scala
---
@@ -0,0 +1,255 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.deploy.yarn
+
+import org.apache.hadoop.yarn.api.records.Resource
+import org.apache.hadoop.yarn.util.Records
+import org.scalatest.{BeforeAndAfterAll, Matchers}
+
+import org.apache.spark.{SparkConf, SparkException, SparkFunSuite}
+import
org.apache.spark.deploy.yarn.ResourceRequestTestHelper.ResourceInformation
+import org.apache.spark.deploy.yarn.config.{AM_CORES, AM_MEMORY,
DRIVER_CORES, EXECUTOR_CORES, YARN_AM_RESOURCE_TYPES_PREFIX,
YARN_DRIVER_RESOURCE_TYPES_PREFIX, YARN_EXECUTOR_RESOURCE_TYPES_PREFIX}
+import org.apache.spark.internal.config.{DRIVER_MEMORY, EXECUTOR_MEMORY}
+
+class ResourceRequestHelperSuite extends SparkFunSuite with Matchers with
BeforeAndAfterAll {
+
+ private val CUSTOM_RES_1 = "custom-resource-type-1"
+ private val CUSTOM_RES_2 = "custom-resource-type-2"
+ private val MEMORY = "memory"
+ private val CORES = "cores"
+ private val NEW_CONFIG_EXECUTOR_MEMORY =
YARN_EXECUTOR_RESOURCE_TYPES_PREFIX + MEMORY
+ private val NEW_CONFIG_EXECUTOR_CORES =
YARN_EXECUTOR_RESOURCE_TYPES_PREFIX + CORES
+ private val NEW_CONFIG_AM_MEMORY = YARN_AM_RESOURCE_TYPES_PREFIX + MEMORY
+ private val NEW_CONFIG_AM_CORES = YARN_AM_RESOURCE_TYPES_PREFIX + CORES
+ private val NEW_CONFIG_DRIVER_MEMORY = YARN_DRIVER_RESOURCE_TYPES_PREFIX
+ MEMORY
+ private val NEW_CONFIG_DRIVER_CORES = YARN_DRIVER_RESOURCE_TYPES_PREFIX
+ CORES
+
+ override def beforeAll(): Unit = {
+super.beforeAll()
+ }
+
+ private def getExpectedUnmatchedErrorMessage(name: String, value:
String): String = {
+s"Resource request for '$name' ('$value') does not match pattern
([0-9]+)([A-Za-z]*)."
+ }
+
+ test("resource type value does not match pattern") {
+assume(ResourceRequestHelper.isYarnResourceTypesAvailable())
+ResourceRequestTestHelper.initializeResourceTypes(List(CUSTOM_RES_1))
+
+val resourceTypes = Map(CUSTOM_RES_1 -> "**@#")
+
+val thrown = intercept[IllegalArgumentException] {
+ ResourceRequestHelper.setResourceRequests(resourceTypes,
createAResource)
+}
+thrown.getMessage should equal
(getExpectedUnmatchedErrorMessage(CUSTOM_RES_1, "**@#"))
+ }
+
+ test("resource type just unit defined") {
+assume(ResourceRequestHelper.isYarnResourceTypesAvailable())
+ResourceRequestTestHelper.initializeResourceTypes(List())
+
+val resourceTypes = Map(CUSTOM_RES_1 -> "m")
+
+val thrown = intercept[IllegalArgumentException] {
+ ResourceRequestHelper.setResourceRequests(resourceTypes,
createAResource)
+}
+thrown.getMessage should equal
(getExpectedUnmatchedErrorMessage(CUSTOM_RES_1, "m"))
+ }
+
+ test("resource type with null value should not be allowed") {
+assume(ResourceRequestHelper.isYarnResourceTypesAvailable())
+ResourceRequestTestHelper.initializeResourceTypes(List())
+
+val resourceTypes = Map(CUSTOM_RES_1 -> "123")
+
+val thrown = intercept[IllegalArgumentException] {
+ ResourceRequestHelper.setResourceRequests(resourceTypes, null)
+}
+thrown.getMessage should equal ("requirement failed: Resource
parameter should not be null!")
+ }
+
+ test("resource type with valid value and invalid unit") {
+assume(ResourceRequestHelper.isYarnResourceTypesAvailable())
+ResourceRequestTestHelper.initializeResourceTypes(List(CUSTOM_RES_1))
+
+val resourceTypes = Map(CUSTOM_RES_1 -> "123ppp")
+val resource = createAResource
+
+val thrown = intercept[IllegalArgumentException] {
+ ResourceRequestHelper.setResourceRequests(resourceTypes,
[GitHub] spark pull request #20761: [SPARK-20327][CORE][YARN] Add CLI support for YAR...
Github user szyszy commented on a diff in the pull request:
https://github.com/apache/spark/pull/20761#discussion_r223917264
--- Diff:
resource-managers/yarn/src/test/scala/org/apache/spark/deploy/yarn/ResourceRequestHelperSuite.scala
---
@@ -0,0 +1,255 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.deploy.yarn
+
+import org.apache.hadoop.yarn.api.records.Resource
+import org.apache.hadoop.yarn.util.Records
+import org.scalatest.{BeforeAndAfterAll, Matchers}
+
+import org.apache.spark.{SparkConf, SparkException, SparkFunSuite}
+import
org.apache.spark.deploy.yarn.ResourceRequestTestHelper.ResourceInformation
+import org.apache.spark.deploy.yarn.config.{AM_CORES, AM_MEMORY,
DRIVER_CORES, EXECUTOR_CORES, YARN_AM_RESOURCE_TYPES_PREFIX,
YARN_DRIVER_RESOURCE_TYPES_PREFIX, YARN_EXECUTOR_RESOURCE_TYPES_PREFIX}
+import org.apache.spark.internal.config.{DRIVER_MEMORY, EXECUTOR_MEMORY}
+
+class ResourceRequestHelperSuite extends SparkFunSuite with Matchers with
BeforeAndAfterAll {
+
+ private val CUSTOM_RES_1 = "custom-resource-type-1"
+ private val CUSTOM_RES_2 = "custom-resource-type-2"
+ private val MEMORY = "memory"
+ private val CORES = "cores"
+ private val NEW_CONFIG_EXECUTOR_MEMORY =
YARN_EXECUTOR_RESOURCE_TYPES_PREFIX + MEMORY
+ private val NEW_CONFIG_EXECUTOR_CORES =
YARN_EXECUTOR_RESOURCE_TYPES_PREFIX + CORES
+ private val NEW_CONFIG_AM_MEMORY = YARN_AM_RESOURCE_TYPES_PREFIX + MEMORY
+ private val NEW_CONFIG_AM_CORES = YARN_AM_RESOURCE_TYPES_PREFIX + CORES
+ private val NEW_CONFIG_DRIVER_MEMORY = YARN_DRIVER_RESOURCE_TYPES_PREFIX
+ MEMORY
+ private val NEW_CONFIG_DRIVER_CORES = YARN_DRIVER_RESOURCE_TYPES_PREFIX
+ CORES
+
+ override def beforeAll(): Unit = {
+super.beforeAll()
+ }
+
+ private def getExpectedUnmatchedErrorMessage(name: String, value:
String): String = {
+s"Resource request for '$name' ('$value') does not match pattern
([0-9]+)([A-Za-z]*)."
+ }
+
+ test("resource type value does not match pattern") {
--- End diff --
fixed
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #20761: [SPARK-20327][CORE][YARN] Add CLI support for YAR...
Github user szyszy commented on a diff in the pull request:
https://github.com/apache/spark/pull/20761#discussion_r223916534
--- Diff:
resource-managers/yarn/src/test/scala/org/apache/spark/deploy/yarn/ResourceRequestHelperSuite.scala
---
@@ -0,0 +1,255 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.deploy.yarn
+
+import org.apache.hadoop.yarn.api.records.Resource
+import org.apache.hadoop.yarn.util.Records
+import org.scalatest.{BeforeAndAfterAll, Matchers}
+
+import org.apache.spark.{SparkConf, SparkException, SparkFunSuite}
+import
org.apache.spark.deploy.yarn.ResourceRequestTestHelper.ResourceInformation
+import org.apache.spark.deploy.yarn.config.{AM_CORES, AM_MEMORY,
DRIVER_CORES, EXECUTOR_CORES, YARN_AM_RESOURCE_TYPES_PREFIX,
YARN_DRIVER_RESOURCE_TYPES_PREFIX, YARN_EXECUTOR_RESOURCE_TYPES_PREFIX}
+import org.apache.spark.internal.config.{DRIVER_MEMORY, EXECUTOR_MEMORY}
+
+class ResourceRequestHelperSuite extends SparkFunSuite with Matchers with
BeforeAndAfterAll {
+
+ private val CUSTOM_RES_1 = "custom-resource-type-1"
+ private val CUSTOM_RES_2 = "custom-resource-type-2"
+ private val MEMORY = "memory"
+ private val CORES = "cores"
+ private val NEW_CONFIG_EXECUTOR_MEMORY =
YARN_EXECUTOR_RESOURCE_TYPES_PREFIX + MEMORY
+ private val NEW_CONFIG_EXECUTOR_CORES =
YARN_EXECUTOR_RESOURCE_TYPES_PREFIX + CORES
+ private val NEW_CONFIG_AM_MEMORY = YARN_AM_RESOURCE_TYPES_PREFIX + MEMORY
+ private val NEW_CONFIG_AM_CORES = YARN_AM_RESOURCE_TYPES_PREFIX + CORES
+ private val NEW_CONFIG_DRIVER_MEMORY = YARN_DRIVER_RESOURCE_TYPES_PREFIX
+ MEMORY
+ private val NEW_CONFIG_DRIVER_CORES = YARN_DRIVER_RESOURCE_TYPES_PREFIX
+ CORES
+
+ override def beforeAll(): Unit = {
--- End diff --
removed
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #20761: [SPARK-20327][CORE][YARN] Add CLI support for YAR...
Github user szyszy commented on a diff in the pull request:
https://github.com/apache/spark/pull/20761#discussion_r223916183
--- Diff:
resource-managers/yarn/src/test/scala/org/apache/spark/deploy/yarn/ResourceRequestHelperSuite.scala
---
@@ -0,0 +1,255 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.deploy.yarn
+
+import org.apache.hadoop.yarn.api.records.Resource
+import org.apache.hadoop.yarn.util.Records
+import org.scalatest.{BeforeAndAfterAll, Matchers}
+
+import org.apache.spark.{SparkConf, SparkException, SparkFunSuite}
+import
org.apache.spark.deploy.yarn.ResourceRequestTestHelper.ResourceInformation
+import org.apache.spark.deploy.yarn.config.{AM_CORES, AM_MEMORY,
DRIVER_CORES, EXECUTOR_CORES, YARN_AM_RESOURCE_TYPES_PREFIX,
YARN_DRIVER_RESOURCE_TYPES_PREFIX, YARN_EXECUTOR_RESOURCE_TYPES_PREFIX}
--- End diff --
fixed
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #20761: [SPARK-20327][CORE][YARN] Add CLI support for YAR...
Github user szyszy commented on a diff in the pull request:
https://github.com/apache/spark/pull/20761#discussion_r223915996
--- Diff:
resource-managers/yarn/src/test/scala/org/apache/spark/deploy/yarn/ClientSuite.scala
---
@@ -35,7 +36,7 @@ import org.apache.hadoop.yarn.conf.YarnConfiguration
import org.apache.hadoop.yarn.util.Records
import org.mockito.Matchers.{eq => meq, _}
import org.mockito.Mockito._
-import org.scalatest.Matchers
+import org.scalatest.{BeforeAndAfterAll, Matchers}
--- End diff --
yep, it was unused, removed it.
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #20761: [SPARK-20327][CORE][YARN] Add CLI support for YAR...
Github user szyszy commented on a diff in the pull request:
https://github.com/apache/spark/pull/20761#discussion_r223915882
--- Diff:
resource-managers/yarn/src/main/scala/org/apache/spark/deploy/yarn/ResourceRequestHelper.scala
---
@@ -0,0 +1,148 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.deploy.yarn
+
+import java.lang.{Integer => JInteger, Long => JLong}
+import java.lang.reflect.InvocationTargetException
+
+import scala.collection.mutable
+import scala.util.Try
+
+import org.apache.hadoop.yarn.api.records.Resource
+
+import org.apache.spark.{SparkConf, SparkException}
+import org.apache.spark.deploy.yarn.config.{AM_CORES, AM_MEMORY,
DRIVER_CORES, EXECUTOR_CORES, YARN_AM_RESOURCE_TYPES_PREFIX,
YARN_DRIVER_RESOURCE_TYPES_PREFIX, YARN_EXECUTOR_RESOURCE_TYPES_PREFIX}
+import org.apache.spark.internal.Logging
+import org.apache.spark.internal.config.{DRIVER_MEMORY, EXECUTOR_MEMORY}
+import org.apache.spark.util.Utils
+
+/**
+ * This helper class uses some of Hadoop 3 methods from the YARN API,
+ * so we need to use reflection to avoid compile error when building
against Hadoop 2.x
+ */
+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"
+ private val ERROR_PREFIX: String = "Error:"
+
+ /**
+ * Validates sparkConf and throws a SparkException if any of standard
resources (memory or cores)
+ * is defined with the property spark.yarn.x.resource.y
+ * @param sparkConf
+ */
+ def validateResources(sparkConf: SparkConf): Unit = {
+val resourceDefinitions = Seq[(String, String)](
+ (AM_MEMORY.key, YARN_AM_RESOURCE_TYPES_PREFIX + "memory"),
+ (AM_CORES.key, YARN_AM_RESOURCE_TYPES_PREFIX + "cores"),
+ (DRIVER_MEMORY.key, YARN_DRIVER_RESOURCE_TYPES_PREFIX + "memory"),
+ (DRIVER_CORES.key, YARN_DRIVER_RESOURCE_TYPES_PREFIX + "cores"),
+ (EXECUTOR_MEMORY.key, YARN_EXECUTOR_RESOURCE_TYPES_PREFIX +
"memory"),
+ (EXECUTOR_CORES.key, YARN_EXECUTOR_RESOURCE_TYPES_PREFIX + "cores"))
+val errorMessage = new mutable.StringBuilder()
+
+resourceDefinitions.foreach { case (sparkName, resourceRequest) =>
+ if (sparkConf.contains(resourceRequest)) {
+errorMessage.append(s"$ERROR_PREFIX Do not use $resourceRequest, "
+
+s"please use $sparkName instead!\n")
+ }
+}
+
+if (errorMessage.nonEmpty) {
+ throw new SparkException(errorMessage.toString())
+}
+ }
+
+ /**
+ * Sets resource amount with the corresponding unit to the passed
resource object.
+ * @param resources resource values to set
+ * @param resource resource object to update
+ */
+ def setResourceRequests(
+ resources: Map[String, String],
+ resource: Resource): Unit = {
+require(resource != null, "Resource parameter should not be null!")
+require(!resources.contains("memory"),
+ "This method is meant to set custom resources to the resource
object, " +
+ "but memory is a standard resource!")
+require(!resources.contains("cores"),
+ "This method is meant to set custom resources to the resource
object, " +
+ "but cores is a standard resource!")
+
+logDebug(s"Custom resources requested: $resources")
+if (!isYarnResourceTypesAvailable()) {
+ if (resources.nonEmpty) {
+logWarning("Ignoring updating resource with resource types because
" +
--- End diff --
Fixed
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #20761: [SPARK-20327][CORE][YARN] Add CLI support for YAR...
Github user szyszy commented on a diff in the pull request:
https://github.com/apache/spark/pull/20761#discussion_r223915417
--- Diff:
resource-managers/yarn/src/main/scala/org/apache/spark/deploy/yarn/ResourceRequestHelper.scala
---
@@ -0,0 +1,148 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.deploy.yarn
+
+import java.lang.{Integer => JInteger, Long => JLong}
+import java.lang.reflect.InvocationTargetException
+
+import scala.collection.mutable
+import scala.util.Try
+
+import org.apache.hadoop.yarn.api.records.Resource
+
+import org.apache.spark.{SparkConf, SparkException}
+import org.apache.spark.deploy.yarn.config.{AM_CORES, AM_MEMORY,
DRIVER_CORES, EXECUTOR_CORES, YARN_AM_RESOURCE_TYPES_PREFIX,
YARN_DRIVER_RESOURCE_TYPES_PREFIX, YARN_EXECUTOR_RESOURCE_TYPES_PREFIX}
+import org.apache.spark.internal.Logging
+import org.apache.spark.internal.config.{DRIVER_MEMORY, EXECUTOR_MEMORY}
+import org.apache.spark.util.Utils
+
+/**
+ * This helper class uses some of Hadoop 3 methods from the YARN API,
+ * so we need to use reflection to avoid compile error when building
against Hadoop 2.x
+ */
+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"
+ private val ERROR_PREFIX: String = "Error:"
--- End diff --
Fixed
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #20761: [SPARK-20327][CORE][YARN] Add CLI support for YAR...
Github user szyszy commented on a diff in the pull request:
https://github.com/apache/spark/pull/20761#discussion_r223915451
--- Diff:
resource-managers/yarn/src/main/scala/org/apache/spark/deploy/yarn/ResourceRequestHelper.scala
---
@@ -0,0 +1,148 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.deploy.yarn
+
+import java.lang.{Integer => JInteger, Long => JLong}
+import java.lang.reflect.InvocationTargetException
+
+import scala.collection.mutable
+import scala.util.Try
+
+import org.apache.hadoop.yarn.api.records.Resource
+
+import org.apache.spark.{SparkConf, SparkException}
+import org.apache.spark.deploy.yarn.config.{AM_CORES, AM_MEMORY,
DRIVER_CORES, EXECUTOR_CORES, YARN_AM_RESOURCE_TYPES_PREFIX,
YARN_DRIVER_RESOURCE_TYPES_PREFIX, YARN_EXECUTOR_RESOURCE_TYPES_PREFIX}
+import org.apache.spark.internal.Logging
+import org.apache.spark.internal.config.{DRIVER_MEMORY, EXECUTOR_MEMORY}
+import org.apache.spark.util.Utils
+
+/**
+ * This helper class uses some of Hadoop 3 methods from the YARN API,
+ * so we need to use reflection to avoid compile error when building
against Hadoop 2.x
+ */
+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"
+ private val ERROR_PREFIX: String = "Error:"
+
+ /**
+ * Validates sparkConf and throws a SparkException if any of standard
resources (memory or cores)
+ * is defined with the property spark.yarn.x.resource.y
+ * @param sparkConf
--- End diff --
fixed
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #20761: [SPARK-20327][CORE][YARN] Add CLI support for YAR...
Github user vanzin commented on a diff in the pull request:
https://github.com/apache/spark/pull/20761#discussion_r223436285
--- Diff:
resource-managers/yarn/src/test/scala/org/apache/spark/deploy/yarn/ResourceRequestHelperSuite.scala
---
@@ -0,0 +1,255 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.deploy.yarn
+
+import org.apache.hadoop.yarn.api.records.Resource
+import org.apache.hadoop.yarn.util.Records
+import org.scalatest.{BeforeAndAfterAll, Matchers}
+
+import org.apache.spark.{SparkConf, SparkException, SparkFunSuite}
+import
org.apache.spark.deploy.yarn.ResourceRequestTestHelper.ResourceInformation
+import org.apache.spark.deploy.yarn.config.{AM_CORES, AM_MEMORY,
DRIVER_CORES, EXECUTOR_CORES, YARN_AM_RESOURCE_TYPES_PREFIX,
YARN_DRIVER_RESOURCE_TYPES_PREFIX, YARN_EXECUTOR_RESOURCE_TYPES_PREFIX}
--- End diff --
Just import `._` instead of adding so many individual imports.
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #20761: [SPARK-20327][CORE][YARN] Add CLI support for YAR...
Github user vanzin commented on a diff in the pull request:
https://github.com/apache/spark/pull/20761#discussion_r223440836
--- Diff:
resource-managers/yarn/src/test/scala/org/apache/spark/deploy/yarn/YarnAllocatorSuite.scala
---
@@ -134,6 +163,29 @@ class YarnAllocatorSuite extends SparkFunSuite with
Matchers with BeforeAndAfter
size should be (0)
}
+ test("custom resource requested from yarn") {
+assume(ResourceRequestHelper.isYarnResourceTypesAvailable())
+ResourceRequestTestHelper.initializeResourceTypes(List("gpu"))
+
+val mockAmClient = mock(classOf[AMRMClient[ContainerRequest]])
+val handler = createAllocatorWithAdditionalConfigs(1, Map(
+ YARN_EXECUTOR_RESOURCE_TYPES_PREFIX + "gpu" -> "2G"), mockAmClient)
+
+handler.updateResourceRequests()
+val container = createContainerWithResource("host1", handler.resource)
+handler.handleAllocatedContainers(Array(container))
+
+// get amount of memory and vcores from resource, so effectively
skipping their validation
+val expectedResources =
Resource.newInstance(handler.resource.getMemory(),
+ handler.resource.getVirtualCores)
+ResourceRequestHelper.setResourceRequests(Map("gpu" -> "2G"),
expectedResources)
+val captor = ArgumentCaptor.forClass(classOf[ContainerRequest])
+
+verify(mockAmClient).addContainerRequest(captor.capture())
+val containerRequest: ContainerRequest = captor.getValue
+assert(containerRequest.getCapability == expectedResources)
--- End diff --
`===`
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #20761: [SPARK-20327][CORE][YARN] Add CLI support for YAR...
Github user vanzin commented on a diff in the pull request:
https://github.com/apache/spark/pull/20761#discussion_r223439004
--- Diff:
resource-managers/yarn/src/test/scala/org/apache/spark/deploy/yarn/ResourceRequestTestHelper.scala
---
@@ -0,0 +1,90 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.deploy.yarn
+
+import scala.collection.JavaConverters._
+import scala.collection.mutable.ListBuffer
+
+import org.apache.hadoop.yarn.api.records.Resource
+
+import org.apache.spark.util.Utils
+
+object ResourceRequestTestHelper {
+ def initializeResourceTypes(resourceTypes: List[String]): Unit = {
+if (!ResourceRequestHelper.isYarnResourceTypesAvailable()) {
+ throw new IllegalStateException("initializeResourceTypes() should
not be invoked " +
+"since YARN resource types is not available because of old Hadoop
version!" )
+}
+
+val allResourceTypes = new ListBuffer[AnyRef]
+val defaultResourceTypes = List(
+ createResourceTypeInfo("memory-mb"),
+ createResourceTypeInfo("vcores"))
+val customResourceTypes = resourceTypes.map(rt =>
createResourceTypeInfo(rt))
+allResourceTypes ++= defaultResourceTypes
+allResourceTypes ++= customResourceTypes
+
+reinitializeResources(allResourceTypes)
+ }
+
+ private def createResourceTypeInfo(resourceName: String): AnyRef = {
+val resTypeInfoClass =
Utils.classForName("org.apache.hadoop.yarn.api.records.ResourceTypeInfo")
+val resTypeInfoNewInstanceMethod =
resTypeInfoClass.getMethod("newInstance", classOf[String])
+resTypeInfoNewInstanceMethod.invoke(null, resourceName)
+ }
+
+ private def reinitializeResources(resourceTypes: ListBuffer[AnyRef]):
Unit = {
--- End diff --
This is only called in one place. Inline it.
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #20761: [SPARK-20327][CORE][YARN] Add CLI support for YAR...
Github user vanzin commented on a diff in the pull request:
https://github.com/apache/spark/pull/20761#discussion_r223433455
--- Diff:
resource-managers/yarn/src/main/scala/org/apache/spark/deploy/yarn/ResourceRequestHelper.scala
---
@@ -0,0 +1,148 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.deploy.yarn
+
+import java.lang.{Integer => JInteger, Long => JLong}
+import java.lang.reflect.InvocationTargetException
+
+import scala.collection.mutable
+import scala.util.Try
+
+import org.apache.hadoop.yarn.api.records.Resource
+
+import org.apache.spark.{SparkConf, SparkException}
+import org.apache.spark.deploy.yarn.config.{AM_CORES, AM_MEMORY,
DRIVER_CORES, EXECUTOR_CORES, YARN_AM_RESOURCE_TYPES_PREFIX,
YARN_DRIVER_RESOURCE_TYPES_PREFIX, YARN_EXECUTOR_RESOURCE_TYPES_PREFIX}
+import org.apache.spark.internal.Logging
+import org.apache.spark.internal.config.{DRIVER_MEMORY, EXECUTOR_MEMORY}
+import org.apache.spark.util.Utils
+
+/**
+ * This helper class uses some of Hadoop 3 methods from the YARN API,
+ * so we need to use reflection to avoid compile error when building
against Hadoop 2.x
+ */
+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"
+ private val ERROR_PREFIX: String = "Error:"
+
+ /**
+ * Validates sparkConf and throws a SparkException if any of standard
resources (memory or cores)
+ * is defined with the property spark.yarn.x.resource.y
+ * @param sparkConf
+ */
+ def validateResources(sparkConf: SparkConf): Unit = {
+val resourceDefinitions = Seq[(String, String)](
+ (AM_MEMORY.key, YARN_AM_RESOURCE_TYPES_PREFIX + "memory"),
+ (AM_CORES.key, YARN_AM_RESOURCE_TYPES_PREFIX + "cores"),
+ (DRIVER_MEMORY.key, YARN_DRIVER_RESOURCE_TYPES_PREFIX + "memory"),
+ (DRIVER_CORES.key, YARN_DRIVER_RESOURCE_TYPES_PREFIX + "cores"),
+ (EXECUTOR_MEMORY.key, YARN_EXECUTOR_RESOURCE_TYPES_PREFIX +
"memory"),
+ (EXECUTOR_CORES.key, YARN_EXECUTOR_RESOURCE_TYPES_PREFIX + "cores"))
+val errorMessage = new mutable.StringBuilder()
+
+resourceDefinitions.foreach { case (sparkName, resourceRequest) =>
+ if (sparkConf.contains(resourceRequest)) {
+errorMessage.append(s"$ERROR_PREFIX Do not use $resourceRequest, "
+
+s"please use $sparkName instead!\n")
+ }
+}
+
+if (errorMessage.nonEmpty) {
+ throw new SparkException(errorMessage.toString())
+}
+ }
+
+ /**
+ * Sets resource amount with the corresponding unit to the passed
resource object.
+ * @param resources resource values to set
+ * @param resource resource object to update
+ */
+ def setResourceRequests(
+ resources: Map[String, String],
+ resource: Resource): Unit = {
+require(resource != null, "Resource parameter should not be null!")
+require(!resources.contains("memory"),
--- End diff --
Isn't this done by the validation method above? Either do it there or here,
but in both places it's a little overkill.
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #20761: [SPARK-20327][CORE][YARN] Add CLI support for YAR...
Github user vanzin commented on a diff in the pull request:
https://github.com/apache/spark/pull/20761#discussion_r223439326
--- Diff:
resource-managers/yarn/src/test/scala/org/apache/spark/deploy/yarn/ResourceRequestHelperSuite.scala
---
@@ -0,0 +1,255 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.deploy.yarn
+
+import org.apache.hadoop.yarn.api.records.Resource
+import org.apache.hadoop.yarn.util.Records
+import org.scalatest.{BeforeAndAfterAll, Matchers}
+
+import org.apache.spark.{SparkConf, SparkException, SparkFunSuite}
+import
org.apache.spark.deploy.yarn.ResourceRequestTestHelper.ResourceInformation
+import org.apache.spark.deploy.yarn.config.{AM_CORES, AM_MEMORY,
DRIVER_CORES, EXECUTOR_CORES, YARN_AM_RESOURCE_TYPES_PREFIX,
YARN_DRIVER_RESOURCE_TYPES_PREFIX, YARN_EXECUTOR_RESOURCE_TYPES_PREFIX}
+import org.apache.spark.internal.config.{DRIVER_MEMORY, EXECUTOR_MEMORY}
+
+class ResourceRequestHelperSuite extends SparkFunSuite with Matchers with
BeforeAndAfterAll {
+
+ private val CUSTOM_RES_1 = "custom-resource-type-1"
+ private val CUSTOM_RES_2 = "custom-resource-type-2"
+ private val MEMORY = "memory"
+ private val CORES = "cores"
+ private val NEW_CONFIG_EXECUTOR_MEMORY =
YARN_EXECUTOR_RESOURCE_TYPES_PREFIX + MEMORY
+ private val NEW_CONFIG_EXECUTOR_CORES =
YARN_EXECUTOR_RESOURCE_TYPES_PREFIX + CORES
+ private val NEW_CONFIG_AM_MEMORY = YARN_AM_RESOURCE_TYPES_PREFIX + MEMORY
+ private val NEW_CONFIG_AM_CORES = YARN_AM_RESOURCE_TYPES_PREFIX + CORES
+ private val NEW_CONFIG_DRIVER_MEMORY = YARN_DRIVER_RESOURCE_TYPES_PREFIX
+ MEMORY
+ private val NEW_CONFIG_DRIVER_CORES = YARN_DRIVER_RESOURCE_TYPES_PREFIX
+ CORES
+
+ override def beforeAll(): Unit = {
+super.beforeAll()
+ }
+
+ private def getExpectedUnmatchedErrorMessage(name: String, value:
String): String = {
+s"Resource request for '$name' ('$value') does not match pattern
([0-9]+)([A-Za-z]*)."
+ }
+
+ test("resource type value does not match pattern") {
+assume(ResourceRequestHelper.isYarnResourceTypesAvailable())
+ResourceRequestTestHelper.initializeResourceTypes(List(CUSTOM_RES_1))
+
+val resourceTypes = Map(CUSTOM_RES_1 -> "**@#")
+
+val thrown = intercept[IllegalArgumentException] {
+ ResourceRequestHelper.setResourceRequests(resourceTypes,
createAResource)
+}
+thrown.getMessage should equal
(getExpectedUnmatchedErrorMessage(CUSTOM_RES_1, "**@#"))
+ }
+
+ test("resource type just unit defined") {
+assume(ResourceRequestHelper.isYarnResourceTypesAvailable())
+ResourceRequestTestHelper.initializeResourceTypes(List())
+
+val resourceTypes = Map(CUSTOM_RES_1 -> "m")
+
+val thrown = intercept[IllegalArgumentException] {
+ ResourceRequestHelper.setResourceRequests(resourceTypes,
createAResource)
+}
+thrown.getMessage should equal
(getExpectedUnmatchedErrorMessage(CUSTOM_RES_1, "m"))
+ }
+
+ test("resource type with null value should not be allowed") {
+assume(ResourceRequestHelper.isYarnResourceTypesAvailable())
+ResourceRequestTestHelper.initializeResourceTypes(List())
+
+val resourceTypes = Map(CUSTOM_RES_1 -> "123")
+
+val thrown = intercept[IllegalArgumentException] {
+ ResourceRequestHelper.setResourceRequests(resourceTypes, null)
+}
+thrown.getMessage should equal ("requirement failed: Resource
parameter should not be null!")
+ }
+
+ test("resource type with valid value and invalid unit") {
+assume(ResourceRequestHelper.isYarnResourceTypesAvailable())
+ResourceRequestTestHelper.initializeResourceTypes(List(CUSTOM_RES_1))
+
+val resourceTypes = Map(CUSTOM_RES_1 -> "123ppp")
+val resource = createAResource
+
+val thrown = intercept[IllegalArgumentException] {
+ ResourceRequestHelper.setResourceRequests(resourceTypes,
[GitHub] spark pull request #20761: [SPARK-20327][CORE][YARN] Add CLI support for YAR...
Github user vanzin commented on a diff in the pull request:
https://github.com/apache/spark/pull/20761#discussion_r223441224
--- Diff:
resource-managers/yarn/src/test/scala/org/apache/spark/deploy/yarn/ResourceRequestHelperSuite.scala
---
@@ -0,0 +1,255 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.deploy.yarn
+
+import org.apache.hadoop.yarn.api.records.Resource
+import org.apache.hadoop.yarn.util.Records
+import org.scalatest.{BeforeAndAfterAll, Matchers}
+
+import org.apache.spark.{SparkConf, SparkException, SparkFunSuite}
+import
org.apache.spark.deploy.yarn.ResourceRequestTestHelper.ResourceInformation
+import org.apache.spark.deploy.yarn.config.{AM_CORES, AM_MEMORY,
DRIVER_CORES, EXECUTOR_CORES, YARN_AM_RESOURCE_TYPES_PREFIX,
YARN_DRIVER_RESOURCE_TYPES_PREFIX, YARN_EXECUTOR_RESOURCE_TYPES_PREFIX}
+import org.apache.spark.internal.config.{DRIVER_MEMORY, EXECUTOR_MEMORY}
+
+class ResourceRequestHelperSuite extends SparkFunSuite with Matchers with
BeforeAndAfterAll {
+
+ private val CUSTOM_RES_1 = "custom-resource-type-1"
+ private val CUSTOM_RES_2 = "custom-resource-type-2"
+ private val MEMORY = "memory"
+ private val CORES = "cores"
+ private val NEW_CONFIG_EXECUTOR_MEMORY =
YARN_EXECUTOR_RESOURCE_TYPES_PREFIX + MEMORY
+ private val NEW_CONFIG_EXECUTOR_CORES =
YARN_EXECUTOR_RESOURCE_TYPES_PREFIX + CORES
+ private val NEW_CONFIG_AM_MEMORY = YARN_AM_RESOURCE_TYPES_PREFIX + MEMORY
+ private val NEW_CONFIG_AM_CORES = YARN_AM_RESOURCE_TYPES_PREFIX + CORES
+ private val NEW_CONFIG_DRIVER_MEMORY = YARN_DRIVER_RESOURCE_TYPES_PREFIX
+ MEMORY
+ private val NEW_CONFIG_DRIVER_CORES = YARN_DRIVER_RESOURCE_TYPES_PREFIX
+ CORES
+
+ override def beforeAll(): Unit = {
+super.beforeAll()
+ }
+
+ private def getExpectedUnmatchedErrorMessage(name: String, value:
String): String = {
+s"Resource request for '$name' ('$value') does not match pattern
([0-9]+)([A-Za-z]*)."
+ }
+
+ test("resource type value does not match pattern") {
+assume(ResourceRequestHelper.isYarnResourceTypesAvailable())
+ResourceRequestTestHelper.initializeResourceTypes(List(CUSTOM_RES_1))
+
+val resourceTypes = Map(CUSTOM_RES_1 -> "**@#")
+
+val thrown = intercept[IllegalArgumentException] {
+ ResourceRequestHelper.setResourceRequests(resourceTypes,
createAResource)
+}
+thrown.getMessage should equal
(getExpectedUnmatchedErrorMessage(CUSTOM_RES_1, "**@#"))
+ }
+
+ test("resource type just unit defined") {
+assume(ResourceRequestHelper.isYarnResourceTypesAvailable())
+ResourceRequestTestHelper.initializeResourceTypes(List())
+
+val resourceTypes = Map(CUSTOM_RES_1 -> "m")
+
+val thrown = intercept[IllegalArgumentException] {
+ ResourceRequestHelper.setResourceRequests(resourceTypes,
createAResource)
+}
+thrown.getMessage should equal
(getExpectedUnmatchedErrorMessage(CUSTOM_RES_1, "m"))
+ }
+
+ test("resource type with null value should not be allowed") {
+assume(ResourceRequestHelper.isYarnResourceTypesAvailable())
+ResourceRequestTestHelper.initializeResourceTypes(List())
+
+val resourceTypes = Map(CUSTOM_RES_1 -> "123")
+
+val thrown = intercept[IllegalArgumentException] {
+ ResourceRequestHelper.setResourceRequests(resourceTypes, null)
+}
+thrown.getMessage should equal ("requirement failed: Resource
parameter should not be null!")
+ }
+
+ test("resource type with valid value and invalid unit") {
+assume(ResourceRequestHelper.isYarnResourceTypesAvailable())
+ResourceRequestTestHelper.initializeResourceTypes(List(CUSTOM_RES_1))
+
+val resourceTypes = Map(CUSTOM_RES_1 -> "123ppp")
+val resource = createAResource
+
+val thrown = intercept[IllegalArgumentException] {
+ ResourceRequestHelper.setResourceRequests(resourceTypes,
[GitHub] spark pull request #20761: [SPARK-20327][CORE][YARN] Add CLI support for YAR...
Github user vanzin commented on a diff in the pull request:
https://github.com/apache/spark/pull/20761#discussion_r223437860
--- Diff:
resource-managers/yarn/src/test/scala/org/apache/spark/deploy/yarn/ResourceRequestHelperSuite.scala
---
@@ -0,0 +1,255 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.deploy.yarn
+
+import org.apache.hadoop.yarn.api.records.Resource
+import org.apache.hadoop.yarn.util.Records
+import org.scalatest.{BeforeAndAfterAll, Matchers}
+
+import org.apache.spark.{SparkConf, SparkException, SparkFunSuite}
+import
org.apache.spark.deploy.yarn.ResourceRequestTestHelper.ResourceInformation
+import org.apache.spark.deploy.yarn.config.{AM_CORES, AM_MEMORY,
DRIVER_CORES, EXECUTOR_CORES, YARN_AM_RESOURCE_TYPES_PREFIX,
YARN_DRIVER_RESOURCE_TYPES_PREFIX, YARN_EXECUTOR_RESOURCE_TYPES_PREFIX}
+import org.apache.spark.internal.config.{DRIVER_MEMORY, EXECUTOR_MEMORY}
+
+class ResourceRequestHelperSuite extends SparkFunSuite with Matchers with
BeforeAndAfterAll {
+
+ private val CUSTOM_RES_1 = "custom-resource-type-1"
+ private val CUSTOM_RES_2 = "custom-resource-type-2"
+ private val MEMORY = "memory"
+ private val CORES = "cores"
+ private val NEW_CONFIG_EXECUTOR_MEMORY =
YARN_EXECUTOR_RESOURCE_TYPES_PREFIX + MEMORY
+ private val NEW_CONFIG_EXECUTOR_CORES =
YARN_EXECUTOR_RESOURCE_TYPES_PREFIX + CORES
+ private val NEW_CONFIG_AM_MEMORY = YARN_AM_RESOURCE_TYPES_PREFIX + MEMORY
+ private val NEW_CONFIG_AM_CORES = YARN_AM_RESOURCE_TYPES_PREFIX + CORES
+ private val NEW_CONFIG_DRIVER_MEMORY = YARN_DRIVER_RESOURCE_TYPES_PREFIX
+ MEMORY
+ private val NEW_CONFIG_DRIVER_CORES = YARN_DRIVER_RESOURCE_TYPES_PREFIX
+ CORES
+
+ override def beforeAll(): Unit = {
+super.beforeAll()
+ }
+
+ private def getExpectedUnmatchedErrorMessage(name: String, value:
String): String = {
+s"Resource request for '$name' ('$value') does not match pattern
([0-9]+)([A-Za-z]*)."
+ }
+
+ test("resource type value does not match pattern") {
+assume(ResourceRequestHelper.isYarnResourceTypesAvailable())
+ResourceRequestTestHelper.initializeResourceTypes(List(CUSTOM_RES_1))
+
+val resourceTypes = Map(CUSTOM_RES_1 -> "**@#")
+
+val thrown = intercept[IllegalArgumentException] {
+ ResourceRequestHelper.setResourceRequests(resourceTypes,
createAResource)
+}
+thrown.getMessage should equal
(getExpectedUnmatchedErrorMessage(CUSTOM_RES_1, "**@#"))
+ }
+
+ test("resource type just unit defined") {
--- End diff --
These tests are a whole lot of code to just test the regex parsing request
values in `setResourceRequests`.
If you're so worried about testing that particular regex + parsing code,
then put it in a separate method and call it with a bunch of difference
parameters here. It would make this code a whole lot simpler.
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #20761: [SPARK-20327][CORE][YARN] Add CLI support for YAR...
Github user vanzin commented on a diff in the pull request:
https://github.com/apache/spark/pull/20761#discussion_r223437109
--- Diff:
resource-managers/yarn/src/test/scala/org/apache/spark/deploy/yarn/ResourceRequestHelperSuite.scala
---
@@ -0,0 +1,255 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.deploy.yarn
+
+import org.apache.hadoop.yarn.api.records.Resource
+import org.apache.hadoop.yarn.util.Records
+import org.scalatest.{BeforeAndAfterAll, Matchers}
+
+import org.apache.spark.{SparkConf, SparkException, SparkFunSuite}
+import
org.apache.spark.deploy.yarn.ResourceRequestTestHelper.ResourceInformation
+import org.apache.spark.deploy.yarn.config.{AM_CORES, AM_MEMORY,
DRIVER_CORES, EXECUTOR_CORES, YARN_AM_RESOURCE_TYPES_PREFIX,
YARN_DRIVER_RESOURCE_TYPES_PREFIX, YARN_EXECUTOR_RESOURCE_TYPES_PREFIX}
+import org.apache.spark.internal.config.{DRIVER_MEMORY, EXECUTOR_MEMORY}
+
+class ResourceRequestHelperSuite extends SparkFunSuite with Matchers with
BeforeAndAfterAll {
+
+ private val CUSTOM_RES_1 = "custom-resource-type-1"
+ private val CUSTOM_RES_2 = "custom-resource-type-2"
+ private val MEMORY = "memory"
+ private val CORES = "cores"
+ private val NEW_CONFIG_EXECUTOR_MEMORY =
YARN_EXECUTOR_RESOURCE_TYPES_PREFIX + MEMORY
+ private val NEW_CONFIG_EXECUTOR_CORES =
YARN_EXECUTOR_RESOURCE_TYPES_PREFIX + CORES
+ private val NEW_CONFIG_AM_MEMORY = YARN_AM_RESOURCE_TYPES_PREFIX + MEMORY
+ private val NEW_CONFIG_AM_CORES = YARN_AM_RESOURCE_TYPES_PREFIX + CORES
+ private val NEW_CONFIG_DRIVER_MEMORY = YARN_DRIVER_RESOURCE_TYPES_PREFIX
+ MEMORY
+ private val NEW_CONFIG_DRIVER_CORES = YARN_DRIVER_RESOURCE_TYPES_PREFIX
+ CORES
+
+ override def beforeAll(): Unit = {
+super.beforeAll()
+ }
+
+ private def getExpectedUnmatchedErrorMessage(name: String, value:
String): String = {
+s"Resource request for '$name' ('$value') does not match pattern
([0-9]+)([A-Za-z]*)."
+ }
+
+ test("resource type value does not match pattern") {
--- End diff --
In all these tests, s/resource type/resource request/, including in
variable names.
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #20761: [SPARK-20327][CORE][YARN] Add CLI support for YAR...
Github user vanzin commented on a diff in the pull request:
https://github.com/apache/spark/pull/20761#discussion_r223438072
--- Diff:
resource-managers/yarn/src/test/scala/org/apache/spark/deploy/yarn/ResourceRequestHelperSuite.scala
---
@@ -0,0 +1,255 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.deploy.yarn
+
+import org.apache.hadoop.yarn.api.records.Resource
+import org.apache.hadoop.yarn.util.Records
+import org.scalatest.{BeforeAndAfterAll, Matchers}
+
+import org.apache.spark.{SparkConf, SparkException, SparkFunSuite}
+import
org.apache.spark.deploy.yarn.ResourceRequestTestHelper.ResourceInformation
+import org.apache.spark.deploy.yarn.config.{AM_CORES, AM_MEMORY,
DRIVER_CORES, EXECUTOR_CORES, YARN_AM_RESOURCE_TYPES_PREFIX,
YARN_DRIVER_RESOURCE_TYPES_PREFIX, YARN_EXECUTOR_RESOURCE_TYPES_PREFIX}
+import org.apache.spark.internal.config.{DRIVER_MEMORY, EXECUTOR_MEMORY}
+
+class ResourceRequestHelperSuite extends SparkFunSuite with Matchers with
BeforeAndAfterAll {
+
+ private val CUSTOM_RES_1 = "custom-resource-type-1"
+ private val CUSTOM_RES_2 = "custom-resource-type-2"
+ private val MEMORY = "memory"
+ private val CORES = "cores"
+ private val NEW_CONFIG_EXECUTOR_MEMORY =
YARN_EXECUTOR_RESOURCE_TYPES_PREFIX + MEMORY
+ private val NEW_CONFIG_EXECUTOR_CORES =
YARN_EXECUTOR_RESOURCE_TYPES_PREFIX + CORES
+ private val NEW_CONFIG_AM_MEMORY = YARN_AM_RESOURCE_TYPES_PREFIX + MEMORY
+ private val NEW_CONFIG_AM_CORES = YARN_AM_RESOURCE_TYPES_PREFIX + CORES
+ private val NEW_CONFIG_DRIVER_MEMORY = YARN_DRIVER_RESOURCE_TYPES_PREFIX
+ MEMORY
+ private val NEW_CONFIG_DRIVER_CORES = YARN_DRIVER_RESOURCE_TYPES_PREFIX
+ CORES
+
+ override def beforeAll(): Unit = {
+super.beforeAll()
+ }
+
+ private def getExpectedUnmatchedErrorMessage(name: String, value:
String): String = {
+s"Resource request for '$name' ('$value') does not match pattern
([0-9]+)([A-Za-z]*)."
+ }
+
+ test("resource type value does not match pattern") {
+assume(ResourceRequestHelper.isYarnResourceTypesAvailable())
+ResourceRequestTestHelper.initializeResourceTypes(List(CUSTOM_RES_1))
+
+val resourceTypes = Map(CUSTOM_RES_1 -> "**@#")
+
+val thrown = intercept[IllegalArgumentException] {
+ ResourceRequestHelper.setResourceRequests(resourceTypes,
createAResource)
+}
+thrown.getMessage should equal
(getExpectedUnmatchedErrorMessage(CUSTOM_RES_1, "**@#"))
+ }
+
+ test("resource type just unit defined") {
+assume(ResourceRequestHelper.isYarnResourceTypesAvailable())
+ResourceRequestTestHelper.initializeResourceTypes(List())
+
+val resourceTypes = Map(CUSTOM_RES_1 -> "m")
+
+val thrown = intercept[IllegalArgumentException] {
+ ResourceRequestHelper.setResourceRequests(resourceTypes,
createAResource)
+}
+thrown.getMessage should equal
(getExpectedUnmatchedErrorMessage(CUSTOM_RES_1, "m"))
+ }
+
+ test("resource type with null value should not be allowed") {
+assume(ResourceRequestHelper.isYarnResourceTypesAvailable())
+ResourceRequestTestHelper.initializeResourceTypes(List())
+
+val resourceTypes = Map(CUSTOM_RES_1 -> "123")
+
+val thrown = intercept[IllegalArgumentException] {
+ ResourceRequestHelper.setResourceRequests(resourceTypes, null)
+}
+thrown.getMessage should equal ("requirement failed: Resource
parameter should not be null!")
+ }
+
+ test("resource type with valid value and invalid unit") {
+assume(ResourceRequestHelper.isYarnResourceTypesAvailable())
+ResourceRequestTestHelper.initializeResourceTypes(List(CUSTOM_RES_1))
+
+val resourceTypes = Map(CUSTOM_RES_1 -> "123ppp")
+val resource = createAResource
+
+val thrown = intercept[IllegalArgumentException] {
+ ResourceRequestHelper.setResourceRequests(resourceTypes,
[GitHub] spark pull request #20761: [SPARK-20327][CORE][YARN] Add CLI support for YAR...
Github user vanzin commented on a diff in the pull request:
https://github.com/apache/spark/pull/20761#discussion_r223438810
--- Diff:
resource-managers/yarn/src/test/scala/org/apache/spark/deploy/yarn/ResourceRequestHelperSuite.scala
---
@@ -0,0 +1,255 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.deploy.yarn
+
+import org.apache.hadoop.yarn.api.records.Resource
+import org.apache.hadoop.yarn.util.Records
+import org.scalatest.{BeforeAndAfterAll, Matchers}
+
+import org.apache.spark.{SparkConf, SparkException, SparkFunSuite}
+import
org.apache.spark.deploy.yarn.ResourceRequestTestHelper.ResourceInformation
+import org.apache.spark.deploy.yarn.config.{AM_CORES, AM_MEMORY,
DRIVER_CORES, EXECUTOR_CORES, YARN_AM_RESOURCE_TYPES_PREFIX,
YARN_DRIVER_RESOURCE_TYPES_PREFIX, YARN_EXECUTOR_RESOURCE_TYPES_PREFIX}
+import org.apache.spark.internal.config.{DRIVER_MEMORY, EXECUTOR_MEMORY}
+
+class ResourceRequestHelperSuite extends SparkFunSuite with Matchers with
BeforeAndAfterAll {
+
+ private val CUSTOM_RES_1 = "custom-resource-type-1"
+ private val CUSTOM_RES_2 = "custom-resource-type-2"
+ private val MEMORY = "memory"
+ private val CORES = "cores"
+ private val NEW_CONFIG_EXECUTOR_MEMORY =
YARN_EXECUTOR_RESOURCE_TYPES_PREFIX + MEMORY
+ private val NEW_CONFIG_EXECUTOR_CORES =
YARN_EXECUTOR_RESOURCE_TYPES_PREFIX + CORES
+ private val NEW_CONFIG_AM_MEMORY = YARN_AM_RESOURCE_TYPES_PREFIX + MEMORY
+ private val NEW_CONFIG_AM_CORES = YARN_AM_RESOURCE_TYPES_PREFIX + CORES
+ private val NEW_CONFIG_DRIVER_MEMORY = YARN_DRIVER_RESOURCE_TYPES_PREFIX
+ MEMORY
+ private val NEW_CONFIG_DRIVER_CORES = YARN_DRIVER_RESOURCE_TYPES_PREFIX
+ CORES
+
+ override def beforeAll(): Unit = {
+super.beforeAll()
+ }
+
+ private def getExpectedUnmatchedErrorMessage(name: String, value:
String): String = {
+s"Resource request for '$name' ('$value') does not match pattern
([0-9]+)([A-Za-z]*)."
+ }
+
+ test("resource type value does not match pattern") {
+assume(ResourceRequestHelper.isYarnResourceTypesAvailable())
+ResourceRequestTestHelper.initializeResourceTypes(List(CUSTOM_RES_1))
+
+val resourceTypes = Map(CUSTOM_RES_1 -> "**@#")
+
+val thrown = intercept[IllegalArgumentException] {
+ ResourceRequestHelper.setResourceRequests(resourceTypes,
createAResource)
+}
+thrown.getMessage should equal
(getExpectedUnmatchedErrorMessage(CUSTOM_RES_1, "**@#"))
+ }
+
+ test("resource type just unit defined") {
+assume(ResourceRequestHelper.isYarnResourceTypesAvailable())
+ResourceRequestTestHelper.initializeResourceTypes(List())
+
+val resourceTypes = Map(CUSTOM_RES_1 -> "m")
+
+val thrown = intercept[IllegalArgumentException] {
+ ResourceRequestHelper.setResourceRequests(resourceTypes,
createAResource)
+}
+thrown.getMessage should equal
(getExpectedUnmatchedErrorMessage(CUSTOM_RES_1, "m"))
+ }
+
+ test("resource type with null value should not be allowed") {
+assume(ResourceRequestHelper.isYarnResourceTypesAvailable())
+ResourceRequestTestHelper.initializeResourceTypes(List())
+
+val resourceTypes = Map(CUSTOM_RES_1 -> "123")
+
+val thrown = intercept[IllegalArgumentException] {
+ ResourceRequestHelper.setResourceRequests(resourceTypes, null)
+}
+thrown.getMessage should equal ("requirement failed: Resource
parameter should not be null!")
+ }
+
+ test("resource type with valid value and invalid unit") {
+assume(ResourceRequestHelper.isYarnResourceTypesAvailable())
+ResourceRequestTestHelper.initializeResourceTypes(List(CUSTOM_RES_1))
+
+val resourceTypes = Map(CUSTOM_RES_1 -> "123ppp")
+val resource = createAResource
+
+val thrown = intercept[IllegalArgumentException] {
+ ResourceRequestHelper.setResourceRequests(resourceTypes,
[GitHub] spark pull request #20761: [SPARK-20327][CORE][YARN] Add CLI support for YAR...
Github user vanzin commented on a diff in the pull request:
https://github.com/apache/spark/pull/20761#discussion_r223434738
--- Diff:
resource-managers/yarn/src/main/scala/org/apache/spark/deploy/yarn/YarnAllocator.scala
---
@@ -288,9 +296,14 @@ private[yarn] class YarnAllocator(
s"executorsStarting: ${numExecutorsStarting.get}")
if (missing > 0) {
- logInfo(s"Will request $missing executor container(s), each with " +
-s"${resource.getVirtualCores} core(s) and " +
-s"${resource.getMemory} MB memory (including $memoryOverhead MB of
overhead)")
+ var requestContainerMessage = s"Will request $missing executor
container(s), each with " +
--- End diff --
Minor, but this now should be inside a `if (log.isInfoEnabled())` block.
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #20761: [SPARK-20327][CORE][YARN] Add CLI support for YAR...
Github user vanzin commented on a diff in the pull request:
https://github.com/apache/spark/pull/20761#discussion_r223436523
--- Diff:
resource-managers/yarn/src/test/scala/org/apache/spark/deploy/yarn/ResourceRequestHelperSuite.scala
---
@@ -0,0 +1,255 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.deploy.yarn
+
+import org.apache.hadoop.yarn.api.records.Resource
+import org.apache.hadoop.yarn.util.Records
+import org.scalatest.{BeforeAndAfterAll, Matchers}
+
+import org.apache.spark.{SparkConf, SparkException, SparkFunSuite}
+import
org.apache.spark.deploy.yarn.ResourceRequestTestHelper.ResourceInformation
+import org.apache.spark.deploy.yarn.config.{AM_CORES, AM_MEMORY,
DRIVER_CORES, EXECUTOR_CORES, YARN_AM_RESOURCE_TYPES_PREFIX,
YARN_DRIVER_RESOURCE_TYPES_PREFIX, YARN_EXECUTOR_RESOURCE_TYPES_PREFIX}
+import org.apache.spark.internal.config.{DRIVER_MEMORY, EXECUTOR_MEMORY}
+
+class ResourceRequestHelperSuite extends SparkFunSuite with Matchers with
BeforeAndAfterAll {
+
+ private val CUSTOM_RES_1 = "custom-resource-type-1"
+ private val CUSTOM_RES_2 = "custom-resource-type-2"
+ private val MEMORY = "memory"
+ private val CORES = "cores"
+ private val NEW_CONFIG_EXECUTOR_MEMORY =
YARN_EXECUTOR_RESOURCE_TYPES_PREFIX + MEMORY
+ private val NEW_CONFIG_EXECUTOR_CORES =
YARN_EXECUTOR_RESOURCE_TYPES_PREFIX + CORES
+ private val NEW_CONFIG_AM_MEMORY = YARN_AM_RESOURCE_TYPES_PREFIX + MEMORY
+ private val NEW_CONFIG_AM_CORES = YARN_AM_RESOURCE_TYPES_PREFIX + CORES
+ private val NEW_CONFIG_DRIVER_MEMORY = YARN_DRIVER_RESOURCE_TYPES_PREFIX
+ MEMORY
+ private val NEW_CONFIG_DRIVER_CORES = YARN_DRIVER_RESOURCE_TYPES_PREFIX
+ CORES
+
+ override def beforeAll(): Unit = {
+super.beforeAll()
+ }
+
+ private def getExpectedUnmatchedErrorMessage(name: String, value:
String): String = {
--- End diff --
I thought we discussed how checking for exact error messages is not a good
test.
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #20761: [SPARK-20327][CORE][YARN] Add CLI support for YAR...
Github user vanzin commented on a diff in the pull request:
https://github.com/apache/spark/pull/20761#discussion_r223435325
--- Diff:
resource-managers/yarn/src/test/scala/org/apache/spark/deploy/yarn/ClientSuite.scala
---
@@ -199,6 +200,37 @@ class ClientSuite extends SparkFunSuite with Matchers {
appContext.getMaxAppAttempts should be (42)
}
+ test("resource request for invalid resource") {
--- End diff --
Is this testing any Spark code at all? It seems to just be a long way to
test YARN code that checks whether a resource type has been registered.
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #20761: [SPARK-20327][CORE][YARN] Add CLI support for YAR...
Github user vanzin commented on a diff in the pull request:
https://github.com/apache/spark/pull/20761#discussion_r223442688
--- Diff:
resource-managers/yarn/src/test/scala/org/apache/spark/deploy/yarn/YarnAllocatorSuite.scala
---
@@ -87,6 +88,20 @@ class YarnAllocatorSuite extends SparkFunSuite with
Matchers with BeforeAndAfter
def createAllocator(
maxExecutors: Int = 5,
rmClient: AMRMClient[ContainerRequest] = rmClient): YarnAllocator = {
+createAllocatorInternal(maxExecutors, rmClient, Map())
+ }
+
+ def createAllocatorWithAdditionalConfigs(
--- End diff --
You can just add a new parameter with a default value to the existing
method and avoid this.
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #20761: [SPARK-20327][CORE][YARN] Add CLI support for YAR...
Github user vanzin commented on a diff in the pull request:
https://github.com/apache/spark/pull/20761#discussion_r223433617
--- Diff:
resource-managers/yarn/src/main/scala/org/apache/spark/deploy/yarn/ResourceRequestHelper.scala
---
@@ -0,0 +1,148 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.deploy.yarn
+
+import java.lang.{Integer => JInteger, Long => JLong}
+import java.lang.reflect.InvocationTargetException
+
+import scala.collection.mutable
+import scala.util.Try
+
+import org.apache.hadoop.yarn.api.records.Resource
+
+import org.apache.spark.{SparkConf, SparkException}
+import org.apache.spark.deploy.yarn.config.{AM_CORES, AM_MEMORY,
DRIVER_CORES, EXECUTOR_CORES, YARN_AM_RESOURCE_TYPES_PREFIX,
YARN_DRIVER_RESOURCE_TYPES_PREFIX, YARN_EXECUTOR_RESOURCE_TYPES_PREFIX}
+import org.apache.spark.internal.Logging
+import org.apache.spark.internal.config.{DRIVER_MEMORY, EXECUTOR_MEMORY}
+import org.apache.spark.util.Utils
+
+/**
+ * This helper class uses some of Hadoop 3 methods from the YARN API,
+ * so we need to use reflection to avoid compile error when building
against Hadoop 2.x
+ */
+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"
+ private val ERROR_PREFIX: String = "Error:"
+
+ /**
+ * Validates sparkConf and throws a SparkException if any of standard
resources (memory or cores)
+ * is defined with the property spark.yarn.x.resource.y
+ * @param sparkConf
+ */
+ def validateResources(sparkConf: SparkConf): Unit = {
+val resourceDefinitions = Seq[(String, String)](
+ (AM_MEMORY.key, YARN_AM_RESOURCE_TYPES_PREFIX + "memory"),
+ (AM_CORES.key, YARN_AM_RESOURCE_TYPES_PREFIX + "cores"),
+ (DRIVER_MEMORY.key, YARN_DRIVER_RESOURCE_TYPES_PREFIX + "memory"),
+ (DRIVER_CORES.key, YARN_DRIVER_RESOURCE_TYPES_PREFIX + "cores"),
+ (EXECUTOR_MEMORY.key, YARN_EXECUTOR_RESOURCE_TYPES_PREFIX +
"memory"),
+ (EXECUTOR_CORES.key, YARN_EXECUTOR_RESOURCE_TYPES_PREFIX + "cores"))
+val errorMessage = new mutable.StringBuilder()
+
+resourceDefinitions.foreach { case (sparkName, resourceRequest) =>
+ if (sparkConf.contains(resourceRequest)) {
+errorMessage.append(s"$ERROR_PREFIX Do not use $resourceRequest, "
+
+s"please use $sparkName instead!\n")
+ }
+}
+
+if (errorMessage.nonEmpty) {
+ throw new SparkException(errorMessage.toString())
+}
+ }
+
+ /**
+ * Sets resource amount with the corresponding unit to the passed
resource object.
+ * @param resources resource values to set
+ * @param resource resource object to update
+ */
+ def setResourceRequests(
+ resources: Map[String, String],
+ resource: Resource): Unit = {
+require(resource != null, "Resource parameter should not be null!")
+require(!resources.contains("memory"),
+ "This method is meant to set custom resources to the resource
object, " +
+ "but memory is a standard resource!")
+require(!resources.contains("cores"),
+ "This method is meant to set custom resources to the resource
object, " +
+ "but cores is a standard resource!")
+
+logDebug(s"Custom resources requested: $resources")
+if (!isYarnResourceTypesAvailable()) {
+ if (resources.nonEmpty) {
+logWarning("Ignoring updating resource with resource types because
" +
--- End diff --
better: "Ignoring custom resource requests because..."
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #20761: [SPARK-20327][CORE][YARN] Add CLI support for YAR...
Github user vanzin commented on a diff in the pull request:
https://github.com/apache/spark/pull/20761#discussion_r223436337
--- Diff:
resource-managers/yarn/src/test/scala/org/apache/spark/deploy/yarn/ResourceRequestHelperSuite.scala
---
@@ -0,0 +1,255 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.deploy.yarn
+
+import org.apache.hadoop.yarn.api.records.Resource
+import org.apache.hadoop.yarn.util.Records
+import org.scalatest.{BeforeAndAfterAll, Matchers}
+
+import org.apache.spark.{SparkConf, SparkException, SparkFunSuite}
+import
org.apache.spark.deploy.yarn.ResourceRequestTestHelper.ResourceInformation
+import org.apache.spark.deploy.yarn.config.{AM_CORES, AM_MEMORY,
DRIVER_CORES, EXECUTOR_CORES, YARN_AM_RESOURCE_TYPES_PREFIX,
YARN_DRIVER_RESOURCE_TYPES_PREFIX, YARN_EXECUTOR_RESOURCE_TYPES_PREFIX}
+import org.apache.spark.internal.config.{DRIVER_MEMORY, EXECUTOR_MEMORY}
+
+class ResourceRequestHelperSuite extends SparkFunSuite with Matchers with
BeforeAndAfterAll {
+
+ private val CUSTOM_RES_1 = "custom-resource-type-1"
+ private val CUSTOM_RES_2 = "custom-resource-type-2"
+ private val MEMORY = "memory"
+ private val CORES = "cores"
+ private val NEW_CONFIG_EXECUTOR_MEMORY =
YARN_EXECUTOR_RESOURCE_TYPES_PREFIX + MEMORY
+ private val NEW_CONFIG_EXECUTOR_CORES =
YARN_EXECUTOR_RESOURCE_TYPES_PREFIX + CORES
+ private val NEW_CONFIG_AM_MEMORY = YARN_AM_RESOURCE_TYPES_PREFIX + MEMORY
+ private val NEW_CONFIG_AM_CORES = YARN_AM_RESOURCE_TYPES_PREFIX + CORES
+ private val NEW_CONFIG_DRIVER_MEMORY = YARN_DRIVER_RESOURCE_TYPES_PREFIX
+ MEMORY
+ private val NEW_CONFIG_DRIVER_CORES = YARN_DRIVER_RESOURCE_TYPES_PREFIX
+ CORES
+
+ override def beforeAll(): Unit = {
--- End diff --
This isn't doing anything.
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #20761: [SPARK-20327][CORE][YARN] Add CLI support for YAR...
Github user vanzin commented on a diff in the pull request:
https://github.com/apache/spark/pull/20761#discussion_r223440106
--- Diff:
resource-managers/yarn/src/test/scala/org/apache/spark/deploy/yarn/ResourceRequestTestHelper.scala
---
@@ -0,0 +1,90 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.deploy.yarn
+
+import scala.collection.JavaConverters._
+import scala.collection.mutable.ListBuffer
+
+import org.apache.hadoop.yarn.api.records.Resource
+
+import org.apache.spark.util.Utils
+
+object ResourceRequestTestHelper {
+ def initializeResourceTypes(resourceTypes: List[String]): Unit = {
+if (!ResourceRequestHelper.isYarnResourceTypesAvailable()) {
+ throw new IllegalStateException("initializeResourceTypes() should
not be invoked " +
+"since YARN resource types is not available because of old Hadoop
version!" )
+}
+
+val allResourceTypes = new ListBuffer[AnyRef]
+val defaultResourceTypes = List(
+ createResourceTypeInfo("memory-mb"),
+ createResourceTypeInfo("vcores"))
+val customResourceTypes = resourceTypes.map(rt =>
createResourceTypeInfo(rt))
+allResourceTypes ++= defaultResourceTypes
+allResourceTypes ++= customResourceTypes
+
+reinitializeResources(allResourceTypes)
+ }
+
+ private def createResourceTypeInfo(resourceName: String): AnyRef = {
+val resTypeInfoClass =
Utils.classForName("org.apache.hadoop.yarn.api.records.ResourceTypeInfo")
+val resTypeInfoNewInstanceMethod =
resTypeInfoClass.getMethod("newInstance", classOf[String])
+resTypeInfoNewInstanceMethod.invoke(null, resourceName)
+ }
+
+ private def reinitializeResources(resourceTypes: ListBuffer[AnyRef]):
Unit = {
+val resourceUtilsClass =
+
Utils.classForName("org.apache.hadoop.yarn.util.resource.ResourceUtils")
+val reinitializeResourcesMethod =
resourceUtilsClass.getMethod("reinitializeResources",
+ classOf[java.util.List[AnyRef]])
+reinitializeResourcesMethod.invoke(null, resourceTypes.asJava)
+ }
+
+ def getResourceTypeValue(res: Resource, name: String): AnyRef = {
+val resourceInformation: AnyRef = getResourceInformation(res, name)
--- End diff --
type is not necessary
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #20761: [SPARK-20327][CORE][YARN] Add CLI support for YAR...
Github user vanzin commented on a diff in the pull request:
https://github.com/apache/spark/pull/20761#discussion_r223436057
--- Diff:
resource-managers/yarn/src/test/scala/org/apache/spark/deploy/yarn/ClientSuite.scala
---
@@ -433,4 +465,36 @@ class ClientSuite extends SparkFunSuite with Matchers {
classpath(env)
}
+ private def testResourceRequest(expectedResources: Seq[(String, Long)],
--- End diff --
This is weird. The expected resources are a parameter, but the config is
hardcoded in this test method. That is kinda brittle.
Instead, use the `deployMode` parameter to define the expected resources.
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #20761: [SPARK-20327][CORE][YARN] Add CLI support for YAR...
Github user vanzin commented on a diff in the pull request:
https://github.com/apache/spark/pull/20761#discussion_r223433078
--- Diff:
resource-managers/yarn/src/main/scala/org/apache/spark/deploy/yarn/ResourceRequestHelper.scala
---
@@ -0,0 +1,148 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.deploy.yarn
+
+import java.lang.{Integer => JInteger, Long => JLong}
+import java.lang.reflect.InvocationTargetException
+
+import scala.collection.mutable
+import scala.util.Try
+
+import org.apache.hadoop.yarn.api.records.Resource
+
+import org.apache.spark.{SparkConf, SparkException}
+import org.apache.spark.deploy.yarn.config.{AM_CORES, AM_MEMORY,
DRIVER_CORES, EXECUTOR_CORES, YARN_AM_RESOURCE_TYPES_PREFIX,
YARN_DRIVER_RESOURCE_TYPES_PREFIX, YARN_EXECUTOR_RESOURCE_TYPES_PREFIX}
+import org.apache.spark.internal.Logging
+import org.apache.spark.internal.config.{DRIVER_MEMORY, EXECUTOR_MEMORY}
+import org.apache.spark.util.Utils
+
+/**
+ * This helper class uses some of Hadoop 3 methods from the YARN API,
+ * so we need to use reflection to avoid compile error when building
against Hadoop 2.x
+ */
+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"
+ private val ERROR_PREFIX: String = "Error:"
+
+ /**
+ * Validates sparkConf and throws a SparkException if any of standard
resources (memory or cores)
+ * is defined with the property spark.yarn.x.resource.y
+ * @param sparkConf
--- End diff --
No need to add `@param` if you're not going to write any description of it.
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #20761: [SPARK-20327][CORE][YARN] Add CLI support for YAR...
Github user vanzin commented on a diff in the pull request:
https://github.com/apache/spark/pull/20761#discussion_r223435056
--- Diff:
resource-managers/yarn/src/test/scala/org/apache/spark/deploy/yarn/ClientSuite.scala
---
@@ -35,7 +36,7 @@ import org.apache.hadoop.yarn.conf.YarnConfiguration
import org.apache.hadoop.yarn.util.Records
import org.mockito.Matchers.{eq => meq, _}
import org.mockito.Mockito._
-import org.scalatest.Matchers
+import org.scalatest.{BeforeAndAfterAll, Matchers}
--- End diff --
You're not using the added import?
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #20761: [SPARK-20327][CORE][YARN] Add CLI support for YAR...
Github user szyszy commented on a diff in the pull request:
https://github.com/apache/spark/pull/20761#discussion_r223159523
--- Diff:
resource-managers/yarn/src/test/scala/org/apache/spark/deploy/yarn/YarnAllocatorSuite.scala
---
@@ -134,6 +166,42 @@ class YarnAllocatorSuite extends SparkFunSuite with
Matchers with BeforeAndAfter
size should be (0)
}
+ test("custom resource type requested from yarn") {
+assume(ResourceRequestHelper.isYarnResourceTypesAvailable())
+TestYarnResourceRequestHelper.initializeResourceTypes(List("gpu"))
+
+// request a single container and receive it
+val handler = createAllocatorWithAdditionalConfigs(1, Map(
+ YARN_EXECUTOR_RESOURCE_TYPES_PREFIX + "gpu" -> "2G",
+ YARN_EXECUTOR_RESOURCE_TYPES_PREFIX + "memory" -> "1G"
+))
+handler.updateResourceRequests()
+handler.getNumExecutorsRunning should be (0)
+handler.getPendingAllocate.size should be (1)
+
+val resource = Resource.newInstance(3072, 6)
+val resourceTypes = Map("gpu" -> "2G")
+ResourceRequestHelper.setResourceRequests(resourceTypes, resource)
+
+val container = createContainerWithResource("host1", resource)
+handler.handleAllocatedContainers(Array(container))
+
+// verify custom resource type is part of rmClient.ask set
+val askField = rmClient.getClass.getDeclaredField("ask")
--- End diff --
Good point.
I found a better way to test what I wanted and it also covers verifying
whether the `resource` field is set correctly.
Please check the new testcase's implementation.
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #20761: [SPARK-20327][CORE][YARN] Add CLI support for YAR...
Github user szyszy commented on a diff in the pull request:
https://github.com/apache/spark/pull/20761#discussion_r223150095
--- Diff:
resource-managers/yarn/src/test/scala/org/apache/spark/deploy/yarn/TestYarnResourceRequestHelper.scala
---
@@ -0,0 +1,92 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.deploy.yarn
+
+import scala.collection.JavaConverters._
+import scala.collection.mutable.ListBuffer
+
+import org.apache.hadoop.yarn.api.records.Resource
+
+import org.apache.spark.internal.Logging
+import org.apache.spark.util.Utils
+
+object TestYarnResourceRequestHelper extends Logging {
+ def initializeResourceTypes(resourceTypes: List[String]): Unit = {
+if (!ResourceRequestHelper.isYarnResourceTypesAvailable()) {
+ throw new IllegalStateException("initializeResourceTypes() should
not be invoked " +
+"since YARN resource types is not available because of old Hadoop
version!" )
+}
+
+val allResourceTypes = new ListBuffer[AnyRef]
+val defaultResourceTypes = List(
+ createResourceTypeInfo("memory-mb"),
+ createResourceTypeInfo("vcores"))
+val customResourceTypes = resourceTypes.map(rt =>
createResourceTypeInfo(rt))
+allResourceTypes ++= defaultResourceTypes
+allResourceTypes ++= customResourceTypes
+
+reinitializeResources(allResourceTypes)
+ }
+
+ private def createResourceTypeInfo(resourceName: String): AnyRef = {
+val resTypeInfoClass =
Utils.classForName("org.apache.hadoop.yarn.api.records.ResourceTypeInfo")
+val resTypeInfoNewInstanceMethod =
resTypeInfoClass.getMethod("newInstance", classOf[String])
+resTypeInfoNewInstanceMethod.invoke(null, resourceName)
+ }
+
+ private def reinitializeResources(resourceTypes: ListBuffer[AnyRef]):
Unit = {
+val resourceUtilsClass =
+
Utils.classForName("org.apache.hadoop.yarn.util.resource.ResourceUtils")
+val reinitializeResourcesMethod =
resourceUtilsClass.getMethod("reinitializeResources",
+ classOf[java.util.List[AnyRef]])
+reinitializeResourcesMethod.invoke(null, resourceTypes.asJava)
+ }
+
+ def getResourceTypeValue(res: Resource, name: String): AnyRef = {
+val resourceInformation: AnyRef = getResourceInformation(res, name)
+invokeMethod(resourceInformation, "getValue")
+ }
+
+ def getResourceInformationByName(res: Resource, nameParam: String):
ResourceInformation = {
+val resourceInformation: AnyRef = getResourceInformation(res,
nameParam)
+val name = invokeMethod(resourceInformation,
"getName").asInstanceOf[String]
+val value = invokeMethod(resourceInformation,
"getValue").asInstanceOf[Long]
+val units = invokeMethod(resourceInformation,
"getUnits").asInstanceOf[String]
+new ResourceInformation(name, value, units)
+ }
+
+ private def getResourceInformation(res: Resource, name: String) = {
--- End diff --
Fixed with AnyRef
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #20761: [SPARK-20327][CORE][YARN] Add CLI support for YAR...
Github user szyszy commented on a diff in the pull request:
https://github.com/apache/spark/pull/20761#discussion_r223147289
--- Diff:
resource-managers/yarn/src/test/scala/org/apache/spark/deploy/yarn/ResourceRequestValidatorSuite.scala
---
@@ -0,0 +1,132 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.deploy.yarn
+
+import org.scalatest.{BeforeAndAfterAll, Matchers}
+
+import org.apache.spark.{SparkConf, SparkException, SparkFunSuite}
+
+class ResourceRequestValidatorSuite extends SparkFunSuite with Matchers
with BeforeAndAfterAll {
+
+ override def beforeAll(): Unit = {
+super.beforeAll()
+ }
+
+ test("empty SparkConf should be valid") {
+val sparkConf = new SparkConf()
+ResourceRequestValidator.validateResources(sparkConf)
+ }
+
+ test("just normal resources are defined") {
+val sparkConf = new SparkConf()
+sparkConf.set("spark.driver.memory", "3G")
+sparkConf.set("spark.driver.cores", "4")
+sparkConf.set("spark.executor.memory", "4G")
+sparkConf.set("spark.executor.cores", "2")
+
+ResourceRequestValidator.validateResources(sparkConf)
+ }
+
+ test("Memory defined with new config for executor") {
+val sparkConf = new SparkConf()
+sparkConf.set("spark.yarn.executor.resource.memory", "30G")
+
+val thrown = intercept[SparkException] {
+ ResourceRequestValidator.validateResources(sparkConf)
+}
+thrown.getMessage should include
("spark.yarn.executor.resource.memory")
+ }
+
+ test("Cores defined with new config for executor") {
+val sparkConf = new SparkConf()
+sparkConf.set("spark.yarn.executor.resource.cores", "5")
+
+val thrown = intercept[SparkException] {
+ ResourceRequestValidator.validateResources(sparkConf)
+}
+thrown.getMessage should include ("spark.yarn.executor.resource.cores")
+ }
+
+ test("Memory defined with new config, client mode") {
+val sparkConf = new SparkConf()
+sparkConf.set("spark.yarn.am.resource.memory", "1G")
+
+val thrown = intercept[SparkException] {
+ ResourceRequestValidator.validateResources(sparkConf)
+}
+thrown.getMessage should include ("spark.yarn.am.resource.memory")
+ }
+
+ test("Memory defined with new config for driver, cluster mode") {
+val sparkConf = new SparkConf()
+sparkConf.set("spark.yarn.driver.resource.memory", "1G")
+
+val thrown = intercept[SparkException] {
+ ResourceRequestValidator.validateResources(sparkConf)
+}
+thrown.getMessage should include ("spark.yarn.driver.resource.memory")
+ }
+
+ test("Cores defined with new config, client mode") {
+val sparkConf = new SparkConf()
+sparkConf.set("spark.driver.memory", "2G")
+sparkConf.set("spark.driver.cores", "4")
+sparkConf.set("spark.executor.memory", "4G")
+sparkConf.set("spark.yarn.am.cores", "2")
+
+sparkConf.set("spark.yarn.am.resource.cores", "3")
+
+val thrown = intercept[SparkException] {
+ ResourceRequestValidator.validateResources(sparkConf)
+}
+thrown.getMessage should include ("spark.yarn.am.resource.cores")
+ }
+
+ test("Cores defined with new config for driver, cluster mode") {
+val sparkConf = new SparkConf()
+sparkConf.set("spark.yarn.driver.resource.cores", "1G")
+
+val thrown = intercept[SparkException] {
+ ResourceRequestValidator.validateResources(sparkConf)
+}
+thrown.getMessage should include ("spark.yarn.driver.resource.cores")
+ }
+
+ test("various duplicated definitions") {
+val sparkConf = new SparkConf()
+sparkConf.set("spark.driver.memory", "2G")
--- End diff --
Fixed, using the constants all over this test class from now
---
-
To
[GitHub] spark pull request #20761: [SPARK-20327][CORE][YARN] Add CLI support for YAR...
Github user szyszy commented on a diff in the pull request:
https://github.com/apache/spark/pull/20761#discussion_r223142574
--- Diff:
resource-managers/yarn/src/test/scala/org/apache/spark/deploy/yarn/TestYarnResourceTypeHelper.scala
---
@@ -0,0 +1,93 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.deploy.yarn
+
+import scala.collection.JavaConverters._
+import scala.collection.mutable.ListBuffer
+
+import org.apache.hadoop.yarn.api.records.Resource
+import org.junit.Assert
+
+import org.apache.spark.internal.Logging
+import org.apache.spark.util.Utils
+
+object TestYarnResourceTypeHelper extends Logging {
--- End diff --
Renamed as you suggested in a later comment.
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #20761: [SPARK-20327][CORE][YARN] Add CLI support for YAR...
Github user szyszy commented on a diff in the pull request:
https://github.com/apache/spark/pull/20761#discussion_r223142490
--- Diff:
resource-managers/yarn/src/test/scala/org/apache/spark/deploy/yarn/ResourceTypeValidatorSuite.scala
---
@@ -0,0 +1,150 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.deploy.yarn
+
+import org.scalatest.{BeforeAndAfterAll, Matchers}
+
+import org.apache.spark.{SparkConf, SparkException, SparkFunSuite}
+
+
+class ResourceTypeValidatorSuite extends SparkFunSuite with Matchers with
BeforeAndAfterAll {
+
+ override def beforeAll(): Unit = {
+super.beforeAll()
+ }
+
+ private def getExpectedErrorMessage(
--- End diff --
Fixed
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #20761: [SPARK-20327][CORE][YARN] Add CLI support for YAR...
Github user szyszy commented on a diff in the pull request:
https://github.com/apache/spark/pull/20761#discussion_r223139838
--- Diff:
resource-managers/yarn/src/test/scala/org/apache/spark/deploy/yarn/ResourceRequestTestHelper.scala
---
@@ -0,0 +1,95 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.deploy.yarn
+
+import scala.collection.JavaConverters._
+import scala.collection.mutable.ListBuffer
+
+import org.apache.hadoop.yarn.api.records.Resource
+
+import org.apache.spark.util.Utils
+
+object ResourceRequestTestHelper {
+ def initializeResourceTypes(resourceTypes: List[String]): Unit = {
+if (!ResourceRequestHelper.isYarnResourceTypesAvailable()) {
+ throw new IllegalStateException("initializeResourceTypes() should
not be invoked " +
+"since YARN resource types is not available because of old Hadoop
version!" )
+}
+
+val allResourceTypes = new ListBuffer[AnyRef]
+val defaultResourceTypes = List(
+ createResourceTypeInfo("memory-mb"),
+ createResourceTypeInfo("vcores"))
+val customResourceTypes = resourceTypes.map(rt =>
createResourceTypeInfo(rt))
+allResourceTypes ++= defaultResourceTypes
+allResourceTypes ++= customResourceTypes
+
+reinitializeResources(allResourceTypes)
+ }
+
+ private def createResourceTypeInfo(resourceName: String): AnyRef = {
+val resTypeInfoClass =
Utils.classForName("org.apache.hadoop.yarn.api.records.ResourceTypeInfo")
+val resTypeInfoNewInstanceMethod =
resTypeInfoClass.getMethod("newInstance", classOf[String])
+resTypeInfoNewInstanceMethod.invoke(null, resourceName)
+ }
+
+ private def reinitializeResources(resourceTypes: ListBuffer[AnyRef]):
Unit = {
+val resourceUtilsClass =
+
Utils.classForName("org.apache.hadoop.yarn.util.resource.ResourceUtils")
+val reinitializeResourcesMethod =
resourceUtilsClass.getMethod("reinitializeResources",
+ classOf[java.util.List[AnyRef]])
+reinitializeResourcesMethod.invoke(null, resourceTypes.asJava)
+ }
+
+ def getResourceTypeValue(res: Resource, name: String): AnyRef = {
+val resourceInformation: AnyRef = getResourceInformation(res, name)
+invokeMethod(resourceInformation, "getValue")
+ }
+
+ def getResourceInformationByName(res: Resource, nameParam: String):
ResourceInformation = {
+val resourceInformation: AnyRef = getResourceInformation(res,
nameParam)
+val name = invokeMethod(resourceInformation,
"getName").asInstanceOf[String]
+val value = invokeMethod(resourceInformation,
"getValue").asInstanceOf[Long]
+val units = invokeMethod(resourceInformation,
"getUnits").asInstanceOf[String]
+ResourceInformation(name, value, units)
+ }
+
+ private def getResourceInformation(res: Resource, name: String) = {
+if (!ResourceRequestHelper.isYarnResourceTypesAvailable()) {
+ throw new IllegalStateException("assertResourceTypeValue() should
not be invoked " +
+"since yarn resource types is not available because of old Hadoop
version!")
+}
+
+val getResourceInformationMethod =
res.getClass.getMethod("getResourceInformation",
+ classOf[String])
+var yarnResourceName = name
+if (name == "memory") {
+ yarnResourceName = "memory-mb"
+}
+val resourceInformation = getResourceInformationMethod.invoke(res,
yarnResourceName)
+resourceInformation
+ }
+
+ private def invokeMethod(resourceInformation: AnyRef, methodName:
String): AnyRef = {
+val getValueMethod = resourceInformation.getClass.getMethod(methodName)
+getValueMethod.invoke(resourceInformation)
+ }
+
+ case class ResourceInformation(name: String, value: Long, units: String)
{
--- End diff --
fixed
---
-
To unsubscribe, e-mail:
[GitHub] spark pull request #20761: [SPARK-20327][CORE][YARN] Add CLI support for YAR...
Github user szyszy commented on a diff in the pull request:
https://github.com/apache/spark/pull/20761#discussion_r223139747
--- Diff:
resource-managers/yarn/src/test/scala/org/apache/spark/deploy/yarn/YarnAllocatorSuite.scala
---
@@ -63,6 +63,10 @@ class YarnAllocatorSuite extends SparkFunSuite with
Matchers with BeforeAndAfter
var containerNum = 0
+ override def beforeAll(): Unit = {
--- End diff --
removed
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #20761: [SPARK-20327][CORE][YARN] Add CLI support for YAR...
Github user szyszy commented on a diff in the pull request:
https://github.com/apache/spark/pull/20761#discussion_r223139279
--- Diff:
resource-managers/yarn/src/main/scala/org/apache/spark/deploy/yarn/ResourceTypeHelper.scala
---
@@ -0,0 +1,128 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.deploy.yarn
+
+import java.lang.reflect.InvocationTargetException
+
+import scala.util.control.NonFatal
+
+import org.apache.hadoop.yarn.api.records.Resource
+
+import org.apache.spark.internal.Logging
+import org.apache.spark.util.Utils
+
+private[yarn] class ResourceTypeHelper {
+}
+
+object ResourceTypeHelper extends Logging {
+ private val resourceTypesNotAvailableErrorMessage =
+"Ignoring updating resource with resource types because " +
+"the version of YARN does not support it!"
+
+ def setResourceInfoFromResourceTypes(resourceTypesParam: Map[String,
String],
+ resource: Resource): Resource = {
+if (resource == null) {
+ throw new IllegalArgumentException("Resource parameter should not be
null!")
+}
+
+if (!ResourceTypeHelper.isYarnResourceTypesAvailable()) {
+ logWarning(resourceTypesNotAvailableErrorMessage)
+ return resource
+}
+
+val resourceTypes = resourceTypesParam.map { case (k, v) => (
+ if (k.equals("memory")) {
--- End diff --
This was fixed before.
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #20761: [SPARK-20327][CORE][YARN] Add CLI support for YAR...
Github user szyszy commented on a diff in the pull request:
https://github.com/apache/spark/pull/20761#discussion_r223139202
--- Diff:
resource-managers/yarn/src/test/scala/org/apache/spark/deploy/yarn/ClientSuite.scala
---
@@ -199,6 +200,92 @@ class ClientSuite extends SparkFunSuite with Matchers {
appContext.getMaxAppAttempts should be (42)
}
+ test("Resource type args propagate, resource type not defined") {
+assume(ResourceRequestHelper.isYarnResourceTypesAvailable())
+val sparkConf = new SparkConf()
+ .set(YARN_AM_RESOURCE_TYPES_PREFIX + "some_resource_with_units_1",
"121m")
+val args = new ClientArguments(Array())
+
+val appContext =
Records.newRecord(classOf[ApplicationSubmissionContext])
+val getNewApplicationResponse =
Records.newRecord(classOf[GetNewApplicationResponse])
+val containerLaunchContext =
Records.newRecord(classOf[ContainerLaunchContext])
+
+val client = new Client(args, sparkConf)
+
+try {
+ client.createApplicationSubmissionContext(
+new YarnClientApplication(getNewApplicationResponse, appContext),
+containerLaunchContext)
+} catch {
+ case NonFatal(e) =>
+val expectedExceptionClass =
"org.apache.hadoop.yarn.exceptions.ResourceNotFoundException"
+if (e.getClass.getName != expectedExceptionClass) {
+ fail(s"Exception caught: $e is not an instance of
$expectedExceptionClass!")
+}
+}
+ }
+
+ test("Resource type args propagate (client mode)") {
+assume(ResourceRequestHelper.isYarnResourceTypesAvailable())
+TestYarnResourceRequestHelper.initializeResourceTypes(List("gpu",
"fpga"))
+
+val sparkConf = new SparkConf()
+ .set(YARN_AM_RESOURCE_TYPES_PREFIX + "some_resource_with_units_1",
"121m")
--- End diff --
Nevermind what I commented about memory-mb. Fixed with latest commit pushed
moments ago. The translation was only required because the test was wrong. We
agreed on before that we don't need any translation in spark code from memory
to memory-mb as memory is a standard resource and should be used as it was used
to, with the standard spark config.
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #20761: [SPARK-20327][CORE][YARN] Add CLI support for YAR...
Github user vanzin commented on a diff in the pull request:
https://github.com/apache/spark/pull/20761#discussion_r223065758
--- Diff:
resource-managers/yarn/src/test/scala/org/apache/spark/deploy/yarn/ResourceRequestTestHelper.scala
---
@@ -0,0 +1,95 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.deploy.yarn
+
+import scala.collection.JavaConverters._
+import scala.collection.mutable.ListBuffer
+
+import org.apache.hadoop.yarn.api.records.Resource
+
+import org.apache.spark.util.Utils
+
+object ResourceRequestTestHelper {
+ def initializeResourceTypes(resourceTypes: List[String]): Unit = {
+if (!ResourceRequestHelper.isYarnResourceTypesAvailable()) {
+ throw new IllegalStateException("initializeResourceTypes() should
not be invoked " +
+"since YARN resource types is not available because of old Hadoop
version!" )
+}
+
+val allResourceTypes = new ListBuffer[AnyRef]
+val defaultResourceTypes = List(
+ createResourceTypeInfo("memory-mb"),
+ createResourceTypeInfo("vcores"))
+val customResourceTypes = resourceTypes.map(rt =>
createResourceTypeInfo(rt))
+allResourceTypes ++= defaultResourceTypes
+allResourceTypes ++= customResourceTypes
+
+reinitializeResources(allResourceTypes)
+ }
+
+ private def createResourceTypeInfo(resourceName: String): AnyRef = {
+val resTypeInfoClass =
Utils.classForName("org.apache.hadoop.yarn.api.records.ResourceTypeInfo")
+val resTypeInfoNewInstanceMethod =
resTypeInfoClass.getMethod("newInstance", classOf[String])
+resTypeInfoNewInstanceMethod.invoke(null, resourceName)
+ }
+
+ private def reinitializeResources(resourceTypes: ListBuffer[AnyRef]):
Unit = {
+val resourceUtilsClass =
+
Utils.classForName("org.apache.hadoop.yarn.util.resource.ResourceUtils")
+val reinitializeResourcesMethod =
resourceUtilsClass.getMethod("reinitializeResources",
+ classOf[java.util.List[AnyRef]])
+reinitializeResourcesMethod.invoke(null, resourceTypes.asJava)
+ }
+
+ def getResourceTypeValue(res: Resource, name: String): AnyRef = {
+val resourceInformation: AnyRef = getResourceInformation(res, name)
+invokeMethod(resourceInformation, "getValue")
+ }
+
+ def getResourceInformationByName(res: Resource, nameParam: String):
ResourceInformation = {
+val resourceInformation: AnyRef = getResourceInformation(res,
nameParam)
+val name = invokeMethod(resourceInformation,
"getName").asInstanceOf[String]
+val value = invokeMethod(resourceInformation,
"getValue").asInstanceOf[Long]
+val units = invokeMethod(resourceInformation,
"getUnits").asInstanceOf[String]
+ResourceInformation(name, value, units)
+ }
+
+ private def getResourceInformation(res: Resource, name: String) = {
+if (!ResourceRequestHelper.isYarnResourceTypesAvailable()) {
+ throw new IllegalStateException("assertResourceTypeValue() should
not be invoked " +
+"since yarn resource types is not available because of old Hadoop
version!")
+}
+
+val getResourceInformationMethod =
res.getClass.getMethod("getResourceInformation",
+ classOf[String])
+var yarnResourceName = name
+if (name == "memory") {
+ yarnResourceName = "memory-mb"
+}
+val resourceInformation = getResourceInformationMethod.invoke(res,
yarnResourceName)
+resourceInformation
+ }
+
+ private def invokeMethod(resourceInformation: AnyRef, methodName:
String): AnyRef = {
+val getValueMethod = resourceInformation.getClass.getMethod(methodName)
+getValueMethod.invoke(resourceInformation)
+ }
+
+ case class ResourceInformation(name: String, value: Long, units: String)
{
--- End diff --
Drop the braces.
---
-
To unsubscribe, e-mail:
[GitHub] spark pull request #20761: [SPARK-20327][CORE][YARN] Add CLI support for YAR...
Github user vanzin commented on a diff in the pull request:
https://github.com/apache/spark/pull/20761#discussion_r223065648
--- Diff:
resource-managers/yarn/src/test/scala/org/apache/spark/deploy/yarn/YarnAllocatorSuite.scala
---
@@ -63,6 +63,10 @@ class YarnAllocatorSuite extends SparkFunSuite with
Matchers with BeforeAndAfter
var containerNum = 0
+ override def beforeAll(): Unit = {
--- End diff --
This isn't doing anything useful.
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #20761: [SPARK-20327][CORE][YARN] Add CLI support for YAR...
Github user vanzin commented on a diff in the pull request:
https://github.com/apache/spark/pull/20761#discussion_r223061049
--- Diff:
resource-managers/yarn/src/test/scala/org/apache/spark/deploy/yarn/TestYarnResourceRequestHelper.scala
---
@@ -0,0 +1,92 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.deploy.yarn
+
+import scala.collection.JavaConverters._
+import scala.collection.mutable.ListBuffer
+
+import org.apache.hadoop.yarn.api.records.Resource
+
+import org.apache.spark.internal.Logging
+import org.apache.spark.util.Utils
+
+object TestYarnResourceRequestHelper extends Logging {
+ def initializeResourceTypes(resourceTypes: List[String]): Unit = {
+if (!ResourceRequestHelper.isYarnResourceTypesAvailable()) {
+ throw new IllegalStateException("initializeResourceTypes() should
not be invoked " +
+"since YARN resource types is not available because of old Hadoop
version!" )
+}
+
+val allResourceTypes = new ListBuffer[AnyRef]
+val defaultResourceTypes = List(
+ createResourceTypeInfo("memory-mb"),
+ createResourceTypeInfo("vcores"))
+val customResourceTypes = resourceTypes.map(rt =>
createResourceTypeInfo(rt))
+allResourceTypes ++= defaultResourceTypes
+allResourceTypes ++= customResourceTypes
+
+reinitializeResources(allResourceTypes)
+ }
+
+ private def createResourceTypeInfo(resourceName: String): AnyRef = {
+val resTypeInfoClass =
Utils.classForName("org.apache.hadoop.yarn.api.records.ResourceTypeInfo")
+val resTypeInfoNewInstanceMethod =
resTypeInfoClass.getMethod("newInstance", classOf[String])
+resTypeInfoNewInstanceMethod.invoke(null, resourceName)
+ }
+
+ private def reinitializeResources(resourceTypes: ListBuffer[AnyRef]):
Unit = {
+val resourceUtilsClass =
+
Utils.classForName("org.apache.hadoop.yarn.util.resource.ResourceUtils")
+val reinitializeResourcesMethod =
resourceUtilsClass.getMethod("reinitializeResources",
+ classOf[java.util.List[AnyRef]])
+reinitializeResourcesMethod.invoke(null, resourceTypes.asJava)
+ }
+
+ def getResourceTypeValue(res: Resource, name: String): AnyRef = {
+val resourceInformation: AnyRef = getResourceInformation(res, name)
+invokeMethod(resourceInformation, "getValue")
+ }
+
+ def getResourceInformationByName(res: Resource, nameParam: String):
ResourceInformation = {
+val resourceInformation: AnyRef = getResourceInformation(res,
nameParam)
+val name = invokeMethod(resourceInformation,
"getName").asInstanceOf[String]
+val value = invokeMethod(resourceInformation,
"getValue").asInstanceOf[Long]
+val units = invokeMethod(resourceInformation,
"getUnits").asInstanceOf[String]
+new ResourceInformation(name, value, units)
+ }
+
+ private def getResourceInformation(res: Resource, name: String) = {
--- End diff --
`Any` or `AnyRef` like you're doing in a bunch o other places.
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #20761: [SPARK-20327][CORE][YARN] Add CLI support for YAR...
Github user vanzin commented on a diff in the pull request:
https://github.com/apache/spark/pull/20761#discussion_r223059478
--- Diff:
resource-managers/yarn/src/test/scala/org/apache/spark/deploy/yarn/ResourceRequestValidatorSuite.scala
---
@@ -0,0 +1,132 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.deploy.yarn
+
+import org.scalatest.{BeforeAndAfterAll, Matchers}
+
+import org.apache.spark.{SparkConf, SparkException, SparkFunSuite}
+
+class ResourceRequestValidatorSuite extends SparkFunSuite with Matchers
with BeforeAndAfterAll {
+
+ override def beforeAll(): Unit = {
+super.beforeAll()
+ }
+
+ test("empty SparkConf should be valid") {
+val sparkConf = new SparkConf()
+ResourceRequestValidator.validateResources(sparkConf)
+ }
+
+ test("just normal resources are defined") {
+val sparkConf = new SparkConf()
+sparkConf.set("spark.driver.memory", "3G")
+sparkConf.set("spark.driver.cores", "4")
+sparkConf.set("spark.executor.memory", "4G")
+sparkConf.set("spark.executor.cores", "2")
+
+ResourceRequestValidator.validateResources(sparkConf)
+ }
+
+ test("Memory defined with new config for executor") {
+val sparkConf = new SparkConf()
+sparkConf.set("spark.yarn.executor.resource.memory", "30G")
+
+val thrown = intercept[SparkException] {
+ ResourceRequestValidator.validateResources(sparkConf)
+}
+thrown.getMessage should include
("spark.yarn.executor.resource.memory")
+ }
+
+ test("Cores defined with new config for executor") {
+val sparkConf = new SparkConf()
+sparkConf.set("spark.yarn.executor.resource.cores", "5")
+
+val thrown = intercept[SparkException] {
+ ResourceRequestValidator.validateResources(sparkConf)
+}
+thrown.getMessage should include ("spark.yarn.executor.resource.cores")
+ }
+
+ test("Memory defined with new config, client mode") {
+val sparkConf = new SparkConf()
+sparkConf.set("spark.yarn.am.resource.memory", "1G")
+
+val thrown = intercept[SparkException] {
+ ResourceRequestValidator.validateResources(sparkConf)
+}
+thrown.getMessage should include ("spark.yarn.am.resource.memory")
+ }
+
+ test("Memory defined with new config for driver, cluster mode") {
+val sparkConf = new SparkConf()
+sparkConf.set("spark.yarn.driver.resource.memory", "1G")
+
+val thrown = intercept[SparkException] {
+ ResourceRequestValidator.validateResources(sparkConf)
+}
+thrown.getMessage should include ("spark.yarn.driver.resource.memory")
+ }
+
+ test("Cores defined with new config, client mode") {
+val sparkConf = new SparkConf()
+sparkConf.set("spark.driver.memory", "2G")
+sparkConf.set("spark.driver.cores", "4")
+sparkConf.set("spark.executor.memory", "4G")
+sparkConf.set("spark.yarn.am.cores", "2")
+
+sparkConf.set("spark.yarn.am.resource.cores", "3")
+
+val thrown = intercept[SparkException] {
+ ResourceRequestValidator.validateResources(sparkConf)
+}
+thrown.getMessage should include ("spark.yarn.am.resource.cores")
+ }
+
+ test("Cores defined with new config for driver, cluster mode") {
+val sparkConf = new SparkConf()
+sparkConf.set("spark.yarn.driver.resource.cores", "1G")
+
+val thrown = intercept[SparkException] {
+ ResourceRequestValidator.validateResources(sparkConf)
+}
+thrown.getMessage should include ("spark.yarn.driver.resource.cores")
+ }
+
+ test("various duplicated definitions") {
+val sparkConf = new SparkConf()
+sparkConf.set("spark.driver.memory", "2G")
--- End diff --
Config constants haven't existed forever. There's a lot of code written
before them, and it makes no sense to go and change all of those
[GitHub] spark pull request #20761: [SPARK-20327][CORE][YARN] Add CLI support for YAR...
Github user tgravescs commented on a diff in the pull request:
https://github.com/apache/spark/pull/20761#discussion_r223023894
--- Diff: docs/running-on-yarn.md ---
@@ -121,6 +121,43 @@ To use a custom metrics.properties for the application
master and executors, upd
Use lower-case suffixes, e.g. k, m,
g, t, and p, for kibi-, mebi-, gibi-,
tebi-, and pebibytes, respectively.
+
+ spark.yarn.am.resource.{resource-type}
+ (none)
+
+Amount of resource to use for the YARN Application Master in client
mode.
+In cluster mode, use
spark.yarn.driver.resource.resource-type 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
+
+Example:
+To request GPU resources from YARN, use:
spark.yarn.am.resource.yarn.io/gpu
+
+
+
+ spark.yarn.driver.resource.{resource-type}
+ (none)
+
+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
+
+Example:
+To request GPU resources from YARN, use:
spark.yarn.driver.resource.yarn.io/gpu
+
+
+
+ spark.yarn.executor.resource.{resource-type}
--- End diff --
sounds good.
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #20761: [SPARK-20327][CORE][YARN] Add CLI support for YAR...
Github user szyszy commented on a diff in the pull request:
https://github.com/apache/spark/pull/20761#discussion_r222897804
--- Diff:
resource-managers/yarn/src/test/scala/org/apache/spark/deploy/yarn/YarnAllocatorSuite.scala
---
@@ -134,6 +166,42 @@ class YarnAllocatorSuite extends SparkFunSuite with
Matchers with BeforeAndAfter
size should be (0)
}
+ test("custom resource type requested from yarn") {
+assume(ResourceRequestHelper.isYarnResourceTypesAvailable())
+TestYarnResourceRequestHelper.initializeResourceTypes(List("gpu"))
+
+// request a single container and receive it
+val handler = createAllocatorWithAdditionalConfigs(1, Map(
+ YARN_EXECUTOR_RESOURCE_TYPES_PREFIX + "gpu" -> "2G",
+ YARN_EXECUTOR_RESOURCE_TYPES_PREFIX + "memory" -> "1G"
+))
+handler.updateResourceRequests()
+handler.getNumExecutorsRunning should be (0)
+handler.getPendingAllocate.size should be (1)
+
+val resource = Resource.newInstance(3072, 6)
+val resourceTypes = Map("gpu" -> "2G")
+ResourceRequestHelper.setResourceRequests(resourceTypes, resource)
+
+val container = createContainerWithResource("host1", resource)
+handler.handleAllocatedContainers(Array(container))
+
+// verify custom resource type is part of rmClient.ask set
+val askField = rmClient.getClass.getDeclaredField("ask")
--- End diff --
Will look into this tomorrow morning
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #20761: [SPARK-20327][CORE][YARN] Add CLI support for YAR...
Github user szyszy commented on a diff in the pull request:
https://github.com/apache/spark/pull/20761#discussion_r222895698
--- Diff:
resource-managers/yarn/src/test/scala/org/apache/spark/deploy/yarn/ClientSuite.scala
---
@@ -199,6 +200,92 @@ class ClientSuite extends SparkFunSuite with Matchers {
appContext.getMaxAppAttempts should be (42)
}
+ test("Resource type args propagate, resource type not defined") {
+assume(ResourceRequestHelper.isYarnResourceTypesAvailable())
+val sparkConf = new SparkConf()
+ .set(YARN_AM_RESOURCE_TYPES_PREFIX + "some_resource_with_units_1",
"121m")
+val args = new ClientArguments(Array())
+
+val appContext =
Records.newRecord(classOf[ApplicationSubmissionContext])
+val getNewApplicationResponse =
Records.newRecord(classOf[GetNewApplicationResponse])
+val containerLaunchContext =
Records.newRecord(classOf[ContainerLaunchContext])
+
+val client = new Client(args, sparkConf)
+
+try {
+ client.createApplicationSubmissionContext(
+new YarnClientApplication(getNewApplicationResponse, appContext),
+containerLaunchContext)
+} catch {
+ case NonFatal(e) =>
+val expectedExceptionClass =
"org.apache.hadoop.yarn.exceptions.ResourceNotFoundException"
+if (e.getClass.getName != expectedExceptionClass) {
+ fail(s"Exception caught: $e is not an instance of
$expectedExceptionClass!")
+}
+}
+ }
+
+ test("Resource type args propagate (client mode)") {
+assume(ResourceRequestHelper.isYarnResourceTypesAvailable())
+TestYarnResourceRequestHelper.initializeResourceTypes(List("gpu",
"fpga"))
+
+val sparkConf = new SparkConf()
+ .set(YARN_AM_RESOURCE_TYPES_PREFIX + "some_resource_with_units_1",
"121m")
+ .set(YARN_DRIVER_RESOURCE_TYPES_PREFIX +
"some_resource_with_units_1", "122m")
+ .set(YARN_AM_RESOURCE_TYPES_PREFIX + "fpga", "222m")
+ .set(YARN_DRIVER_RESOURCE_TYPES_PREFIX + "fpga", "223m")
+ .set(YARN_AM_RESOURCE_TYPES_PREFIX + "memory", "1G")
+ .set(YARN_DRIVER_RESOURCE_TYPES_PREFIX + "memory", "2G")
+val args = new ClientArguments(Array())
+
+val appContext =
Records.newRecord(classOf[ApplicationSubmissionContext])
+val getNewApplicationResponse =
Records.newRecord(classOf[GetNewApplicationResponse])
+val containerLaunchContext =
Records.newRecord(classOf[ContainerLaunchContext])
+
+val client = new Client(args, sparkConf)
+client.createApplicationSubmissionContext(
+ new YarnClientApplication(getNewApplicationResponse, appContext),
+ containerLaunchContext)
+
+appContext.getAMContainerSpec should be (containerLaunchContext)
+appContext.getApplicationType should be ("SPARK")
+
TestYarnResourceRequestHelper.getResourceTypeValue(appContext.getResource,
+ "some_resource_with_units_1") should be (121)
+TestYarnResourceRequestHelper
+.getResourceTypeValue(appContext.getResource, "fpga") should be
(222)
+ }
+
+ test("configuration and resource type args propagate (cluster mode)") {
--- End diff --
Indeed, extracted a common method, passing the expected values as a
sequence of tuples.
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #20761: [SPARK-20327][CORE][YARN] Add CLI support for YAR...
Github user szyszy commented on a diff in the pull request:
https://github.com/apache/spark/pull/20761#discussion_r222894767
--- Diff:
resource-managers/yarn/src/test/scala/org/apache/spark/deploy/yarn/ClientSuite.scala
---
@@ -199,6 +200,92 @@ class ClientSuite extends SparkFunSuite with Matchers {
appContext.getMaxAppAttempts should be (42)
}
+ test("Resource type args propagate, resource type not defined") {
+assume(ResourceRequestHelper.isYarnResourceTypesAvailable())
+val sparkConf = new SparkConf()
+ .set(YARN_AM_RESOURCE_TYPES_PREFIX + "some_resource_with_units_1",
"121m")
+val args = new ClientArguments(Array())
+
+val appContext =
Records.newRecord(classOf[ApplicationSubmissionContext])
+val getNewApplicationResponse =
Records.newRecord(classOf[GetNewApplicationResponse])
+val containerLaunchContext =
Records.newRecord(classOf[ContainerLaunchContext])
+
+val client = new Client(args, sparkConf)
+
+try {
+ client.createApplicationSubmissionContext(
+new YarnClientApplication(getNewApplicationResponse, appContext),
+containerLaunchContext)
+} catch {
+ case NonFatal(e) =>
+val expectedExceptionClass =
"org.apache.hadoop.yarn.exceptions.ResourceNotFoundException"
+if (e.getClass.getName != expectedExceptionClass) {
+ fail(s"Exception caught: $e is not an instance of
$expectedExceptionClass!")
+}
+}
+ }
+
+ test("Resource type args propagate (client mode)") {
+assume(ResourceRequestHelper.isYarnResourceTypesAvailable())
+TestYarnResourceRequestHelper.initializeResourceTypes(List("gpu",
"fpga"))
+
+val sparkConf = new SparkConf()
+ .set(YARN_AM_RESOURCE_TYPES_PREFIX + "some_resource_with_units_1",
"121m")
+ .set(YARN_DRIVER_RESOURCE_TYPES_PREFIX +
"some_resource_with_units_1", "122m")
+ .set(YARN_AM_RESOURCE_TYPES_PREFIX + "fpga", "222m")
+ .set(YARN_DRIVER_RESOURCE_TYPES_PREFIX + "fpga", "223m")
+ .set(YARN_AM_RESOURCE_TYPES_PREFIX + "memory", "1G")
--- End diff --
exactly, see my comment for the other testcase.
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #20761: [SPARK-20327][CORE][YARN] Add CLI support for YAR...
Github user szyszy commented on a diff in the pull request:
https://github.com/apache/spark/pull/20761#discussion_r222894728
--- Diff:
resource-managers/yarn/src/test/scala/org/apache/spark/deploy/yarn/ClientSuite.scala
---
@@ -199,6 +200,92 @@ class ClientSuite extends SparkFunSuite with Matchers {
appContext.getMaxAppAttempts should be (42)
}
+ test("Resource type args propagate, resource type not defined") {
+assume(ResourceRequestHelper.isYarnResourceTypesAvailable())
+val sparkConf = new SparkConf()
+ .set(YARN_AM_RESOURCE_TYPES_PREFIX + "some_resource_with_units_1",
"121m")
+val args = new ClientArguments(Array())
+
+val appContext =
Records.newRecord(classOf[ApplicationSubmissionContext])
+val getNewApplicationResponse =
Records.newRecord(classOf[GetNewApplicationResponse])
+val containerLaunchContext =
Records.newRecord(classOf[ContainerLaunchContext])
+
+val client = new Client(args, sparkConf)
+
+try {
+ client.createApplicationSubmissionContext(
+new YarnClientApplication(getNewApplicationResponse, appContext),
+containerLaunchContext)
+} catch {
+ case NonFatal(e) =>
+val expectedExceptionClass =
"org.apache.hadoop.yarn.exceptions.ResourceNotFoundException"
+if (e.getClass.getName != expectedExceptionClass) {
+ fail(s"Exception caught: $e is not an instance of
$expectedExceptionClass!")
+}
+}
+ }
+
+ test("Resource type args propagate (client mode)") {
+assume(ResourceRequestHelper.isYarnResourceTypesAvailable())
+TestYarnResourceRequestHelper.initializeResourceTypes(List("gpu",
"fpga"))
+
+val sparkConf = new SparkConf()
+ .set(YARN_AM_RESOURCE_TYPES_PREFIX + "some_resource_with_units_1",
"121m")
--- End diff --
Actually, these two tests were indeed failing. Forgot to run them lately
and jenkins does not run them because the Hadoop version is not > 3.1
This also pointed to another issue: in
`ResourceRequestHelper.setResourceRequests`, the `setResourceInformationMethod`
was invoked with value `"memory` which is not accepted by the YARN API so I had
to translate it to `"memory-mb"` in order to work correctly.
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #20761: [SPARK-20327][CORE][YARN] Add CLI support for YAR...
Github user szyszy commented on a diff in the pull request:
https://github.com/apache/spark/pull/20761#discussion_r222890990
--- Diff:
resource-managers/yarn/src/main/scala/org/apache/spark/deploy/yarn/ResourceRequestHelper.scala
---
@@ -0,0 +1,106 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.deploy.yarn
+
+import java.lang.{Integer => JInteger, Long => JLong}
+import java.lang.reflect.InvocationTargetException
+
+import scala.util.Try
+
+import org.apache.hadoop.yarn.api.records.Resource
+
+import org.apache.spark.internal.Logging
+import org.apache.spark.util.Utils
+
+/**
+ * This helper class uses some of Hadoop 3 methods from the YARN API,
+ * so we need to use reflection to avoid compile error when building
against Hadoop 2.x
+ */
+private object ResourceRequestHelper extends Logging {
+ private val AMOUNT_AND_UNIT_REGEX = "([0-9]+)([A-Za-z]*)".r
+
+ def setResourceRequests(
--- End diff --
added doc
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
