Skip to content

Commit e5796de

Browse files
jerryshaoalex-the-man
jerryshao
authored and
alex-the-man
committed
LIVY-227. Include both repl-2.10 and repl-2.11 bundles in Livy assembly.
Closes #205
1 parent 33b20f5 commit e5796de

File tree

10 files changed

+150
-70
lines changed

10 files changed

+150
-70
lines changed

‎assembly/assembly.xml

+10-3
Original file line numberDiff line numberDiff line change
@@ -29,8 +29,15 @@
2929
</includes>
3030
</fileSet>
3131
<fileSet>
32-
<directory>${project.parent.basedir}/${repl.dir}/target/jars</directory>
33-
<outputDirectory>${assembly.name}/repl-jars</outputDirectory>
32+
<directory>${project.parent.basedir}/repl/scala-2.10/target/jars</directory>
33+
<outputDirectory>${assembly.name}/repl_2.10-jars</outputDirectory>
34+
<includes>
35+
<include>*</include>
36+
</includes>
37+
</fileSet>
38+
<fileSet>
39+
<directory>${project.parent.basedir}/repl/scala-2.11/target/jars</directory>
40+
<outputDirectory>${assembly.name}/repl_2.11-jars</outputDirectory>
3441
<includes>
3542
<include>*</include>
3643
</includes>
@@ -43,4 +50,4 @@
4350
</includes>
4451
</fileSet>
4552
</fileSets>
46-
</assembly>
53+
</assembly>

‎assembly/pom.xml

+12-36
Original file line numberDiff line numberDiff line change
@@ -33,7 +33,6 @@
3333
<assembly.name>livy-server-${project.version}</assembly.name>
3434
<assembly.format>zip</assembly.format>
3535
<skipDeploy>true</skipDeploy>
36-
<repl.dir>repl/scala-2.10</repl.dir>
3736
</properties>
3837

3938
<dependencies>
@@ -43,6 +42,18 @@
4342
<version>${project.version}</version>
4443
</dependency>
4544

45+
<dependency>
46+
<groupId>${project.groupId}</groupId>
47+
<artifactId>livy-repl_2.10</artifactId>
48+
<version>${project.version}</version>
49+
</dependency>
50+
51+
<dependency>
52+
<groupId>${project.groupId}</groupId>
53+
<artifactId>livy-repl_2.11</artifactId>
54+
<version>${project.version}</version>
55+
</dependency>
56+
4657
<dependency>
4758
<groupId>${project.groupId}</groupId>
4859
<artifactId>livy-server</artifactId>
@@ -84,41 +95,6 @@
8495
<assembly.format>tar.gz</assembly.format>
8596
</properties>
8697
</profile>
87-
88-
<profile>
89-
<id>scala-2.10</id>
90-
<activation>
91-
<property>
92-
<name>!scala-2.11</name>
93-
</property>
94-
</activation>
95-
<dependencies>
96-
<dependency>
97-
<groupId>${project.groupId}</groupId>
98-
<artifactId>livy-repl_2.10</artifactId>
99-
<version>${project.version}</version>
100-
</dependency>
101-
</dependencies>
102-
</profile>
103-
104-
<profile>
105-
<id>scala-2.11</id>
106-
<activation>
107-
<property>
108-
<name>scala-2.11</name>
109-
</property>
110-
</activation>
111-
<properties>
112-
<repl.dir>repl/scala-2.11</repl.dir>
113-
</properties>
114-
<dependencies>
115-
<dependency>
116-
<groupId>${project.groupId}</groupId>
117-
<artifactId>livy-repl_2.11</artifactId>
118-
<version>${project.version}</version>
119-
</dependency>
120-
</dependencies>
121-
</profile>
12298
</profiles>
12399

124100
</project>

‎integration-test/pom.xml

-7
Original file line numberDiff line numberDiff line change
@@ -43,13 +43,6 @@
4343

4444
<dependencies>
4545

