Explorar o código

Merge pull request #196 from xiaop1987/add_num_executor

Add numExecutors and fix files option issue
Erick Tryzelaar %!s(int64=10) %!d(string=hai) anos
pai
achega
d12c3b62d2

+ 6 - 0
apps/spark/java/README.rst

@@ -255,6 +255,7 @@ Request Body
 | pyFiles        | files to be placed on the PYTHONPATH             | list of paths   |
 +----------------+--------------------------------------------------+-----------------+
 | files          | files to be placed in executor working directory | list of paths   |
+|                | (Add local path as "file:///etc/...")            |                 |
 +----------------+--------------------------------------------------+-----------------+
 | driverMemory   | memory for driver                                | string          |
 +----------------+--------------------------------------------------+-----------------+
@@ -264,6 +265,8 @@ Request Body
 +----------------+--------------------------------------------------+-----------------+
 | executorCores  | number of cores used by executor                 | int             |
 +----------------+--------------------------------------------------+-----------------+
+| numExecutors   | number of executor                               | int             |
++----------------+--------------------------------------------------+-----------------+
 | archives       |                                                  | list of paths   |
 +----------------+--------------------------------------------------+-----------------+
 
@@ -412,6 +415,7 @@ Batch
 | pyFiles        | files to be placed on the PYTHONPATH             | list of paths   |
 +----------------+--------------------------------------------------+-----------------+
 | files          | files to be placed in executor working directory | list of paths   |
+|                | (Add local path as  "file:///etc/...")           |                 |
 +----------------+--------------------------------------------------+-----------------+
 | driverMemory   | memory for driver                                | string          |
 +----------------+--------------------------------------------------+-----------------+
@@ -421,6 +425,8 @@ Batch
 +----------------+--------------------------------------------------+-----------------+
 | executorCores  | number of cores used by executor                 | int             |
 +----------------+--------------------------------------------------+-----------------+
+| numExecutors   | number of executor                               | int             |
++----------------+--------------------------------------------------+-----------------+
 | archives       |                                                  | list of paths   |
 +----------------+--------------------------------------------------+-----------------+
 

+ 11 - 0
apps/spark/java/livy-core/src/main/scala/com/cloudera/hue/livy/spark/SparkSubmitProcessBuilder.scala

@@ -181,6 +181,16 @@ class SparkSubmitProcessBuilder(livyConf: LivyConf) extends Logging {
     this
   }
 
+
+  def numExecutors(numExecutors: Int): SparkSubmitProcessBuilder = {
+    this.numExecutors(numExecutors.toString)
+  }
+
+  def numExecutors(numExecutors: String): SparkSubmitProcessBuilder = {
+    _numExecutors = Some(numExecutors)
+    this
+  }
+
   def queue(queue: String): SparkSubmitProcessBuilder = {
     _queue = Some(queue)
     this
@@ -249,6 +259,7 @@ class SparkSubmitProcessBuilder(livyConf: LivyConf) extends Logging {
     addOpt("--proxy-user", _proxyUser)
     addOpt("--driver-cores", _driverCores)
     addOpt("--executor-cores", _executorCores)
+    addOpt("--num-executors", _numExecutors)
     addOpt("--queue", _queue)
     addList("--archives", _archives.map(fromPath))
 

+ 1 - 0
apps/spark/java/livy-server/src/main/scala/com/cloudera/hue/livy/server/batch/BatchSessionManager.scala

@@ -65,4 +65,5 @@ case class CreateBatchRequest(file: String,
                               driverCores: Option[Int] = None,
                               executorMemory: Option[String] = None,
                               executorCores: Option[Int] = None,
+                              numExecutors: Option[Int] = None,
                               archives: List[String] = List())

+ 1 - 0
apps/spark/java/livy-server/src/main/scala/com/cloudera/hue/livy/server/batch/BatchSessionProcess.scala

@@ -46,6 +46,7 @@ object BatchSessionProcess {
     createBatchRequest.driverCores.foreach(builder.driverCores)
     createBatchRequest.executorMemory.foreach(builder.executorMemory)
     createBatchRequest.executorCores.foreach(builder.executorCores)
+    createBatchRequest.numExecutors.foreach(builder.numExecutors)
     createBatchRequest.archives.map(RelativePath).foreach(builder.archive)
     createBatchRequest.proxyUser.foreach(builder.proxyUser)
 

+ 1 - 0
apps/spark/java/livy-server/src/main/scala/com/cloudera/hue/livy/server/batch/BatchSessionYarn.scala

@@ -59,6 +59,7 @@ object BatchSessionYarn {
     createBatchRequest.driverCores.foreach(builder.driverCores)
     createBatchRequest.executorMemory.foreach(builder.executorMemory)
     createBatchRequest.executorCores.foreach(builder.executorCores)
+    createBatchRequest.numExecutors.foreach(builder.numExecutors)
     createBatchRequest.archives.map(RelativePath).foreach(builder.archive)
 
     builder.redirectOutput(Redirect.PIPE)