Browse Source

[livy] Convert some builder options to generate the spark.* options

Erick Tryzelaar 10 years ago
parent
commit
08786ef

+ 15 - 33
apps/spark/java/livy-spark/src/main/scala/com/cloudera/hue/livy/spark/SparkProcessBuilder.scala

@@ -50,16 +50,11 @@ class SparkProcessBuilder(livyConf: LivyConf) extends Logging {
   private[this] var _pyFiles: ArrayBuffer[Path] = ArrayBuffer()
   private[this] var _pyFiles: ArrayBuffer[Path] = ArrayBuffer()
   private[this] var _files: ArrayBuffer[Path] = ArrayBuffer()
   private[this] var _files: ArrayBuffer[Path] = ArrayBuffer()
   private[this] var _conf: ArrayBuffer[(String, String)] = ArrayBuffer()
   private[this] var _conf: ArrayBuffer[(String, String)] = ArrayBuffer()
-  private[this] var _driverMemory: Option[String] = None
   private[this] var _driverJavaOptions: Option[String] = None
   private[this] var _driverJavaOptions: Option[String] = None
   private[this] var _driverClassPath: ArrayBuffer[String] = ArrayBuffer()
   private[this] var _driverClassPath: ArrayBuffer[String] = ArrayBuffer()
-  private[this] var _executorMemory: Option[String] = None
   private[this] var _proxyUser: Option[String] = None
   private[this] var _proxyUser: Option[String] = None
 
 
-  private[this] var _driverCores: Option[String] = None
-  private[this] var _executorCores: Option[String] = None
   private[this] var _queue: Option[String] = None
   private[this] var _queue: Option[String] = None
-  private[this] var _numExecutors: Option[String] = None
   private[this] var _archives: ArrayBuffer[Path] = ArrayBuffer()
   private[this] var _archives: ArrayBuffer[Path] = ArrayBuffer()
 
 
   private[this] var _env: ArrayBuffer[(String, String)] = ArrayBuffer()
   private[this] var _env: ArrayBuffer[(String, String)] = ArrayBuffer()
@@ -132,14 +127,8 @@ class SparkProcessBuilder(livyConf: LivyConf) extends Logging {
     this
     this
   }
   }
 
 
-  def driverMemory(driverMemory: String): SparkProcessBuilder = {
-    _driverMemory = Some(driverMemory)
-    this
-  }
-
   def driverJavaOptions(driverJavaOptions: String): SparkProcessBuilder = {
   def driverJavaOptions(driverJavaOptions: String): SparkProcessBuilder = {
-    _driverJavaOptions = Some(driverJavaOptions)
-    this
+    conf("spark.driver.extraJavaOptions", driverJavaOptions)
   }
   }
 
 
   def driverClassPath(classPath: String): SparkProcessBuilder = {
   def driverClassPath(classPath: String): SparkProcessBuilder = {
@@ -152,23 +141,16 @@ class SparkProcessBuilder(livyConf: LivyConf) extends Logging {
     this
     this
   }
   }
 
 
-  def executorMemory(executorMemory: String): SparkProcessBuilder = {
-    _executorMemory = Some(executorMemory)
-    this
-  }
-
-  def proxyUser(proxyUser: String): SparkProcessBuilder = {
-    _proxyUser = Some(proxyUser)
-    this
-  }
-
   def driverCores(driverCores: Int): SparkProcessBuilder = {
   def driverCores(driverCores: Int): SparkProcessBuilder = {
     this.driverCores(driverCores.toString)
     this.driverCores(driverCores.toString)
   }
   }
 
 
+  def driverMemory(driverMemory: String): SparkProcessBuilder = {
+    conf("spark.driver.memory", driverMemory)
+  }
+
   def driverCores(driverCores: String): SparkProcessBuilder = {
   def driverCores(driverCores: String): SparkProcessBuilder = {
-    _driverCores = Some(driverCores)
-    this
+    conf("spark.driver.cores", driverCores)
   }
   }
 
 
   def executorCores(executorCores: Int): SparkProcessBuilder = {
   def executorCores(executorCores: Int): SparkProcessBuilder = {
@@ -176,17 +158,23 @@ class SparkProcessBuilder(livyConf: LivyConf) extends Logging {
   }
   }
 
 
   def executorCores(executorCores: String): SparkProcessBuilder = {
   def executorCores(executorCores: String): SparkProcessBuilder = {
-    _executorCores = Some(executorCores)
-    this
+    conf("spark.executor.cores", executorCores)
   }
   }
 
 
+  def executorMemory(executorMemory: String): SparkProcessBuilder = {
+    conf("spark.executor.memory", executorMemory)
+  }
 
 
   def numExecutors(numExecutors: Int): SparkProcessBuilder = {
   def numExecutors(numExecutors: Int): SparkProcessBuilder = {
     this.numExecutors(numExecutors.toString)
     this.numExecutors(numExecutors.toString)
   }
   }
 
 
   def numExecutors(numExecutors: String): SparkProcessBuilder = {
   def numExecutors(numExecutors: String): SparkProcessBuilder = {
-    _numExecutors = Some(numExecutors)
+    this.conf("spark.executor.instances", numExecutors)
+  }
+
+  def proxyUser(proxyUser: String): SparkProcessBuilder = {
+    _proxyUser = Some(proxyUser)
     this
     this
   }
   }
 
 
@@ -250,19 +238,13 @@ class SparkProcessBuilder(livyConf: LivyConf) extends Logging {
     addList("--files", _files.map(fromPath))
     addList("--files", _files.map(fromPath))
     addOpt("--class", _className)
     addOpt("--class", _className)
     addList("--conf", _conf.map { case (key, value) => f"$key=$value" })
     addList("--conf", _conf.map { case (key, value) => f"$key=$value" })
-    addOpt("--driver-memory", _driverMemory)
     addOpt("--driver-java-options", _driverJavaOptions)
     addOpt("--driver-java-options", _driverJavaOptions)
     addList("--driver-class-path", _driverClassPath)
     addList("--driver-class-path", _driverClassPath)
-    addOpt("--driver-cores", _driverCores)
-    addOpt("--executor-memory", _executorMemory)
 
 
     if (livyConf.getBoolean(LivyConf.IMPERSONATION_ENABLED_KEY, true)) {
     if (livyConf.getBoolean(LivyConf.IMPERSONATION_ENABLED_KEY, true)) {
       addOpt("--proxy-user", _proxyUser)
       addOpt("--proxy-user", _proxyUser)
     }
     }
 
 
-    addOpt("--driver-cores", _driverCores)
-    addOpt("--executor-cores", _executorCores)
-    addOpt("--num-executors", _numExecutors)
     addOpt("--queue", _queue)
     addOpt("--queue", _queue)
     addList("--archives", _archives.map(fromPath))
     addList("--archives", _archives.map(fromPath))