LIVY-246 Support multiple Spark home in runtime

cloudera · Apr 26, 2017 · 6926390 · 6926390
1 parent 7212e3f
commit 6926390
Show file tree

Hide file tree

Showing 5 changed files with 24 additions and 6 deletions.
diff --git a/conf/livy.conf.template b/conf/livy.conf.template
@@ -31,6 +31,12 @@
 # If livy should impersonate the requesting users when creating a new session.
 # livy.impersonation.enabled = true
 
+# Livy Spark Home
+# livy.server.spark-home_1.5.2=<path>/spark-1.5.2
+# livy.server.spark-home_1.6.3=<path>/spark-1.6.3
+# livy.server.spark-home_2.0.1=<path>/spark-2.0.1
+# livy.server.spark-home_2.1.0=<path>/spark-2.1.0
+
 # Comma-separated list of Livy RSC jars. By default Livy will upload jars from its installation
 # directory every time a session is started. By caching these files in HDFS, for example, startup
 # time of sessions on YARN can be reduced.

diff --git a/server/src/main/scala/com/cloudera/livy/LivyConf.scala b/server/src/main/scala/com/cloudera/livy/LivyConf.scala
@@ -246,14 +246,20 @@ class LivyConf(loadDefaults: Boolean) extends ClientConf[LivyConf](null) {
   def sparkDeployMode(): Option[String] = Option(get(LIVY_SPARK_DEPLOY_MODE)).filterNot(_.isEmpty)
 
   /** Return the location of the spark home directory */
-  def sparkHome(): Option[String] = Option(get(SPARK_HOME)).orElse(sys.env.get("SPARK_HOME"))
+  def sparkHome(version: Option[String] = None): Option[String] = {
+    version.map {version => Option(get(s"livy.server.spark-home_$version"))
+      .orElse(throw new IllegalArgumentException(
+        s"Spark version: $version is not supported"))
+    }.getOrElse(Option(get(s"livy.server.spark-home")).orElse(sys.env.get("SPARK_HOME")))
+  }
 
   /** Return the spark master Livy sessions should use. */
   def sparkMaster(): String = get(LIVY_SPARK_MASTER)
 
   /** Return the path to the spark-submit executable. */
-  def sparkSubmit(): String = {
-    sparkHome().map { _ + File.separator + "bin" + File.separator + "spark-submit" }.get
+  def sparkSubmit(version: Option[String] = None): String = {
+    sparkHome(version).map { _ + File.separator + "bin" + File.separator +
+      "spark-submit" }.get
   }
 
   /** Return the list of superusers. */

diff --git a/server/src/main/scala/com/cloudera/livy/server/batch/BatchSession.scala b/server/src/main/scala/com/cloudera/livy/server/batch/BatchSession.scala
@@ -18,6 +18,7 @@
 
 package com.cloudera.livy.server.batch
 
+import java.io.File
 import java.lang.ProcessBuilder.Redirect
 
 import scala.concurrent.{ExecutionContext, ExecutionContextExecutor}
@@ -62,7 +63,11 @@ object BatchSession {
           request.conf, request.jars, request.files, request.archives, request.pyFiles, livyConf))
       require(request.file != null, "File is required.")
 
-      val builder = new SparkProcessBuilder(livyConf)
+      val builder = new SparkProcessBuilder(livyConf, request.sparkVersion)
+      request.sparkVersion.map({ value =>
+        builder.env("SPARK_CONF_DIR", livyConf.sparkHome(request.sparkVersion)
+          + File.separator + "conf")
+      })
       builder.conf(conf)
 
       proxyUser.foreach(builder.proxyUser)

diff --git a/server/src/main/scala/com/cloudera/livy/server/batch/CreateBatchRequest.scala b/server/src/main/scala/com/cloudera/livy/server/batch/CreateBatchRequest.scala
@@ -36,5 +36,6 @@ class CreateBatchRequest {
   var queue: Option[String] = None
   var name: Option[String] = None
   var conf: Map[String, String] = Map()
+  var sparkVersion: Option[String] = None
 
 }
diff --git a/server/src/main/scala/com/cloudera/livy/utils/SparkProcessBuilder.scala b/server/src/main/scala/com/cloudera/livy/utils/SparkProcessBuilder.scala
@@ -25,9 +25,9 @@ import scala.collection.mutable.ArrayBuffer
 import com.cloudera.livy.{LivyConf, Logging}
 import com.cloudera.livy.util.LineBufferedProcess
 
-class SparkProcessBuilder(livyConf: LivyConf) extends Logging {
+class SparkProcessBuilder(livyConf: LivyConf, version: Option[String]) extends Logging  {
 
-  private[this] var _executable: String = livyConf.sparkSubmit()
+  private[this] var _executable: String = livyConf.sparkSubmit(version)
   private[this] var _master: Option[String] = None
   private[this] var _deployMode: Option[String] = None
   private[this] var _className: Option[String] = None