Browse Source

Support for queues in Livy

Mariusz Strzelecki 10 năm trước cách đây
mục cha
commit
e3b8418304

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

@@ -49,6 +49,7 @@ object BatchSessionProcess {
     createBatchRequest.numExecutors.foreach(builder.numExecutors)
     createBatchRequest.archives.map(RelativePath).foreach(builder.archive)
     createBatchRequest.proxyUser.foreach(builder.proxyUser)
+    createBatchRequest.queue.foreach(builder.queue)
 
     builder.redirectOutput(Redirect.PIPE)
     builder.redirectErrorStream(true)

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

@@ -61,6 +61,7 @@ object BatchSessionYarn {
     createBatchRequest.executorCores.foreach(builder.executorCores)
     createBatchRequest.numExecutors.foreach(builder.numExecutors)
     createBatchRequest.archives.map(RelativePath).foreach(builder.archive)
+    createBatchRequest.queue.foreach(builder.queue)
 
     builder.redirectOutput(Redirect.PIPE)
     builder.redirectErrorStream(true)

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

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

+ 2 - 1
apps/spark/java/livy-server/src/main/scala/com/cloudera/hue/livy/server/interactive/CreateInteractiveRequest.scala

@@ -31,4 +31,5 @@ case class CreateInteractiveRequest(
     executorMemory: Option[String] = None,
     executorCores: Option[Int] = None,
     numExecutors: Option[Int] = None,
-    archives: List[String] = List())
+    archives: List[String] = List(),
+    queue: Option[String] = None)

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

@@ -56,6 +56,7 @@ object InteractiveSessionProcess extends Logging {
     createInteractiveRequest.jars.map(RelativePath).foreach(builder.jar)
     createInteractiveRequest.proxyUser.foreach(builder.proxyUser)
     createInteractiveRequest.pyFiles.map(RelativePath).foreach(builder.pyFile)
+    createInteractiveRequest.queue.foreach(builder.queue)
 
     sys.env.get("LIVY_REPL_JAVA_OPTS").foreach(builder.driverJavaOptions)
     livyConf.getOption(CONF_LIVY_REPL_DRIVER_CLASS_PATH).foreach(builder.driverClassPath)

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

@@ -57,6 +57,7 @@ object InteractiveSessionYarn {
     createInteractiveRequest.jars.map(RelativePath).foreach(builder.jar)
     createInteractiveRequest.proxyUser.foreach(builder.proxyUser)
     createInteractiveRequest.pyFiles.map(RelativePath).foreach(builder.pyFile)
+    createInteractiveRequest.queue.foreach(builder.queue)
 
     val kind = createInteractiveRequest.kind.toString