46-
<dependency>
47-
<groupId>${project.groupId}</groupId>
48-
<artifactId>livy-assembly</artifactId>
49-
<version>${project.version}</version>
50-
<type>pom</type>
51-
</dependency>
52-
5346
<dependency>
5447
<groupId>${project.groupId}</groupId>
5548
<artifactId>livy-core_${scala.binary.version}</artifactId>

‎server/src/main/scala/com/cloudera/livy/LivyConf.scala

+7
Original file line numberDiff line numberDiff line change
@@ -42,6 +42,13 @@ object LivyConf {
4242
val LIVY_SPARK_MASTER = Entry("livy.spark.master", "local")
4343
val LIVY_SPARK_DEPLOY_MODE = Entry("livy.spark.deployMode", null)
4444

45+
// Two configurations to specify Spark and related Scala version. These are internal
46+
// configurations will be set by LivyServer and used in session creation. It is not required to
47+
// set usually unless running with unofficial Spark + Scala versions
48+
// (like Spark 2.0 + Scala 2.10, Spark 1.6 + Scala 2.11)
49+
val LIVY_SPARK_SCALA_VERSION = Entry("livy.spark.scalaVersion", null)
50+
val LIVY_SPARK_VERSION = Entry("livy.spark.version", null)
51+
4552
val SESSION_STAGING_DIR = Entry("livy.session.staging-dir", null)
4653
val FILE_UPLOAD_MAX_SIZE = Entry("livy.file.upload.max.size", 100L * 1024 * 1024)
4754
val LOCAL_FS_WHITELIST = Entry("livy.file.local-dir-whitelist", null)

‎server/src/main/scala/com/cloudera/livy/server/LivyServer.scala

+29-1
Original file line numberDiff line numberDiff line change
@@ -64,7 +64,35 @@ class LivyServer extends Logging {
6464

6565
// Make sure the `spark-submit` program exists, otherwise much of livy won't work.
6666
testSparkHome(livyConf)
67-
testSparkSubmit(livyConf)
67+
68+
// Test spark-submit and get Spark Scala version accordingly.
69+
val (sparkVersion, scalaVersion) = sparkSubmitVersion(livyConf)
70+
testSparkVersion(sparkVersion)
71+
72+
// If Spark and Scala version is set manually, should verify if they're consistent with
73+
// ones parsed from "spark-submit --version"
74+
val formattedSparkVersion = formatSparkVersion(sparkVersion)
75+
Option(livyConf.get(LIVY_SPARK_VERSION)).map(formatSparkVersion).foreach { version =>
76+
require(formattedSparkVersion == version,
77+
s"Configured Spark version $version is not equal to Spark version $formattedSparkVersion " +
78+
"got from spark-submit -version")
79+
}
80+
81+
if (!scalaVersion.isEmpty && Option(livyConf.get(LIVY_SPARK_SCALA_VERSION)).isDefined) {
82+
Option(livyConf.get(LIVY_SPARK_SCALA_VERSION))
83+
.map(s => formatScalaVersion(s, formattedSparkVersion))
84+
.foreach { version =>
85+
require(version == scalaVersion,
86+
s"Configured Scala version $version is not equal to Scala version $scalaVersion got " +
87+
"from spark-submit -version")
88+
}
89+
}
90+
91+
// Set formatted Spark and Scala version into livy configuration, this will be used by
92+
// session creation.
93+
livyConf.set(LIVY_SPARK_VERSION.key, formattedSparkVersion.productIterator.mkString("."))
94+
livyConf.set(LIVY_SPARK_SCALA_VERSION.key,
95+
formatScalaVersion(scalaVersion, formattedSparkVersion))
6896

6997
testRecovery(livyConf)
7098

‎server/src/main/scala/com/cloudera/livy/server/interactive/InteractiveSession.scala

+12-6
Original file line numberDiff line numberDiff line change
@@ -121,12 +121,18 @@ class InteractiveSession(
121121
}
122122
builderProperties.put(RSCConf.Entry.SESSION_KIND.key, kind.toString)
123123

124-
mergeConfList(livyJars(livyConf), LivyConf.SPARK_JARS)
124+
require(livyConf.get(LivyConf.LIVY_SPARK_VERSION) != null)
125+
require(livyConf.get(LivyConf.LIVY_SPARK_SCALA_VERSION) != null)
126+
127+
val (sparkMajorVersion, _) =
128+
LivySparkUtils.formatSparkVersion(livyConf.get(LivyConf.LIVY_SPARK_VERSION))
129+
val scalaVersion = livyConf.get(LivyConf.LIVY_SPARK_SCALA_VERSION)
130+
131+
mergeConfList(livyJars(livyConf, scalaVersion), LivyConf.SPARK_JARS)
125132
val enableHiveContext = livyConf.getBoolean(LivyConf.ENABLE_HIVE_CONTEXT)
126-
val sparkMajorVersion =
127-
LivySparkUtils.formatSparkVersion(LivySparkUtils.sparkSubmitVersion(livyConf))._1
128133
// pass spark.livy.spark_major_version to driver
129134
builderProperties.put("spark.livy.spark_major_version", sparkMajorVersion.toString)
135+
130136
if (sparkMajorVersion <= 1) {
131137
builderProperties.put("spark.repl.enableHiveContext",
132138
livyConf.getBoolean(LivyConf.ENABLE_HIVE_CONTEXT).toString)
@@ -321,12 +327,12 @@ class InteractiveSession(
321327
}
322328
}
323329

324-
private def livyJars(livyConf: LivyConf): List[String] = {
330+
private def livyJars(livyConf: LivyConf, scalaVersion: String): List[String] = {
325331
Option(livyConf.get(LivyReplJars)).map(_.split(",").toList).getOrElse {
326332
val home = sys.env("LIVY_HOME")
327-
val jars = Option(new File(home, "repl-jars"))
333+
val jars = Option(new File(home, s"repl_$scalaVersion-jars"))
328334
.filter(_.isDirectory())
329-
.getOrElse(new File(home, "repl/scala-2.10/target/jars"))
335+
.getOrElse(new File(home, s"repl/scala-$scalaVersion/target/jars"))
330336
require(jars.isDirectory(), "Cannot find Livy REPL jars.")
331337
jars.listFiles().map(_.getAbsolutePath()).toList
332338
}

‎server/src/main/scala/com/cloudera/livy/utils/LivySparkUtils.scala

+55-16
Original file line numberDiff line numberDiff line change
@@ -27,6 +27,22 @@ import com.cloudera.livy.util.LineBufferedProcess
2727

2828
object LivySparkUtils {
2929

30+
// For each Spark version we supported, we need to add this mapping relation in case Scala
31+
// version cannot be detected from "spark-submit --version".
32+
private val defaultSparkScalaVersion = Map(
33+
// Spark 2.0 + Scala 2.11
34+
(2, 0) -> "2.11",
35+
// Spark 1.6 + Scala 2.10
36+
(1, 6) -> "2.10"
37+
)
38+
39+
// Supported Spark version
40+
private val MIN_VERSION = (1, 6)
41+
private val MAX_VERSION = (2, 1)
42+
43+
private val sparkVersionRegex = """version (.*)""".r.unanchored
44+
private val scalaVersionRegex = """Scala version (.*), Java""".r.unanchored
45+
3046
/**
3147
* Test that Spark home is configured and configured Spark home is a directory.
3248
*/
@@ -45,7 +61,7 @@ object LivySparkUtils {
4561
*/
4662
def testSparkSubmit(livyConf: LivyConf): Unit = {
4763
try {
48-
testSparkVersion(sparkSubmitVersion(livyConf))
64+
testSparkVersion(sparkSubmitVersion(livyConf)._1)
4965
} catch {
5066
case e: IOException =>
5167
throw new IOException("Failed to run spark-submit executable", e)
@@ -57,25 +73,21 @@ object LivySparkUtils {
5773
* @param version Spark version
5874
*/
5975
def testSparkVersion(version: String): Unit = {
60-
// This is exclusive. Version which equals to this will be rejected.
61-
val maxVersion = (2, 1)
62-
val minVersion = (1, 6)
63-
6476
val supportedVersion = formatSparkVersion(version) match {
6577
case v: (Int, Int) =>
66-
v >= minVersion && v < maxVersion
78+
v >= MIN_VERSION && v < MAX_VERSION
6779
case _ => false
6880
}
6981
require(supportedVersion, s"Unsupported Spark version $version.")
7082
}
7183

7284
/**
73-
* Return the version of the configured `spark-submit` version.
85+
* Return the Spark and Scala version of the configured `spark-submit` version.
7486
*
7587
* @param livyConf
76-
* @return the version
88+
* @return Tuple with Spark and Scala version
7789
*/
78-
def sparkSubmitVersion(livyConf: LivyConf): String = {
90+
def sparkSubmitVersion(livyConf: LivyConf): (String, String) = {
7991
val sparkSubmit = livyConf.sparkSubmit()
8092
val pb = new ProcessBuilder(sparkSubmit, "--version")
8193
pb.redirectErrorStream(true)
@@ -89,20 +101,28 @@ object LivySparkUtils {
89101
val exitCode = process.waitFor()
90102
val output = process.inputIterator.mkString("\n")
91103

92-
val regex = """version (.*)""".r.unanchored
93-
104+
var sparkVersion = ""
94105
output match {
95-
case regex(version) => version
106+
case sparkVersionRegex(version) => sparkVersion = version
96107
case _ =>
97108
throw new IOException(f"Unable to determine spark-submit version [$exitCode]:\n$output")
98109
}
110+
111+
var scalaVersion = ""
112+
output match {
113+
case scalaVersionRegex(version) => scalaVersion = version
114+
case _ =>
115+
}
116+
117+
(sparkVersion, scalaVersion)
99118
}
100119

101120
/**
102-
* Return formatted Spark version.
103-
* @param version Spark version
104-
* @return Two element tuple, one is major version and the other is minor version
105-
*/
121+
* Return formatted Spark version.
122+
*
123+
* @param version Spark version
124+
* @return Two element tuple, one is major version and the other is minor version
125+
*/
106126
def formatSparkVersion(version: String): (Int, Int) = {
107127
val versionPattern = """(\d)+\.(\d)+(?:[\.-]\d*)*""".r
108128
version match {
@@ -112,4 +132,23 @@ object LivySparkUtils {
112132
throw new IllegalArgumentException(s"Fail to parse Spark version from $version")
113133
}
114134
}
135+
136+
/**
137+
* Return Scala binary version, if it cannot be parsed from input version string, it will
138+
* pick default Scala version related to Spark version.
139+
*
140+
* @param scalaVersion Scala binary version String
141+
* @param sparkVersion formatted Spark version.
142+
* @return Scala binary version String based on Spark version and livy conf.
143+
*/
144+
def formatScalaVersion(scalaVersion: String, sparkVersion: (Int, Int)): String = {
145+
val versionPattern = """(\d)+\.(\d+)+.*""".r
146+
scalaVersion match {
147+
case versionPattern(major, minor) =>
148+
major + "." + minor
149+
case _ =>
150+
defaultSparkScalaVersion.getOrElse(sparkVersion,
151+
throw new IllegalArgumentException(s"Fail to get Scala version from Spark $sparkVersion"))
152+
}
153+
}
115154
}

‎server/src/test/scala/com/cloudera/livy/server/interactive/BaseInteractiveServletSpec.scala

+2
Original file line numberDiff line numberDiff line change
@@ -50,6 +50,8 @@ abstract class BaseInteractiveServletSpec extends BaseSessionServletSpec[Interac
5050
super.createConf()
5151
.set(LivyConf.SESSION_STAGING_DIR, tempDir.toURI().toString())
5252
.set(InteractiveSession.LivyReplJars, "")
53+
.set(LivyConf.LIVY_SPARK_VERSION, "1.6.0")
54+
.set(LivyConf.LIVY_SPARK_SCALA_VERSION, "2.10.5")
5355
}
5456

5557
protected def createRequest(

‎server/src/test/scala/com/cloudera/livy/server/interactive/InteractiveSessionSpec.scala

+2
Original file line numberDiff line numberDiff line change
@@ -38,6 +38,8 @@ class InteractiveSessionSpec extends FunSpec
3838

3939
private val livyConf = new LivyConf()
4040
livyConf.set(InteractiveSession.LivyReplJars, "")
41+
.set(LivyConf.LIVY_SPARK_VERSION, "1.6.0")
42+
.set(LivyConf.LIVY_SPARK_SCALA_VERSION, "2.10.5")
4143

4244
implicit val formats = DefaultFormats
4345

‎server/src/test/scala/com/cloudera/livy/utils/LivySparkUtilsSuite.scala

+21-1
Original file line numberDiff line numberDiff line change
@@ -19,11 +19,12 @@
1919
package com.cloudera.livy.utils
2020

2121
import org.scalatest.FunSuite
22+
import org.scalatest.Matchers
2223

2324
import com.cloudera.livy.{LivyBaseUnitTestSuite, LivyConf}
2425
import com.cloudera.livy.server.LivyServer
2526

26-
class LivySparkUtilsSuite extends FunSuite with LivyBaseUnitTestSuite {
27+
class LivySparkUtilsSuite extends FunSuite with Matchers with LivyBaseUnitTestSuite {
2728

2829
import LivySparkUtils._
2930

@@ -41,13 +42,16 @@ class LivySparkUtilsSuite extends FunSuite with LivyBaseUnitTestSuite {
4142
testSparkVersion("1.6.0")
4243
testSparkVersion("1.6.1")
4344
testSparkVersion("1.6.2")
45+
testSparkVersion("1.6")
46+
testSparkVersion("1.6.3.2.5.0-12")
4447
}
4548

4649
test("should support Spark 2.0.x") {
4750
testSparkVersion("2.0.0")
4851
testSparkVersion("2.0.1")
4952
testSparkVersion("2.0.2")
5053
testSparkVersion("2.0.0.2.5.1.0-56") // LIVY-229
54+
testSparkVersion("2.0")
5155
}
5256

5357
test("should not support Spark older than 1.6 or newer than 2.0") {
@@ -72,4 +76,20 @@ class LivySparkUtilsSuite extends FunSuite with LivyBaseUnitTestSuite {
7276
val s = new LivyServer()
7377
intercept[IllegalArgumentException] { s.testRecovery(livyConf) }
7478
}
79+
80+
test("get correct Scala version") {
81+
formatScalaVersion("2.10.8", formatSparkVersion("2.0.0")) should be ("2.10")
82+
formatScalaVersion("2.11.4", formatSparkVersion("1.6.0")) should be ("2.11")
83+
formatScalaVersion("2.10", formatSparkVersion("2.0.0")) should be ("2.10")
84+
formatScalaVersion("2.10.x.x.x.x", formatSparkVersion("2.0.0")) should be ("2.10")
85+
86+
// Will pick default Spark Scala version if the input Scala version string is not correct.
87+
formatScalaVersion("", formatSparkVersion("2.0.0")) should be ("2.11")
88+
formatScalaVersion("xxx", formatSparkVersion("1.6.1")) should be ("2.10")
89+
90+
// Throw exception for unsupported Spark version.
91+
intercept[IllegalArgumentException] {
92+
formatScalaVersion("xxx", formatSparkVersion("1.5.0"))
93+
}
94+
}
7595
}

0 commit comments

Comments
 (0)