You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by sr...@apache.org on 2016/02/24 13:15:14 UTC
spark git commit: [SPARK-12759][Core][Spark should fail fast if
--executor-memory is too small for spark to start]
Repository: spark
Updated Branches:
refs/heads/master 831429170 -> bcfd55fa9
[SPARK-12759][Core][Spark should fail fast if --executor-memory is too small for spark to start]
Added an exception to be thrown in UnifiedMemoryManager.scala if the configuration given for executor memory is too low. Also modified the exception message thrown when driver memory is too low.
This patch was tested manually by passing in config options to Spark shell. I also added a test in UnifiedMemoryManagerSuite.scala
Author: Daniel Jalova <dj...@us.ibm.com>
Closes #11255 from djalova/SPARK-12759.
Project: http://git-wip-us.apache.org/repos/asf/spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/bcfd55fa
Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/bcfd55fa
Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/bcfd55fa
Branch: refs/heads/master
Commit: bcfd55fa982b24184c07fcd4ccdd55dcf6465bf4
Parents: 8314291
Author: Daniel Jalova <dj...@us.ibm.com>
Authored: Wed Feb 24 12:15:11 2016 +0000
Committer: Sean Owen <so...@cloudera.com>
Committed: Wed Feb 24 12:15:11 2016 +0000
----------------------------------------------------------------------
.../spark/memory/UnifiedMemoryManager.scala | 12 +++++++++++-
.../memory/UnifiedMemoryManagerSuite.scala | 20 +++++++++++++++++++-
2 files changed, 30 insertions(+), 2 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/spark/blob/bcfd55fa/core/src/main/scala/org/apache/spark/memory/UnifiedMemoryManager.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/memory/UnifiedMemoryManager.scala b/core/src/main/scala/org/apache/spark/memory/UnifiedMemoryManager.scala
index a3321e3..6c57c98 100644
--- a/core/src/main/scala/org/apache/spark/memory/UnifiedMemoryManager.scala
+++ b/core/src/main/scala/org/apache/spark/memory/UnifiedMemoryManager.scala
@@ -183,7 +183,17 @@ object UnifiedMemoryManager {
val minSystemMemory = reservedMemory * 1.5
if (systemMemory < minSystemMemory) {
throw new IllegalArgumentException(s"System memory $systemMemory must " +
- s"be at least $minSystemMemory. Please use a larger heap size.")
+ s"be at least $minSystemMemory. Please increase heap size using the --driver-memory " +
+ s"option or spark.driver.memory in Spark configuration.")
+ }
+ // SPARK-12759 Check executor memory to fail fast if memory is insufficient
+ if (conf.contains("spark.executor.memory")) {
+ val executorMemory = conf.getSizeAsBytes("spark.executor.memory")
+ if (executorMemory < minSystemMemory) {
+ throw new IllegalArgumentException(s"Executor memory $executorMemory must be at least " +
+ s"$minSystemMemory. Please increase executor memory using the " +
+ s"--executor-memory option or spark.executor.memory in Spark configuration.")
+ }
}
val usableMemory = systemMemory - reservedMemory
val memoryFraction = conf.getDouble("spark.memory.fraction", 0.75)
http://git-wip-us.apache.org/repos/asf/spark/blob/bcfd55fa/core/src/test/scala/org/apache/spark/memory/UnifiedMemoryManagerSuite.scala
----------------------------------------------------------------------
diff --git a/core/src/test/scala/org/apache/spark/memory/UnifiedMemoryManagerSuite.scala b/core/src/test/scala/org/apache/spark/memory/UnifiedMemoryManagerSuite.scala
index 0c4359c..9686c66 100644
--- a/core/src/test/scala/org/apache/spark/memory/UnifiedMemoryManagerSuite.scala
+++ b/core/src/test/scala/org/apache/spark/memory/UnifiedMemoryManagerSuite.scala
@@ -227,7 +227,25 @@ class UnifiedMemoryManagerSuite extends MemoryManagerSuite with PrivateMethodTes
val exception = intercept[IllegalArgumentException] {
UnifiedMemoryManager(conf2, numCores = 1)
}
- assert(exception.getMessage.contains("larger heap size"))
+ assert(exception.getMessage.contains("increase heap size"))
+ }
+
+ test("insufficient executor memory") {
+ val systemMemory = 1024 * 1024
+ val reservedMemory = 300 * 1024
+ val memoryFraction = 0.8
+ val conf = new SparkConf()
+ .set("spark.memory.fraction", memoryFraction.toString)
+ .set("spark.testing.memory", systemMemory.toString)
+ .set("spark.testing.reservedMemory", reservedMemory.toString)
+ val mm = UnifiedMemoryManager(conf, numCores = 1)
+
+ // Try using an executor memory that's too small
+ val conf2 = conf.clone().set("spark.executor.memory", (reservedMemory / 2).toString)
+ val exception = intercept[IllegalArgumentException] {
+ UnifiedMemoryManager(conf2, numCores = 1)
+ }
+ assert(exception.getMessage.contains("increase executor memory"))
}
test("execution can evict cached blocks when there are multiple active tasks (SPARK-12155)") {
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org