Browse Source

[livy] Create a SparkProcess wrapper

Erick Tryzelaar 10 years ago
parent
commit
b671ef1

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

@@ -0,0 +1,12 @@
+package com.cloudera.hue.livy.spark
+
+import com.cloudera.hue.livy.LineBufferedProcess
+
+object SparkProcess {
+  def apply(process: Process): SparkProcess = {
+    new SparkProcess(process)
+  }
+}
+
+class SparkProcess(process: Process) extends LineBufferedProcess(process) {
+}

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

@@ -216,7 +216,7 @@ class SparkSubmitProcessBuilder(livyConf: LivyConf) extends Logging {
     this
   }
 
-  def start(file: Path, args: Traversable[String]): Process = {
+  def start(file: Path, args: Traversable[String]): SparkProcess = {
     var args_ = ArrayBuffer(fromPath(_executable))
 
     def addOpt(option: String, value: Option[String]): Unit = {
@@ -268,7 +268,7 @@ class SparkSubmitProcessBuilder(livyConf: LivyConf) extends Logging {
     _redirectError.foreach(pb.redirectError)
     _redirectErrorStream.foreach(pb.redirectErrorStream)
 
-    pb.start()
+    SparkProcess(pb.start())
   }
 
   private def fromPath(path: Path) = path match {

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

@@ -32,7 +32,7 @@ object BatchSessionProcess {
     val builder = sparkBuilder(livyConf, createBatchRequest)
 
     val process = builder.start(RelativePath(createBatchRequest.file), createBatchRequest.args)
-    new BatchSessionProcess(id, new LineBufferedProcess(process))
+    new BatchSessionProcess(id, process)
   }
 
   private def sparkBuilder(livyConf: LivyConf, createBatchRequest: CreateBatchRequest): SparkSubmitProcessBuilder = {

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

@@ -38,7 +38,7 @@ object BatchSessionYarn {
   def apply(livyConf: LivyConf, client: Client, id: Int, createBatchRequest: CreateBatchRequest): BatchSession = {
     val builder = sparkBuilder(livyConf, createBatchRequest)
 
-    val process = new LineBufferedProcess(builder.start(RelativePath(createBatchRequest.file), createBatchRequest.args))
+    val process = builder.start(RelativePath(createBatchRequest.file), createBatchRequest.args)
     val job = Future {
       client.getJobFromProcess(process)
     }

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

@@ -21,7 +21,7 @@ package com.cloudera.hue.livy.server.interactive
 import java.lang.ProcessBuilder.Redirect
 import java.net.URL
 
-import com.cloudera.hue.livy.spark.SparkSubmitProcessBuilder
+import com.cloudera.hue.livy.spark.{SparkProcess, SparkSubmitProcessBuilder}
 import com.cloudera.hue.livy.spark.SparkSubmitProcessBuilder.{RelativePath, AbsolutePath}
 import com.cloudera.hue.livy.{LivyConf, Logging, Utils}
 
@@ -41,7 +41,7 @@ object InteractiveSessionProcess extends Logging {
   }
 
   // Loop until we've started a process with a valid port.
-  private def startProcess(livyConf: LivyConf, id: Int, createInteractiveRequest: CreateInteractiveRequest): Process = {
+  private def startProcess(livyConf: LivyConf, id: Int, createInteractiveRequest: CreateInteractiveRequest): SparkProcess = {
 
     val builder = new SparkSubmitProcessBuilder(livyConf)
 
@@ -80,13 +80,13 @@ object InteractiveSessionProcess extends Logging {
 
 private class InteractiveSessionProcess(id: Int,
                                         createInteractiveRequest: CreateInteractiveRequest,
-                                        process: Process) extends InteractiveWebSession(id, createInteractiveRequest) {
+                                        process: SparkProcess) extends InteractiveWebSession(id, createInteractiveRequest) {
 
   val stdoutThread = new Thread {
     override def run() = {
       val regex = """Starting livy-repl on (https?://.*)""".r
 
-      val lines = Source.fromInputStream(process.getInputStream).getLines()
+      val lines = process.inputIterator
 
       // Loop until we find the ip address to talk to livy-repl.
       @tailrec

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

@@ -36,7 +36,10 @@ object InteractiveSessionYarn {
   private val CONF_LIVY_JAR = "livy.yarn.jar"
   private lazy val regex = """Application report for (\w+)""".r.unanchored
 
-  def create(livyConf: LivyConf, client: Client, id: Int, createInteractiveRequest: CreateInteractiveRequest): InteractiveSession = {
+  def create(livyConf: LivyConf,
+             client: Client,
+             id: Int,
+             createInteractiveRequest: CreateInteractiveRequest): InteractiveSession = {
     val callbackUrl = System.getProperty("livy.server.callback-url")
     val url = f"$callbackUrl/sessions/$id/callback"
 
@@ -61,11 +64,10 @@ object InteractiveSessionYarn {
     val process = builder.start(AbsolutePath(livyJar(livyConf)), List(createInteractiveRequest.kind.toString))
 
     val job = Future {
-      val proc = new LineBufferedProcess(process)
-      val job = client.getJobFromProcess(proc)
+      val job = client.getJobFromProcess(process)
 
       // We don't need the process anymore.
-      proc.destroy()
+      process.destroy()
 
       job
     }