Browse Source

[livy] Make a SparkSubmitProcessBuilder

Erick Tryzelaar 10 years ago
parent
commit
648f0376cb

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

@@ -0,0 +1,230 @@
+package com.cloudera.hue.livy.spark
+
+import com.cloudera.hue.livy.Logging
+
+import scala.collection.JavaConversions._
+import scala.collection.mutable.ArrayBuffer
+
+class SparkProcessBuilder extends Logging {
+
+  private[this] var _executable = "spark-submit"
+  private[this] var _master: Option[String] = None
+  private[this] var _deployMode: Option[String] = None
+  private[this] var _className: Option[String] = None
+  private[this] var _name: Option[String] = None
+  private[this] var _jars: ArrayBuffer[String] = ArrayBuffer()
+  private[this] var _pyFiles: ArrayBuffer[String] = ArrayBuffer()
+  private[this] var _files: ArrayBuffer[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 _driverClassPath: ArrayBuffer[String] = ArrayBuffer()
+  private[this] var _executorMemory: 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 _numExecutors: Option[String] = None
+  private[this] var _archives: ArrayBuffer[String] = ArrayBuffer()
+
+  private[this] var _env: ArrayBuffer[(String, String)] = ArrayBuffer()
+  private[this] var _redirectOutput: Option[ProcessBuilder.Redirect] = None
+  private[this] var _redirectError: Option[ProcessBuilder.Redirect] = None
+  private[this] var _redirectErrorStream: Option[Boolean] = None
+
+  def executable(executable: String): SparkProcessBuilder = {
+    _executable = executable
+    this
+  }
+
+  def master(masterUrl: String): SparkProcessBuilder = {
+    _master = Some(masterUrl)
+    this
+  }
+
+  def deployMode(deployMode: String): SparkProcessBuilder = {
+    _deployMode = Some(deployMode)
+    this
+  }
+
+  def className(className: String): SparkProcessBuilder = {
+    _className = Some(className)
+    this
+  }
+
+  def name(name: String): SparkProcessBuilder = {
+    _name = Some(name)
+    this
+  }
+
+  def jar(jar: String): SparkProcessBuilder = {
+    this._jars += jar
+    this
+  }
+
+  def jars(jars: Traversable[String]): SparkProcessBuilder = {
+    this._jars ++= jars
+    this
+  }
+
+  def pyFile(pyFile: String): SparkProcessBuilder = {
+    this._pyFiles += pyFile
+    this
+  }
+
+  def pyFiles(pyFiles: Traversable[String]): SparkProcessBuilder = {
+    this._pyFiles ++= pyFiles
+    this
+  }
+
+  def file(file: String): SparkProcessBuilder = {
+    this._files += file
+    this
+  }
+
+  def files(files: Traversable[String]): SparkProcessBuilder = {
+    this._files ++= files
+    this
+  }
+
+  def conf(key: String, value: String): SparkProcessBuilder = {
+    this._conf += ((key, value))
+    this
+  }
+
+  def conf(conf: Traversable[(String, String)]): SparkProcessBuilder = {
+    this._conf ++= conf
+    this
+  }
+
+  def driverMemory(driverMemory: String): SparkProcessBuilder = {
+    _driverMemory = Some(driverMemory)
+    this
+  }
+
+  def driverJavaOptions(driverJavaOptions: String): SparkProcessBuilder = {
+    _driverJavaOptions = Some(driverJavaOptions)
+    this
+  }
+
+  def driverClassPath(classPath: String): SparkProcessBuilder = {
+    _driverClassPath += classPath
+    this
+  }
+
+  def driverClassPaths(classPaths: Traversable[String]): SparkProcessBuilder = {
+    _driverClassPath ++= classPaths
+    this
+  }
+
+  def executorMemory(executorMemory: String): SparkProcessBuilder = {
+    _executorMemory = Some(executorMemory)
+    this
+  }
+
+  def proxyUser(proxyUser: String): SparkProcessBuilder = {
+    _proxyUser = Some(proxyUser)
+    this
+  }
+
+  def driverCores(driverCores: String): SparkProcessBuilder = {
+    _driverCores = Some(driverCores)
+    this
+  }
+
+  def executorCores(executorCores: String): SparkProcessBuilder = {
+    _executorCores = Some(executorCores)
+    this
+  }
+
+  def queue(queue: String): SparkProcessBuilder = {
+    _queue = Some(queue)
+    this
+  }
+
+  def archive(archive: String): SparkProcessBuilder = {
+    _archives += archive
+    this
+  }
+
+  def archives(archives: Traversable[String]): SparkProcessBuilder = {
+    _archives ++= archives
+    this
+  }
+
+  def env(key: String, value: String): SparkProcessBuilder = {
+    _env += ((key, value))
+    this
+  }
+
+  def redirectOutput(redirect: ProcessBuilder.Redirect): SparkProcessBuilder = {
+    _redirectOutput = Some(redirect)
+    this
+  }
+
+  def redirectError(redirect: ProcessBuilder.Redirect): SparkProcessBuilder = {
+    _redirectError = Some(redirect)
+    this
+  }
+
+  def redirectErrorStream(redirect: Boolean): SparkProcessBuilder = {
+    _redirectErrorStream = Some(redirect)
+    this
+  }
+
+  def start(file: String, args: Traversable[String]): Process = {
+    var args_ = ArrayBuffer(_executable)
+
+    def addOpt(option: String, value: Option[String]): Unit = {
+      value.foreach { v =>
+        args_ += option
+        args_ += v
+      }
+    }
+
+    def addList(option: String, values: Traversable[String]): Unit = {
+      if (values.nonEmpty) {
+        args_ += option
+        args_ += values.mkString(",")
+      }
+    }
+
+    addOpt("--master", _master)
+    addOpt("--deploy-mode", _deployMode)
+    addOpt("--name", _name)
+    addList("--jars", _jars)
+    addList("--py-files", _pyFiles)
+    addList("--files", _files)
+    addOpt("--class", _className)
+    addList("--conf", _conf.map { case (key, value) => f"$key=$value" })
+    addOpt("--driver-memory", _driverMemory)
+    addOpt("--driver-java-options", _driverJavaOptions)
+    addList("--driver-class-path", _driverClassPath)
+    addOpt("--driver-cores", _driverCores)
+    addOpt("--executor-memory", _executorMemory)
+    addOpt("--proxy-user", _proxyUser)
+    addOpt("--driver-cores", _driverCores)
+    addOpt("--executor-cores", _executorCores)
+    addOpt("--queue", _queue)
+    addList("--archives", _archives)
+
+    args_ += file
+    args_ ++= args
+
+    info("Running %s", args.mkString(" "))
+
+    val pb = new ProcessBuilder(args_)
+    val env = pb.environment()
+
+    for ((key, value) <- _env) {
+      env.put(key, value)
+    }
+
+    _redirectOutput.foreach(pb.redirectOutput)
+    _redirectError.foreach(pb.redirectError)
+    _redirectErrorStream.foreach(pb.redirectErrorStream)
+
+    pb.start()
+  }
+}

+ 11 - 28
apps/spark/java/livy-server/src/main/scala/com/cloudera/hue/livy/server/sessions/ProcessSession.scala

@@ -3,6 +3,7 @@ package com.cloudera.hue.livy.server.sessions
 import java.lang.ProcessBuilder.Redirect
 import java.lang.ProcessBuilder.Redirect
 import java.net.URL
 import java.net.URL
 
 
+import com.cloudera.hue.livy.spark.SparkProcessBuilder
 import com.cloudera.hue.livy.{LivyConf, Logging, Utils}
 import com.cloudera.hue.livy.{LivyConf, Logging, Utils}
 
 
 import scala.annotation.tailrec
 import scala.annotation.tailrec
@@ -24,43 +25,25 @@ object ProcessSession extends Logging {
 
 
   // Loop until we've started a process with a valid port.
   // Loop until we've started a process with a valid port.
   private def startProcess(livyConf: LivyConf, id: String, kind: Session.Kind, proxyUser: Option[String]): Process = {
   private def startProcess(livyConf: LivyConf, id: String, kind: Session.Kind, proxyUser: Option[String]): Process = {
-    val args = ArrayBuffer(
-      "spark-submit",
-      "--class", "com.cloudera.hue.livy.repl.Main"
-    )
-
-    sys.env.get("LIVY_REPL_JAVA_OPTS").foreach { case javaOpts =>
-      args += "--driver-java-options"
-      args += javaOpts
-    }
-
-    livyConf.getOption(CONF_LIVY_REPL_DRIVER_CLASS_PATH).foreach { case extraClassPath =>
-      args += "--driver-class-path"
-      args += extraClassPath
-    }
-
-    proxyUser.foreach { case user =>
-      args += "--proxy-user"
-      args += user
-    }
 
 
-    args += livyJar(livyConf)
-    args += kind.toString
+    val builder = new SparkProcessBuilder()
 
 
-    debug("Running %s", args.mkString(" "))
+    builder.className("com.cloudera.hue.livy.repl.Main")
 
 
-    val pb = new ProcessBuilder(args)
+    sys.env.get("LIVY_REPL_JAVA_OPTS").foreach(builder.driverJavaOptions)
+    livyConf.getOption(CONF_LIVY_REPL_DRIVER_CLASS_PATH).foreach(builder.driverClassPath)
+    proxyUser.foreach(builder.proxyUser)
 
 
     livyConf.getOption(CONF_LIVY_REPL_CALLBACK_URL).foreach { case callbackUrl =>
     livyConf.getOption(CONF_LIVY_REPL_CALLBACK_URL).foreach { case callbackUrl =>
-      pb.environment().put("LIVY_CALLBACK_URL", f"$callbackUrl/sessions/$id/callback")
+      builder.env("LIVY_CALLBACK_URL", f"$callbackUrl/sessions/$id/callback")
     }
     }
 
 
-    pb.environment().put("LIVY_PORT", "0")
+    builder.env("LIVY_PORT", "0")
 
 
-    pb.redirectOutput(Redirect.PIPE)
-    pb.redirectError(Redirect.INHERIT)
+    builder.redirectOutput(Redirect.PIPE)
+    builder.redirectError(Redirect.INHERIT)
 
 
-    pb.start()
+    builder.start(livyJar(livyConf), List(kind.toString))
   }
   }
 
 
   private def livyJar(conf: LivyConf): String = {
   private def livyJar(conf: LivyConf): String = {

+ 9 - 21
apps/spark/java/livy-yarn/src/main/scala/com/cloudera/hue/livy/yarn/Client.scala

@@ -3,6 +3,7 @@ package com.cloudera.hue.livy.yarn
 import java.io.{BufferedReader, InputStreamReader}
 import java.io.{BufferedReader, InputStreamReader}
 import java.lang.ProcessBuilder.Redirect
 import java.lang.ProcessBuilder.Redirect
 
 
+import com.cloudera.hue.livy.spark.SparkProcessBuilder
 import com.cloudera.hue.livy.{LivyConf, Logging, Utils}
 import com.cloudera.hue.livy.{LivyConf, Logging, Utils}
 import org.apache.hadoop.yarn.api.records.{ApplicationId, FinalApplicationStatus, YarnApplicationState}
 import org.apache.hadoop.yarn.api.records.{ApplicationId, FinalApplicationStatus, YarnApplicationState}
 import org.apache.hadoop.yarn.client.api.YarnClient
 import org.apache.hadoop.yarn.client.api.YarnClient
@@ -45,32 +46,19 @@ class Client(livyConf: LivyConf) extends Logging {
                         callbackUrl: String): Future[Job] = {
                         callbackUrl: String): Future[Job] = {
     val url = f"$callbackUrl/sessions/$id/callback"
     val url = f"$callbackUrl/sessions/$id/callback"
 
 
-    val args: ArrayBuffer[String] = ArrayBuffer(
-      "spark-submit",
-      "--master", "yarn-cluster",
-      "--class", "com.cloudera.hue.livy.repl.Main",
-      "--driver-java-options", f"-Dlivy.repl.callback-url=$url -Dlivy.repl.port=0"
-    )
+    val builder = new SparkProcessBuilder()
 
 
-    proxyUser.foreach { case user =>
-      args += "--proxy-user"
-      args += user
-    }
-
-    args += livyJar(livyConf)
-    args += kind
-
-    debug("Running %s", args.mkString(" "))
-
-    val builder: ProcessBuilder = new ProcessBuilder(args)
+    builder.master("yarn-cluster")
+    builder.className("com.cloudera.hue.livy.repl.Main")
+    builder.driverJavaOptions(f"-Dlivy.repl.callback-url=$url -Dlivy.repl.port=0")
+    proxyUser.foreach(builder.proxyUser)
 
 
     builder.redirectOutput(Redirect.PIPE)
     builder.redirectOutput(Redirect.PIPE)
-    builder.redirectErrorStream(true)
+    builder.redirectErrorStream(redirect = true)
 
 
-    Future {
-
-      val process = builder.start()
+    val process = builder.start(livyJar(livyConf), List(kind.toString))
 
 
+    Future {
       val stdout = new BufferedReader(new InputStreamReader(process.getInputStream), 1)
       val stdout = new BufferedReader(new InputStreamReader(process.getInputStream), 1)
 
 
       val applicationId = parseApplicationId(stdout).getOrElse(throw new FailedToSubmitApplication)
       val applicationId = parseApplicationId(stdout).getOrElse(throw new FailedToSubmitApplication)