You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@livy.apache.org by js...@apache.org on 2017/07/05 03:06:56 UTC
[27/33] incubator-livy git commit: LIVY-375. Change Livy code package
name to org.apache.livy
http://git-wip-us.apache.org/repos/asf/incubator-livy/blob/412ccc8f/repl/scala-2.10/src/test/scala/org/apache/livy/repl/SparkInterpreterSpec.scala
----------------------------------------------------------------------
diff --git a/repl/scala-2.10/src/test/scala/org/apache/livy/repl/SparkInterpreterSpec.scala b/repl/scala-2.10/src/test/scala/org/apache/livy/repl/SparkInterpreterSpec.scala
new file mode 100644
index 0000000..e3c849d
--- /dev/null
+++ b/repl/scala-2.10/src/test/scala/org/apache/livy/repl/SparkInterpreterSpec.scala
@@ -0,0 +1,86 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.livy.repl
+
+import org.scalatest._
+
+import org.apache.livy.LivyBaseUnitTestSuite
+
+class SparkInterpreterSpec extends FunSpec with Matchers with LivyBaseUnitTestSuite {
+ describe("SparkInterpreter") {
+ val interpreter = new SparkInterpreter(null)
+
+ it("should parse Scala compile error.") {
+ // Regression test for LIVY-260.
+ val error =
+ """<console>:27: error: type mismatch;
+ | found : Int
+ | required: String
+ | sc.setJobGroup(groupName, groupName, true)
+ | ^
+ |<console>:27: error: type mismatch;
+ | found : Int
+ | required: String
+ | sc.setJobGroup(groupName, groupName, true)
+ | ^
+ |""".stripMargin
+
+ val expectedTraceback = AbstractSparkInterpreter.KEEP_NEWLINE_REGEX.split(
+ """ found : Int
+ | required: String
+ | sc.setJobGroup(groupName, groupName, true)
+ | ^
+ |<console>:27: error: type mismatch;
+ | found : Int
+ | required: String
+ | sc.setJobGroup(groupName, groupName, true)
+ | ^
+ |""".stripMargin)
+
+ val (ename, traceback) = interpreter.parseError(error)
+ ename shouldBe "<console>:27: error: type mismatch;"
+ traceback shouldBe expectedTraceback
+ }
+
+ it("should parse Scala runtime error and remove internal frames.") {
+ val error =
+ """java.lang.RuntimeException: message
+ | at $iwC$$iwC$$iwC$$iwC$$iwC.error(<console>:25)
+ | at $iwC$$iwC$$iwC.error2(<console>:27)
+ | at $iwC$$iwC.<init>(<console>:41)
+ | at $iwC.<init>(<console>:43)
+ | at <init>(<console>:45)
+ | at .<init>(<console>:49)
+ | at .<clinit>(<console>)
+ | at .<init>(<console>:7)
+ | at .<clinit>(<console>)
+ | at $print(<console>)
+ | at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
+ |""".stripMargin
+
+ val expectedTraceback = AbstractSparkInterpreter.KEEP_NEWLINE_REGEX.split(
+ """ at <user code>.error(<console>:25)
+ | at <user code>.error2(<console>:27)
+ |""".stripMargin)
+
+ val (ename, traceback) = interpreter.parseError(error)
+ ename shouldBe "java.lang.RuntimeException: message"
+ traceback shouldBe expectedTraceback
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-livy/blob/412ccc8f/repl/scala-2.11/pom.xml
----------------------------------------------------------------------
diff --git a/repl/scala-2.11/pom.xml b/repl/scala-2.11/pom.xml
index 745997c..97f5c54 100644
--- a/repl/scala-2.11/pom.xml
+++ b/repl/scala-2.11/pom.xml
@@ -19,15 +19,15 @@
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
- <groupId>com.cloudera.livy</groupId>
+ <groupId>org.apache.livy</groupId>
<artifactId>livy-repl_2.11</artifactId>
- <version>0.4.0-SNAPSHOT</version>
+ <version>0.4.0-incubating-SNAPSHOT</version>
<packaging>jar</packaging>
<parent>
- <groupId>com.cloudera.livy</groupId>
+ <groupId>org.apache.livy</groupId>
<artifactId>livy-repl-parent</artifactId>
- <version>0.4.0-SNAPSHOT</version>
+ <version>0.4.0-incubating-SNAPSHOT</version>
<relativePath>../pom.xml</relativePath>
</parent>
http://git-wip-us.apache.org/repos/asf/incubator-livy/blob/412ccc8f/repl/scala-2.11/src/main/scala/com/cloudera/livy/repl/SparkInterpreter.scala
----------------------------------------------------------------------
diff --git a/repl/scala-2.11/src/main/scala/com/cloudera/livy/repl/SparkInterpreter.scala b/repl/scala-2.11/src/main/scala/com/cloudera/livy/repl/SparkInterpreter.scala
deleted file mode 100644
index 925d477..0000000
--- a/repl/scala-2.11/src/main/scala/com/cloudera/livy/repl/SparkInterpreter.scala
+++ /dev/null
@@ -1,174 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package com.cloudera.livy.repl
-
-import java.io.File
-import java.net.URLClassLoader
-import java.nio.file.{Files, Paths}
-
-import scala.tools.nsc.Settings
-import scala.tools.nsc.interpreter.JPrintWriter
-import scala.tools.nsc.interpreter.Results.Result
-import scala.util.control.NonFatal
-
-import org.apache.spark.{SparkConf, SparkContext}
-import org.apache.spark.repl.SparkILoop
-
-/**
- * Scala 2.11 version of SparkInterpreter
- */
-class SparkInterpreter(conf: SparkConf)
- extends AbstractSparkInterpreter with SparkContextInitializer {
-
- protected var sparkContext: SparkContext = _
- private var sparkILoop: SparkILoop = _
- private var sparkHttpServer: Object = _
-
- override def start(): SparkContext = {
- require(sparkILoop == null)
-
- val rootDir = conf.get("spark.repl.classdir", System.getProperty("java.io.tmpdir"))
- val outputDir = Files.createTempDirectory(Paths.get(rootDir), "spark").toFile
- outputDir.deleteOnExit()
- conf.set("spark.repl.class.outputDir", outputDir.getAbsolutePath)
-
- // Only Spark1 requires to create http server, Spark2 removes HttpServer class.
- startHttpServer(outputDir).foreach { case (server, uri) =>
- sparkHttpServer = server
- conf.set("spark.repl.class.uri", uri)
- }
-
- val settings = new Settings()
- settings.processArguments(List("-Yrepl-class-based",
- "-Yrepl-outdir", s"${outputDir.getAbsolutePath}"), true)
- settings.usejavacp.value = true
- settings.embeddedDefaults(Thread.currentThread().getContextClassLoader())
-
- sparkILoop = new SparkILoop(None, new JPrintWriter(outputStream, true))
- sparkILoop.settings = settings
- sparkILoop.createInterpreter()
- sparkILoop.initializeSynchronous()
-
- restoreContextClassLoader {
- sparkILoop.setContextClassLoader()
-
- var classLoader = Thread.currentThread().getContextClassLoader
- while (classLoader != null) {
- if (classLoader.getClass.getCanonicalName ==
- "org.apache.spark.util.MutableURLClassLoader") {
- val extraJarPath = classLoader.asInstanceOf[URLClassLoader].getURLs()
- // Check if the file exists. Otherwise an exception will be thrown.
- .filter { u => u.getProtocol == "file" && new File(u.getPath).isFile }
- // Livy rsc and repl are also in the extra jars list. Filter them out.
- .filterNot { u => Paths.get(u.toURI).getFileName.toString.startsWith("livy-") }
- // Some bad spark packages depend on the wrong version of scala-reflect. Blacklist it.
- .filterNot { u =>
- Paths.get(u.toURI).getFileName.toString.contains("org.scala-lang_scala-reflect")
- }
-
- extraJarPath.foreach { p => debug(s"Adding $p to Scala interpreter's class path...") }
- sparkILoop.addUrlsToClassPath(extraJarPath: _*)
- classLoader = null
- } else {
- classLoader = classLoader.getParent
- }
- }
-
- createSparkContext(conf)
- }
-
- sparkContext
- }
-
- override def close(): Unit = synchronized {
- if (sparkContext != null) {
- sparkContext.stop()
- sparkContext = null
- }
-
- if (sparkILoop != null) {
- sparkILoop.closeInterpreter()
- sparkILoop = null
- }
-
- if (sparkHttpServer != null) {
- val method = sparkHttpServer.getClass.getMethod("stop")
- method.setAccessible(true)
- method.invoke(sparkHttpServer)
- sparkHttpServer = null
- }
- }
-
- override protected def isStarted(): Boolean = {
- sparkContext != null && sparkILoop != null
- }
-
- override protected def interpret(code: String): Result = {
- sparkILoop.interpret(code)
- }
-
- override protected def valueOfTerm(name: String): Option[Any] = {
- // IMain#valueOfTerm will always return None, so use other way instead.
- Option(sparkILoop.lastRequest.lineRep.call("$result"))
- }
-
- protected def bind(name: String, tpe: String, value: Object, modifier: List[String]): Unit = {
- sparkILoop.beQuietDuring {
- sparkILoop.bind(name, tpe, value, modifier)
- }
- }
-
- private def startHttpServer(outputDir: File): Option[(Object, String)] = {
- try {
- val httpServerClass = Class.forName("org.apache.spark.HttpServer")
- val securityManager = {
- val constructor = Class.forName("org.apache.spark.SecurityManager")
- .getConstructor(classOf[SparkConf])
- constructor.setAccessible(true)
- constructor.newInstance(conf).asInstanceOf[Object]
- }
- val httpServerConstructor = httpServerClass
- .getConstructor(classOf[SparkConf],
- classOf[File],
- Class.forName("org.apache.spark.SecurityManager"),
- classOf[Int],
- classOf[String])
- httpServerConstructor.setAccessible(true)
- // Create Http Server
- val port = conf.getInt("spark.replClassServer.port", 0)
- val server = httpServerConstructor
- .newInstance(conf, outputDir, securityManager, new Integer(port), "HTTP server")
- .asInstanceOf[Object]
-
- // Start Http Server
- val startMethod = server.getClass.getMethod("start")
- startMethod.setAccessible(true)
- startMethod.invoke(server)
-
- // Get uri of this Http Server
- val uriMethod = server.getClass.getMethod("uri")
- uriMethod.setAccessible(true)
- val uri = uriMethod.invoke(server).asInstanceOf[String]
- Some((server, uri))
- } catch {
- // Spark 2.0+ removed HttpServer, so return null instead.
- case NonFatal(e) =>
- None
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-livy/blob/412ccc8f/repl/scala-2.11/src/main/scala/org/apache/livy/repl/SparkInterpreter.scala
----------------------------------------------------------------------
diff --git a/repl/scala-2.11/src/main/scala/org/apache/livy/repl/SparkInterpreter.scala b/repl/scala-2.11/src/main/scala/org/apache/livy/repl/SparkInterpreter.scala
new file mode 100644
index 0000000..94cd241
--- /dev/null
+++ b/repl/scala-2.11/src/main/scala/org/apache/livy/repl/SparkInterpreter.scala
@@ -0,0 +1,174 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.livy.repl
+
+import java.io.File
+import java.net.URLClassLoader
+import java.nio.file.{Files, Paths}
+
+import scala.tools.nsc.Settings
+import scala.tools.nsc.interpreter.JPrintWriter
+import scala.tools.nsc.interpreter.Results.Result
+import scala.util.control.NonFatal
+
+import org.apache.spark.{SparkConf, SparkContext}
+import org.apache.spark.repl.SparkILoop
+
+/**
+ * Scala 2.11 version of SparkInterpreter
+ */
+class SparkInterpreter(conf: SparkConf)
+ extends AbstractSparkInterpreter with SparkContextInitializer {
+
+ protected var sparkContext: SparkContext = _
+ private var sparkILoop: SparkILoop = _
+ private var sparkHttpServer: Object = _
+
+ override def start(): SparkContext = {
+ require(sparkILoop == null)
+
+ val rootDir = conf.get("spark.repl.classdir", System.getProperty("java.io.tmpdir"))
+ val outputDir = Files.createTempDirectory(Paths.get(rootDir), "spark").toFile
+ outputDir.deleteOnExit()
+ conf.set("spark.repl.class.outputDir", outputDir.getAbsolutePath)
+
+ // Only Spark1 requires to create http server, Spark2 removes HttpServer class.
+ startHttpServer(outputDir).foreach { case (server, uri) =>
+ sparkHttpServer = server
+ conf.set("spark.repl.class.uri", uri)
+ }
+
+ val settings = new Settings()
+ settings.processArguments(List("-Yrepl-class-based",
+ "-Yrepl-outdir", s"${outputDir.getAbsolutePath}"), true)
+ settings.usejavacp.value = true
+ settings.embeddedDefaults(Thread.currentThread().getContextClassLoader())
+
+ sparkILoop = new SparkILoop(None, new JPrintWriter(outputStream, true))
+ sparkILoop.settings = settings
+ sparkILoop.createInterpreter()
+ sparkILoop.initializeSynchronous()
+
+ restoreContextClassLoader {
+ sparkILoop.setContextClassLoader()
+
+ var classLoader = Thread.currentThread().getContextClassLoader
+ while (classLoader != null) {
+ if (classLoader.getClass.getCanonicalName ==
+ "org.apache.spark.util.MutableURLClassLoader") {
+ val extraJarPath = classLoader.asInstanceOf[URLClassLoader].getURLs()
+ // Check if the file exists. Otherwise an exception will be thrown.
+ .filter { u => u.getProtocol == "file" && new File(u.getPath).isFile }
+ // Livy rsc and repl are also in the extra jars list. Filter them out.
+ .filterNot { u => Paths.get(u.toURI).getFileName.toString.startsWith("livy-") }
+ // Some bad spark packages depend on the wrong version of scala-reflect. Blacklist it.
+ .filterNot { u =>
+ Paths.get(u.toURI).getFileName.toString.contains("org.scala-lang_scala-reflect")
+ }
+
+ extraJarPath.foreach { p => debug(s"Adding $p to Scala interpreter's class path...") }
+ sparkILoop.addUrlsToClassPath(extraJarPath: _*)
+ classLoader = null
+ } else {
+ classLoader = classLoader.getParent
+ }
+ }
+
+ createSparkContext(conf)
+ }
+
+ sparkContext
+ }
+
+ override def close(): Unit = synchronized {
+ if (sparkContext != null) {
+ sparkContext.stop()
+ sparkContext = null
+ }
+
+ if (sparkILoop != null) {
+ sparkILoop.closeInterpreter()
+ sparkILoop = null
+ }
+
+ if (sparkHttpServer != null) {
+ val method = sparkHttpServer.getClass.getMethod("stop")
+ method.setAccessible(true)
+ method.invoke(sparkHttpServer)
+ sparkHttpServer = null
+ }
+ }
+
+ override protected def isStarted(): Boolean = {
+ sparkContext != null && sparkILoop != null
+ }
+
+ override protected def interpret(code: String): Result = {
+ sparkILoop.interpret(code)
+ }
+
+ override protected def valueOfTerm(name: String): Option[Any] = {
+ // IMain#valueOfTerm will always return None, so use other way instead.
+ Option(sparkILoop.lastRequest.lineRep.call("$result"))
+ }
+
+ protected def bind(name: String, tpe: String, value: Object, modifier: List[String]): Unit = {
+ sparkILoop.beQuietDuring {
+ sparkILoop.bind(name, tpe, value, modifier)
+ }
+ }
+
+ private def startHttpServer(outputDir: File): Option[(Object, String)] = {
+ try {
+ val httpServerClass = Class.forName("org.apache.spark.HttpServer")
+ val securityManager = {
+ val constructor = Class.forName("org.apache.spark.SecurityManager")
+ .getConstructor(classOf[SparkConf])
+ constructor.setAccessible(true)
+ constructor.newInstance(conf).asInstanceOf[Object]
+ }
+ val httpServerConstructor = httpServerClass
+ .getConstructor(classOf[SparkConf],
+ classOf[File],
+ Class.forName("org.apache.spark.SecurityManager"),
+ classOf[Int],
+ classOf[String])
+ httpServerConstructor.setAccessible(true)
+ // Create Http Server
+ val port = conf.getInt("spark.replClassServer.port", 0)
+ val server = httpServerConstructor
+ .newInstance(conf, outputDir, securityManager, new Integer(port), "HTTP server")
+ .asInstanceOf[Object]
+
+ // Start Http Server
+ val startMethod = server.getClass.getMethod("start")
+ startMethod.setAccessible(true)
+ startMethod.invoke(server)
+
+ // Get uri of this Http Server
+ val uriMethod = server.getClass.getMethod("uri")
+ uriMethod.setAccessible(true)
+ val uri = uriMethod.invoke(server).asInstanceOf[String]
+ Some((server, uri))
+ } catch {
+ // Spark 2.0+ removed HttpServer, so return null instead.
+ case NonFatal(e) =>
+ None
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-livy/blob/412ccc8f/repl/scala-2.11/src/test/scala/com/cloudera/livy/repl/SparkInterpreterSpec.scala
----------------------------------------------------------------------
diff --git a/repl/scala-2.11/src/test/scala/com/cloudera/livy/repl/SparkInterpreterSpec.scala b/repl/scala-2.11/src/test/scala/com/cloudera/livy/repl/SparkInterpreterSpec.scala
deleted file mode 100644
index a2496a1..0000000
--- a/repl/scala-2.11/src/test/scala/com/cloudera/livy/repl/SparkInterpreterSpec.scala
+++ /dev/null
@@ -1,68 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package com.cloudera.livy.repl
-
-import org.scalatest._
-
-import com.cloudera.livy.LivyBaseUnitTestSuite
-
-class SparkInterpreterSpec extends FunSpec with Matchers with LivyBaseUnitTestSuite {
- describe("SparkInterpreter") {
- val interpreter = new SparkInterpreter(null)
-
- it("should parse Scala compile error.") {
- // Regression test for LIVY-.
- val error =
- """<console>:27: error: type mismatch;
- | found : Int
- | required: String
- | sc.setJobGroup(groupName, groupName, true)
- | ^
- |<console>:27: error: type mismatch;
- | found : Int
- | required: String
- | sc.setJobGroup(groupName, groupName, true)
- | ^
- |""".stripMargin
-
- val parsedError = AbstractSparkInterpreter.KEEP_NEWLINE_REGEX.split(error)
-
- val expectedTraceback = parsedError.tail
-
- val (ename, traceback) = interpreter.parseError(error)
- ename shouldBe "<console>:27: error: type mismatch;"
- traceback shouldBe expectedTraceback
- }
-
- it("should parse Scala runtime error.") {
- val error =
- """java.lang.RuntimeException: message
- | ... 48 elided
- |
- |Tailing message""".stripMargin
-
- val parsedError = AbstractSparkInterpreter.KEEP_NEWLINE_REGEX.split(error)
-
- val expectedTraceback = parsedError.tail
-
- val (ename, traceback) = interpreter.parseError(error)
- ename shouldBe "java.lang.RuntimeException: message"
- traceback shouldBe expectedTraceback
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-livy/blob/412ccc8f/repl/scala-2.11/src/test/scala/org/apache/livy/repl/SparkInterpreterSpec.scala
----------------------------------------------------------------------
diff --git a/repl/scala-2.11/src/test/scala/org/apache/livy/repl/SparkInterpreterSpec.scala b/repl/scala-2.11/src/test/scala/org/apache/livy/repl/SparkInterpreterSpec.scala
new file mode 100644
index 0000000..d922034
--- /dev/null
+++ b/repl/scala-2.11/src/test/scala/org/apache/livy/repl/SparkInterpreterSpec.scala
@@ -0,0 +1,68 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.livy.repl
+
+import org.scalatest._
+
+import org.apache.livy.LivyBaseUnitTestSuite
+
+class SparkInterpreterSpec extends FunSpec with Matchers with LivyBaseUnitTestSuite {
+ describe("SparkInterpreter") {
+ val interpreter = new SparkInterpreter(null)
+
+ it("should parse Scala compile error.") {
+ // Regression test for LIVY-.
+ val error =
+ """<console>:27: error: type mismatch;
+ | found : Int
+ | required: String
+ | sc.setJobGroup(groupName, groupName, true)
+ | ^
+ |<console>:27: error: type mismatch;
+ | found : Int
+ | required: String
+ | sc.setJobGroup(groupName, groupName, true)
+ | ^
+ |""".stripMargin
+
+ val parsedError = AbstractSparkInterpreter.KEEP_NEWLINE_REGEX.split(error)
+
+ val expectedTraceback = parsedError.tail
+
+ val (ename, traceback) = interpreter.parseError(error)
+ ename shouldBe "<console>:27: error: type mismatch;"
+ traceback shouldBe expectedTraceback
+ }
+
+ it("should parse Scala runtime error.") {
+ val error =
+ """java.lang.RuntimeException: message
+ | ... 48 elided
+ |
+ |Tailing message""".stripMargin
+
+ val parsedError = AbstractSparkInterpreter.KEEP_NEWLINE_REGEX.split(error)
+
+ val expectedTraceback = parsedError.tail
+
+ val (ename, traceback) = interpreter.parseError(error)
+ ename shouldBe "java.lang.RuntimeException: message"
+ traceback shouldBe expectedTraceback
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-livy/blob/412ccc8f/repl/src/main/resources/fake_shell.py
----------------------------------------------------------------------
diff --git a/repl/src/main/resources/fake_shell.py b/repl/src/main/resources/fake_shell.py
index 1a8e24e..534e0df 100644
--- a/repl/src/main/resources/fake_shell.py
+++ b/repl/src/main/resources/fake_shell.py
@@ -174,7 +174,7 @@ class PySparkJobProcessorImpl(object):
return os.path.join(job_context.get_local_tmp_dir_path(), '__livy__')
class Scala:
- extends = ['com.cloudera.livy.repl.PySparkJobProcessor']
+ extends = ['org.apache.livy.repl.PySparkJobProcessor']
class ExecutionError(Exception):
http://git-wip-us.apache.org/repos/asf/incubator-livy/blob/412ccc8f/repl/src/main/scala/com/cloudera/livy/repl/AbstractSparkInterpreter.scala
----------------------------------------------------------------------
diff --git a/repl/src/main/scala/com/cloudera/livy/repl/AbstractSparkInterpreter.scala b/repl/src/main/scala/com/cloudera/livy/repl/AbstractSparkInterpreter.scala
deleted file mode 100644
index cffa69d..0000000
--- a/repl/src/main/scala/com/cloudera/livy/repl/AbstractSparkInterpreter.scala
+++ /dev/null
@@ -1,268 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package com.cloudera.livy.repl
-
-import java.io.ByteArrayOutputStream
-
-import scala.tools.nsc.interpreter.Results
-
-import org.apache.spark.rdd.RDD
-import org.json4s.DefaultFormats
-import org.json4s.Extraction
-import org.json4s.JsonAST._
-import org.json4s.JsonDSL._
-
-import com.cloudera.livy.Logging
-
-object AbstractSparkInterpreter {
- private[repl] val KEEP_NEWLINE_REGEX = """(?<=\n)""".r
- private val MAGIC_REGEX = "^%(\\w+)\\W*(.*)".r
-}
-
-abstract class AbstractSparkInterpreter extends Interpreter with Logging {
- import AbstractSparkInterpreter._
-
- private implicit def formats = DefaultFormats
-
- protected val outputStream = new ByteArrayOutputStream()
-
- final def kind: String = "spark"
-
- protected def isStarted(): Boolean
-
- protected def interpret(code: String): Results.Result
-
- protected def valueOfTerm(name: String): Option[Any]
-
- override protected[repl] def execute(code: String): Interpreter.ExecuteResponse =
- restoreContextClassLoader {
- require(isStarted())
-
- executeLines(code.trim.split("\n").toList, Interpreter.ExecuteSuccess(JObject(
- (TEXT_PLAIN, JString(""))
- )))
- }
-
- private def executeMagic(magic: String, rest: String): Interpreter.ExecuteResponse = {
- magic match {
- case "json" => executeJsonMagic(rest)
- case "table" => executeTableMagic(rest)
- case _ =>
- Interpreter.ExecuteError("UnknownMagic", f"Unknown magic command $magic")
- }
- }
-
- private def executeJsonMagic(name: String): Interpreter.ExecuteResponse = {
- try {
- val value = valueOfTerm(name) match {
- case Some(obj: RDD[_]) => obj.asInstanceOf[RDD[_]].take(10)
- case Some(obj) => obj
- case None => return Interpreter.ExecuteError("NameError", f"Value $name does not exist")
- }
-
- Interpreter.ExecuteSuccess(JObject(
- (APPLICATION_JSON, Extraction.decompose(value))
- ))
- } catch {
- case _: Throwable =>
- Interpreter.ExecuteError("ValueError", "Failed to convert value into a JSON value")
- }
- }
-
- private class TypesDoNotMatch extends Exception
-
- private def convertTableType(value: JValue): String = {
- value match {
- case (JNothing | JNull) => "NULL_TYPE"
- case JBool(_) => "BOOLEAN_TYPE"
- case JString(_) => "STRING_TYPE"
- case JInt(_) => "BIGINT_TYPE"
- case JDouble(_) => "DOUBLE_TYPE"
- case JDecimal(_) => "DECIMAL_TYPE"
- case JArray(arr) =>
- if (allSameType(arr.iterator)) {
- "ARRAY_TYPE"
- } else {
- throw new TypesDoNotMatch
- }
- case JObject(obj) =>
- if (allSameType(obj.iterator.map(_._2))) {
- "MAP_TYPE"
- } else {
- throw new TypesDoNotMatch
- }
- }
- }
-
- private def allSameType(values: Iterator[JValue]): Boolean = {
- if (values.hasNext) {
- val type_name = convertTableType(values.next())
- values.forall { case value => type_name.equals(convertTableType(value)) }
- } else {
- true
- }
- }
-
- private def executeTableMagic(name: String): Interpreter.ExecuteResponse = {
- val value = valueOfTerm(name) match {
- case Some(obj: RDD[_]) => obj.asInstanceOf[RDD[_]].take(10)
- case Some(obj) => obj
- case None => return Interpreter.ExecuteError("NameError", f"Value $name does not exist")
- }
-
- extractTableFromJValue(Extraction.decompose(value))
- }
-
- private def extractTableFromJValue(value: JValue): Interpreter.ExecuteResponse = {
- // Convert the value into JSON and map it to a table.
- val rows: List[JValue] = value match {
- case JArray(arr) => arr
- case _ => List(value)
- }
-
- try {
- val headers = scala.collection.mutable.Map[String, Map[String, String]]()
-
- val data = rows.map { case row =>
- val cols: List[JField] = row match {
- case JArray(arr: List[JValue]) =>
- arr.zipWithIndex.map { case (v, index) => JField(index.toString, v) }
- case JObject(obj) => obj.sortBy(_._1)
- case value: JValue => List(JField("0", value))
- }
-
- cols.map { case (k, v) =>
- val typeName = convertTableType(v)
-
- headers.get(k) match {
- case Some(header) =>
- if (header.get("type").get != typeName) {
- throw new TypesDoNotMatch
- }
- case None =>
- headers.put(k, Map(
- "type" -> typeName,
- "name" -> k
- ))
- }
-
- v
- }
- }
-
- Interpreter.ExecuteSuccess(
- APPLICATION_LIVY_TABLE_JSON -> (
- ("headers" -> headers.toSeq.sortBy(_._1).map(_._2)) ~ ("data" -> data)
- ))
- } catch {
- case _: TypesDoNotMatch =>
- Interpreter.ExecuteError("TypeError", "table rows have different types")
- }
- }
-
- private def executeLines(
- lines: List[String],
- resultFromLastLine: Interpreter.ExecuteResponse): Interpreter.ExecuteResponse = {
- lines match {
- case Nil => resultFromLastLine
- case head :: tail =>
- val result = executeLine(head)
-
- result match {
- case Interpreter.ExecuteIncomplete() =>
- tail match {
- case Nil =>
- // ExecuteIncomplete could be caused by an actual incomplete statements (e.g. "sc.")
- // or statements with just comments.
- // To distinguish them, reissue the same statement wrapped in { }.
- // If it is an actual incomplete statement, the interpreter will return an error.
- // If it is some comment, the interpreter will return success.
- executeLine(s"{\n$head\n}") match {
- case Interpreter.ExecuteIncomplete() | Interpreter.ExecuteError(_, _, _) =>
- // Return the original error so users won't get confusing error message.
- result
- case _ => resultFromLastLine
- }
- case next :: nextTail =>
- executeLines(head + "\n" + next :: nextTail, resultFromLastLine)
- }
- case Interpreter.ExecuteError(_, _, _) =>
- result
-
- case _ =>
- executeLines(tail, result)
- }
- }
- }
-
- private def executeLine(code: String): Interpreter.ExecuteResponse = {
- code match {
- case MAGIC_REGEX(magic, rest) =>
- executeMagic(magic, rest)
- case _ =>
- scala.Console.withOut(outputStream) {
- interpret(code) match {
- case Results.Success =>
- Interpreter.ExecuteSuccess(
- TEXT_PLAIN -> readStdout()
- )
- case Results.Incomplete => Interpreter.ExecuteIncomplete()
- case Results.Error =>
- val (ename, traceback) = parseError(readStdout())
- Interpreter.ExecuteError("Error", ename, traceback)
- }
- }
- }
- }
-
- protected[repl] def parseError(stdout: String): (String, Seq[String]) = {
- // An example of Scala compile error message:
- // <console>:27: error: type mismatch;
- // found : Int
- // required: Boolean
-
- // An example of Scala runtime exception error message:
- // java.lang.RuntimeException: message
- // at .error(<console>:11)
- // ... 32 elided
-
- // Return the first line as ename. Lines following as traceback.
-
- val lines = KEEP_NEWLINE_REGEX.split(stdout)
- val ename = lines.headOption.map(_.trim).getOrElse("unknown error")
- val traceback = lines.tail
-
- (ename, traceback)
- }
-
- protected def restoreContextClassLoader[T](fn: => T): T = {
- val currentClassLoader = Thread.currentThread().getContextClassLoader()
- try {
- fn
- } finally {
- Thread.currentThread().setContextClassLoader(currentClassLoader)
- }
- }
-
- private def readStdout() = {
- val output = outputStream.toString("UTF-8").trim
- outputStream.reset()
-
- output
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-livy/blob/412ccc8f/repl/src/main/scala/com/cloudera/livy/repl/BypassPySparkJob.scala
----------------------------------------------------------------------
diff --git a/repl/src/main/scala/com/cloudera/livy/repl/BypassPySparkJob.scala b/repl/src/main/scala/com/cloudera/livy/repl/BypassPySparkJob.scala
deleted file mode 100644
index 1550129..0000000
--- a/repl/src/main/scala/com/cloudera/livy/repl/BypassPySparkJob.scala
+++ /dev/null
@@ -1,40 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package com.cloudera.livy.repl
-
-import java.nio.charset.StandardCharsets
-
-import com.cloudera.livy.{Job, JobContext}
-
-class BypassPySparkJob(
- serializedJob: Array[Byte],
- replDriver: ReplDriver) extends Job[Array[Byte]] {
-
- override def call(jc: JobContext): Array[Byte] = {
- val interpreter = replDriver.interpreter
- require(interpreter != null && interpreter.isInstanceOf[PythonInterpreter])
- val pi = interpreter.asInstanceOf[PythonInterpreter]
-
- val resultByteArray = pi.pysparkJobProcessor.processBypassJob(serializedJob)
- val resultString = new String(resultByteArray, StandardCharsets.UTF_8)
- if (resultString.startsWith("Client job error:")) {
- throw new PythonJobException(resultString)
- }
- resultByteArray
- }
-}
-
http://git-wip-us.apache.org/repos/asf/incubator-livy/blob/412ccc8f/repl/src/main/scala/com/cloudera/livy/repl/Interpreter.scala
----------------------------------------------------------------------
diff --git a/repl/src/main/scala/com/cloudera/livy/repl/Interpreter.scala b/repl/src/main/scala/com/cloudera/livy/repl/Interpreter.scala
deleted file mode 100644
index 5d56725..0000000
--- a/repl/src/main/scala/com/cloudera/livy/repl/Interpreter.scala
+++ /dev/null
@@ -1,54 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package com.cloudera.livy.repl
-
-import org.apache.spark.SparkContext
-import org.json4s.JObject
-
-object Interpreter {
- abstract class ExecuteResponse
-
- case class ExecuteSuccess(content: JObject) extends ExecuteResponse
- case class ExecuteError(ename: String,
- evalue: String,
- traceback: Seq[String] = Seq()) extends ExecuteResponse
- case class ExecuteIncomplete() extends ExecuteResponse
- case class ExecuteAborted(message: String) extends ExecuteResponse
-}
-
-trait Interpreter {
- import Interpreter._
-
- def kind: String
-
- /**
- * Start the Interpreter.
- *
- * @return A SparkContext
- */
- def start(): SparkContext
-
- /**
- * Execute the code and return the result, it may
- * take some time to execute.
- */
- protected[repl] def execute(code: String): ExecuteResponse
-
- /** Shut down the interpreter. */
- def close(): Unit
-}
http://git-wip-us.apache.org/repos/asf/incubator-livy/blob/412ccc8f/repl/src/main/scala/com/cloudera/livy/repl/ProcessInterpreter.scala
----------------------------------------------------------------------
diff --git a/repl/src/main/scala/com/cloudera/livy/repl/ProcessInterpreter.scala b/repl/src/main/scala/com/cloudera/livy/repl/ProcessInterpreter.scala
deleted file mode 100644
index 924c576..0000000
--- a/repl/src/main/scala/com/cloudera/livy/repl/ProcessInterpreter.scala
+++ /dev/null
@@ -1,137 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package com.cloudera.livy.repl
-
-import java.io.{BufferedReader, InputStreamReader, IOException, PrintWriter}
-import java.util.concurrent.locks.ReentrantLock
-
-import scala.concurrent.Promise
-import scala.io.Source
-
-import org.apache.spark.SparkContext
-import org.json4s.JValue
-
-import com.cloudera.livy.{Logging, Utils}
-import com.cloudera.livy.client.common.ClientConf
-
-private sealed trait Request
-private case class ExecuteRequest(code: String, promise: Promise[JValue]) extends Request
-private case class ShutdownRequest(promise: Promise[Unit]) extends Request
-
-/**
- * Abstract class that describes an interpreter that is running in a separate process.
- *
- * This type is not thread safe, so must be protected by a mutex.
- *
- * @param process
- */
-abstract class ProcessInterpreter(process: Process)
- extends Interpreter with Logging {
- protected[this] val stdin = new PrintWriter(process.getOutputStream)
- protected[this] val stdout = new BufferedReader(new InputStreamReader(process.getInputStream), 1)
-
- override def start(): SparkContext = {
- waitUntilReady()
-
- if (ClientConf.TEST_MODE) {
- null.asInstanceOf[SparkContext]
- } else {
- SparkContext.getOrCreate()
- }
- }
-
- override protected[repl] def execute(code: String): Interpreter.ExecuteResponse = {
- try {
- sendExecuteRequest(code)
- } catch {
- case e: Throwable =>
- Interpreter.ExecuteError(e.getClass.getName, e.getMessage)
- }
- }
-
- override def close(): Unit = {
- if (Utils.isProcessAlive(process)) {
- logger.info("Shutting down process")
- sendShutdownRequest()
-
- try {
- process.getInputStream.close()
- process.getOutputStream.close()
- } catch {
- case _: IOException =>
- }
-
- try {
- process.destroy()
- } finally {
- logger.info("process has been shut down")
- }
- }
- }
-
- protected def sendExecuteRequest(request: String): Interpreter.ExecuteResponse
-
- protected def sendShutdownRequest(): Unit = {}
-
- protected def waitUntilReady(): Unit
-
- private[this] val stderrLock = new ReentrantLock()
- private[this] var stderrLines = Seq[String]()
-
- protected def takeErrorLines(): String = {
- stderrLock.lock()
- try {
- val lines = stderrLines
- stderrLines = Seq()
- lines.mkString("\n")
- } finally {
- stderrLock.unlock()
- }
- }
-
- private[this] val stderrThread = new Thread("process stderr thread") {
- override def run() = {
- val lines = Source.fromInputStream(process.getErrorStream).getLines()
-
- for (line <- lines) {
- stderrLock.lock()
- try {
- stderrLines :+= line
- } finally {
- stderrLock.unlock()
- }
- }
- }
- }
-
- stderrThread.setDaemon(true)
- stderrThread.start()
-
- private[this] val processWatcherThread = new Thread("process watcher thread") {
- override def run() = {
- val exitCode = process.waitFor()
- if (exitCode != 0) {
- error(f"Process has died with $exitCode")
- error(stderrLines.mkString("\n"))
- }
- }
- }
-
- processWatcherThread.setDaemon(true)
- processWatcherThread.start()
-}
http://git-wip-us.apache.org/repos/asf/incubator-livy/blob/412ccc8f/repl/src/main/scala/com/cloudera/livy/repl/PySparkJobProcessor.scala
----------------------------------------------------------------------
diff --git a/repl/src/main/scala/com/cloudera/livy/repl/PySparkJobProcessor.scala b/repl/src/main/scala/com/cloudera/livy/repl/PySparkJobProcessor.scala
deleted file mode 100644
index f0cdbd9..0000000
--- a/repl/src/main/scala/com/cloudera/livy/repl/PySparkJobProcessor.scala
+++ /dev/null
@@ -1,27 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package com.cloudera.livy.repl
-
-trait PySparkJobProcessor {
- def processBypassJob(job: Array[Byte]): Array[Byte]
-
- def addFile(path: String)
-
- def addPyFile(path: String)
-
- def getLocalTmpDirPath: String
-}
http://git-wip-us.apache.org/repos/asf/incubator-livy/blob/412ccc8f/repl/src/main/scala/com/cloudera/livy/repl/PythonInterpreter.scala
----------------------------------------------------------------------
diff --git a/repl/src/main/scala/com/cloudera/livy/repl/PythonInterpreter.scala b/repl/src/main/scala/com/cloudera/livy/repl/PythonInterpreter.scala
deleted file mode 100644
index c744c40..0000000
--- a/repl/src/main/scala/com/cloudera/livy/repl/PythonInterpreter.scala
+++ /dev/null
@@ -1,293 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package com.cloudera.livy.repl
-
-import java.io._
-import java.lang.ProcessBuilder.Redirect
-import java.lang.reflect.Proxy
-import java.nio.file.{Files, Paths}
-
-import scala.annotation.tailrec
-import scala.collection.mutable.ArrayBuffer
-import scala.collection.JavaConverters._
-import scala.util.control.NonFatal
-
-import org.apache.spark.{SparkConf, SparkContext}
-import org.json4s.{DefaultFormats, JValue}
-import org.json4s.JsonAST.JObject
-import org.json4s.jackson.JsonMethods._
-import org.json4s.jackson.Serialization.write
-import py4j._
-import py4j.reflection.PythonProxyHandler
-
-import com.cloudera.livy.Logging
-import com.cloudera.livy.client.common.ClientConf
-import com.cloudera.livy.rsc.BaseProtocol
-import com.cloudera.livy.rsc.driver.BypassJobWrapper
-import com.cloudera.livy.sessions._
-
-// scalastyle:off println
-object PythonInterpreter extends Logging {
-
- def apply(conf: SparkConf, kind: Kind): Interpreter = {
- val pythonExec = kind match {
- case PySpark() => sys.env.getOrElse("PYSPARK_PYTHON", "python")
- case PySpark3() => sys.env.getOrElse("PYSPARK3_PYTHON", "python3")
- case _ => throw new IllegalArgumentException(s"Unknown kind: $kind")
- }
-
- val gatewayServer = new GatewayServer(null, 0)
- gatewayServer.start()
-
- val builder = new ProcessBuilder(Seq(pythonExec, createFakeShell().toString).asJava)
-
- val env = builder.environment()
-
- val pythonPath = sys.env.getOrElse("PYTHONPATH", "")
- .split(File.pathSeparator)
- .++(if (!ClientConf.TEST_MODE) findPySparkArchives() else Nil)
- .++(if (!ClientConf.TEST_MODE) findPyFiles() else Nil)
-
- env.put("PYSPARK_PYTHON", pythonExec)
- env.put("PYTHONPATH", pythonPath.mkString(File.pathSeparator))
- env.put("PYTHONUNBUFFERED", "YES")
- env.put("PYSPARK_GATEWAY_PORT", "" + gatewayServer.getListeningPort)
- env.put("SPARK_HOME", sys.env.getOrElse("SPARK_HOME", "."))
- env.put("LIVY_SPARK_MAJOR_VERSION", conf.get("spark.livy.spark_major_version", "1"))
- builder.redirectError(Redirect.PIPE)
- val process = builder.start()
- new PythonInterpreter(process, gatewayServer, kind.toString)
- }
-
- private def findPySparkArchives(): Seq[String] = {
- sys.env.get("PYSPARK_ARCHIVES_PATH")
- .map(_.split(",").toSeq)
- .getOrElse {
- sys.env.get("SPARK_HOME").map { sparkHome =>
- val pyLibPath = Seq(sparkHome, "python", "lib").mkString(File.separator)
- val pyArchivesFile = new File(pyLibPath, "pyspark.zip")
- require(pyArchivesFile.exists(),
- "pyspark.zip not found; cannot run pyspark application in YARN mode.")
-
- val py4jFile = Files.newDirectoryStream(Paths.get(pyLibPath), "py4j-*-src.zip")
- .iterator()
- .next()
- .toFile
-
- require(py4jFile.exists(),
- "py4j-*-src.zip not found; cannot run pyspark application in YARN mode.")
- Seq(pyArchivesFile.getAbsolutePath, py4jFile.getAbsolutePath)
- }.getOrElse(Seq())
- }
- }
-
- private def findPyFiles(): Seq[String] = {
- val pyFiles = sys.props.getOrElse("spark.submit.pyFiles", "").split(",")
-
- if (sys.env.getOrElse("SPARK_YARN_MODE", "") == "true") {
- // In spark mode, these files have been localized into the current directory.
- pyFiles.map { file =>
- val name = new File(file).getName
- new File(name).getAbsolutePath
- }
- } else {
- pyFiles
- }
- }
-
- private def createFakeShell(): File = {
- val source: InputStream = getClass.getClassLoader.getResourceAsStream("fake_shell.py")
-
- val file = Files.createTempFile("", "").toFile
- file.deleteOnExit()
-
- val sink = new FileOutputStream(file)
- val buf = new Array[Byte](1024)
- var n = source.read(buf)
-
- while (n > 0) {
- sink.write(buf, 0, n)
- n = source.read(buf)
- }
-
- source.close()
- sink.close()
-
- file
- }
-
- private def initiatePy4jCallbackGateway(server: GatewayServer): PySparkJobProcessor = {
- val f = server.getClass.getDeclaredField("gateway")
- f.setAccessible(true)
- val gateway = f.get(server).asInstanceOf[Gateway]
- val command: String = "f" + Protocol.ENTRY_POINT_OBJECT_ID + ";" +
- "com.cloudera.livy.repl.PySparkJobProcessor"
- getPythonProxy(command, gateway).asInstanceOf[PySparkJobProcessor]
- }
-
- // This method is a hack to get around the classLoader issues faced in py4j 0.8.2.1 for
- // dynamically adding jars to the driver. The change is to use the context classLoader instead
- // of the system classLoader when initiating a new Proxy instance
- // ISSUE - https://issues.apache.org/jira/browse/SPARK-6047
- // FIX - https://github.com/bartdag/py4j/pull/196
- private def getPythonProxy(commandPart: String, gateway: Gateway): Any = {
- val proxyString = commandPart.substring(1, commandPart.length)
- val parts = proxyString.split(";")
- val length: Int = parts.length
- val interfaces = ArrayBuffer.fill[Class[_]](length - 1){ null }
- if (length < 2) {
- throw new Py4JException("Invalid Python Proxy.")
- }
- else {
- var proxy: Int = 1
- while (proxy < length) {
- try {
- interfaces(proxy - 1) = Class.forName(parts(proxy))
- if (!interfaces(proxy - 1).isInterface) {
- throw new Py4JException("This class " + parts(proxy) +
- " is not an interface and cannot be used as a Python Proxy.")
- }
- } catch {
- case exception: ClassNotFoundException => {
- throw new Py4JException("Invalid interface name: " + parts(proxy))
- }
- }
- proxy += 1
- }
-
- val pythonProxyHandler = try {
- classOf[PythonProxyHandler].getConstructor(classOf[String], classOf[Gateway])
- .newInstance(parts(0), gateway)
- } catch {
- case NonFatal(e) =>
- classOf[PythonProxyHandler].getConstructor(classOf[String],
- Class.forName("py4j.CallbackClient"), classOf[Gateway])
- .newInstance(parts(0), gateway.getCallbackClient, gateway)
- }
-
- Proxy.newProxyInstance(Thread.currentThread.getContextClassLoader,
- interfaces.toArray, pythonProxyHandler.asInstanceOf[PythonProxyHandler])
- }
- }
-}
-
-private class PythonInterpreter(
- process: Process,
- gatewayServer: GatewayServer,
- pyKind: String)
- extends ProcessInterpreter(process)
- with Logging
-{
- implicit val formats = DefaultFormats
-
- override def kind: String = pyKind
-
- private[repl] val pysparkJobProcessor =
- PythonInterpreter.initiatePy4jCallbackGateway(gatewayServer)
-
- override def close(): Unit = {
- try {
- super.close()
- } finally {
- gatewayServer.shutdown()
- }
- }
-
- @tailrec
- final override protected def waitUntilReady(): Unit = {
- val READY_REGEX = "READY\\(port=([0-9]+)\\)".r
- stdout.readLine() match {
- case null =>
- case READY_REGEX(port) => updatePythonGatewayPort(port.toInt)
- case _ => waitUntilReady()
- }
- }
-
- override protected def sendExecuteRequest(code: String): Interpreter.ExecuteResponse = {
- sendRequest(Map("msg_type" -> "execute_request", "content" -> Map("code" -> code))) match {
- case Some(response) =>
- assert((response \ "msg_type").extract[String] == "execute_reply")
-
- val content = response \ "content"
-
- (content \ "status").extract[String] match {
- case "ok" =>
- Interpreter.ExecuteSuccess((content \ "data").extract[JObject])
- case "error" =>
- val ename = (content \ "ename").extract[String]
- val evalue = (content \ "evalue").extract[String]
- val traceback = (content \ "traceback").extract[Seq[String]]
-
- Interpreter.ExecuteError(ename, evalue, traceback)
- case status =>
- Interpreter.ExecuteError("Internal Error", f"Unknown status $status")
- }
- case None =>
- Interpreter.ExecuteAborted(takeErrorLines())
- }
- }
-
- override protected def sendShutdownRequest(): Unit = {
- sendRequest(Map(
- "msg_type" -> "shutdown_request",
- "content" -> ()
- )).foreach { case rep =>
- warn(f"process failed to shut down while returning $rep")
- }
- }
-
- private def sendRequest(request: Map[String, Any]): Option[JValue] = {
- stdin.println(write(request))
- stdin.flush()
-
- Option(stdout.readLine()).map { case line =>
- parse(line)
- }
- }
-
- def addFile(path: String): Unit = {
- pysparkJobProcessor.addFile(path)
- }
-
- def addPyFile(driver: ReplDriver, conf: SparkConf, path: String): Unit = {
- val localCopyDir = new File(pysparkJobProcessor.getLocalTmpDirPath)
- val localCopyFile = driver.copyFileToLocal(localCopyDir, path, SparkContext.getOrCreate(conf))
- pysparkJobProcessor.addPyFile(localCopyFile.getPath)
- if (path.endsWith(".jar")) {
- driver.addLocalFileToClassLoader(localCopyFile)
- }
- }
-
- private def updatePythonGatewayPort(port: Int): Unit = {
- // The python gateway port can be 0 only when LivyConf.TEST_MODE is true
- // Py4j 0.10 has different API signature for "getCallbackClient", use reflection to handle it.
- if (port != 0) {
- val callbackClient = gatewayServer.getClass
- .getMethod("getCallbackClient")
- .invoke(gatewayServer)
-
- val field = Class.forName("py4j.CallbackClient").getDeclaredField("port")
- field.setAccessible(true)
- field.setInt(callbackClient, port.toInt)
- }
- }
-}
-
-case class PythonJobException(message: String) extends Exception(message) {}
-
-// scalastyle:on println
http://git-wip-us.apache.org/repos/asf/incubator-livy/blob/412ccc8f/repl/src/main/scala/com/cloudera/livy/repl/ReplDriver.scala
----------------------------------------------------------------------
diff --git a/repl/src/main/scala/com/cloudera/livy/repl/ReplDriver.scala b/repl/src/main/scala/com/cloudera/livy/repl/ReplDriver.scala
deleted file mode 100644
index 716e814..0000000
--- a/repl/src/main/scala/com/cloudera/livy/repl/ReplDriver.scala
+++ /dev/null
@@ -1,131 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package com.cloudera.livy.repl
-
-import scala.concurrent.Await
-import scala.concurrent.duration.Duration
-
-import io.netty.channel.ChannelHandlerContext
-import org.apache.spark.SparkConf
-import org.apache.spark.api.java.JavaSparkContext
-
-import com.cloudera.livy.Logging
-import com.cloudera.livy.rsc.{BaseProtocol, ReplJobResults, RSCConf}
-import com.cloudera.livy.rsc.BaseProtocol.ReplState
-import com.cloudera.livy.rsc.driver._
-import com.cloudera.livy.rsc.rpc.Rpc
-import com.cloudera.livy.sessions._
-
-class ReplDriver(conf: SparkConf, livyConf: RSCConf)
- extends RSCDriver(conf, livyConf)
- with Logging {
-
- private[repl] var session: Session = _
-
- private val kind = Kind(livyConf.get(RSCConf.Entry.SESSION_KIND))
-
- private[repl] var interpreter: Interpreter = _
-
- override protected def initializeContext(): JavaSparkContext = {
- interpreter = kind match {
- case PySpark() => PythonInterpreter(conf, PySpark())
- case PySpark3() =>
- PythonInterpreter(conf, PySpark3())
- case Spark() => new SparkInterpreter(conf)
- case SparkR() => SparkRInterpreter(conf)
- }
- session = new Session(livyConf, interpreter, { s => broadcast(new ReplState(s.toString)) })
-
- Option(Await.result(session.start(), Duration.Inf))
- .map(new JavaSparkContext(_))
- .orNull
- }
-
- override protected def shutdownContext(): Unit = {
- if (session != null) {
- try {
- session.close()
- } finally {
- super.shutdownContext()
- }
- }
- }
-
- def handle(ctx: ChannelHandlerContext, msg: BaseProtocol.ReplJobRequest): Int = {
- session.execute(msg.code)
- }
-
- def handle(ctx: ChannelHandlerContext, msg: BaseProtocol.CancelReplJobRequest): Unit = {
- session.cancel(msg.id)
- }
-
- /**
- * Return statement results. Results are sorted by statement id.
- */
- def handle(ctx: ChannelHandlerContext, msg: BaseProtocol.GetReplJobResults): ReplJobResults = {
- val statements = if (msg.allResults) {
- session.statements.values.toArray
- } else {
- assert(msg.from != null)
- assert(msg.size != null)
- if (msg.size == 1) {
- session.statements.get(msg.from).toArray
- } else {
- val until = msg.from + msg.size
- session.statements.filterKeys(id => id >= msg.from && id < until).values.toArray
- }
- }
-
- // Update progress of statements when queried
- statements.foreach { s =>
- s.updateProgress(session.progressOfStatement(s.id))
- }
-
- new ReplJobResults(statements.sortBy(_.id))
- }
-
- override protected def createWrapper(msg: BaseProtocol.BypassJobRequest): BypassJobWrapper = {
- kind match {
- case PySpark() | PySpark3() => new BypassJobWrapper(this, msg.id,
- new BypassPySparkJob(msg.serializedJob, this))
- case _ => super.createWrapper(msg)
- }
- }
-
- override protected def addFile(path: String): Unit = {
- require(interpreter != null)
- interpreter match {
- case pi: PythonInterpreter => pi.addFile(path)
- case _ => super.addFile(path)
- }
- }
-
- override protected def addJarOrPyFile(path: String): Unit = {
- require(interpreter != null)
- interpreter match {
- case pi: PythonInterpreter => pi.addPyFile(this, conf, path)
- case _ => super.addJarOrPyFile(path)
- }
- }
-
- override protected def onClientAuthenticated(client: Rpc): Unit = {
- if (session != null) {
- client.call(new ReplState(session.state.toString))
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-livy/blob/412ccc8f/repl/src/main/scala/com/cloudera/livy/repl/Session.scala
----------------------------------------------------------------------
diff --git a/repl/src/main/scala/com/cloudera/livy/repl/Session.scala b/repl/src/main/scala/com/cloudera/livy/repl/Session.scala
deleted file mode 100644
index 4f17978..0000000
--- a/repl/src/main/scala/com/cloudera/livy/repl/Session.scala
+++ /dev/null
@@ -1,289 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package com.cloudera.livy.repl
-
-import java.util.{LinkedHashMap => JLinkedHashMap}
-import java.util.Map.Entry
-import java.util.concurrent.Executors
-import java.util.concurrent.atomic.AtomicInteger
-
-import scala.collection.JavaConverters._
-import scala.concurrent.{ExecutionContext, Future}
-import scala.concurrent.duration._
-
-import org.apache.spark.SparkContext
-import org.json4s.jackson.JsonMethods.{compact, render}
-import org.json4s.DefaultFormats
-import org.json4s.JsonDSL._
-
-import com.cloudera.livy.Logging
-import com.cloudera.livy.rsc.RSCConf
-import com.cloudera.livy.rsc.driver.{Statement, StatementState}
-import com.cloudera.livy.sessions._
-
-object Session {
- val STATUS = "status"
- val OK = "ok"
- val ERROR = "error"
- val EXECUTION_COUNT = "execution_count"
- val DATA = "data"
- val ENAME = "ename"
- val EVALUE = "evalue"
- val TRACEBACK = "traceback"
-}
-
-class Session(
- livyConf: RSCConf,
- interpreter: Interpreter,
- stateChangedCallback: SessionState => Unit = { _ => })
- extends Logging {
- import Session._
-
- private val interpreterExecutor = ExecutionContext.fromExecutorService(
- Executors.newSingleThreadExecutor())
-
- private val cancelExecutor = ExecutionContext.fromExecutorService(
- Executors.newSingleThreadExecutor())
-
- private implicit val formats = DefaultFormats
-
- @volatile private[repl] var _sc: Option[SparkContext] = None
-
- private var _state: SessionState = SessionState.NotStarted()
-
- // Number of statements kept in driver's memory
- private val numRetainedStatements = livyConf.getInt(RSCConf.Entry.RETAINED_STATEMENT_NUMBER)
-
- private val _statements = new JLinkedHashMap[Int, Statement] {
- protected override def removeEldestEntry(eldest: Entry[Int, Statement]): Boolean = {
- size() > numRetainedStatements
- }
- }.asScala
-
- private val newStatementId = new AtomicInteger(0)
-
- stateChangedCallback(_state)
-
- def start(): Future[SparkContext] = {
- val future = Future {
- changeState(SessionState.Starting())
- val sc = interpreter.start()
- _sc = Option(sc)
- changeState(SessionState.Idle())
- sc
- }(interpreterExecutor)
-
- future.onFailure { case _ => changeState(SessionState.Error()) }(interpreterExecutor)
- future
- }
-
- def kind: String = interpreter.kind
-
- def state: SessionState = _state
-
- def statements: collection.Map[Int, Statement] = _statements.synchronized {
- _statements.toMap
- }
-
- def execute(code: String): Int = {
- val statementId = newStatementId.getAndIncrement()
- val statement = new Statement(statementId, code, StatementState.Waiting, null)
- _statements.synchronized { _statements(statementId) = statement }
-
- Future {
- setJobGroup(statementId)
- statement.compareAndTransit(StatementState.Waiting, StatementState.Running)
-
- if (statement.state.get() == StatementState.Running) {
- statement.output = executeCode(statementId, code)
- }
-
- statement.compareAndTransit(StatementState.Running, StatementState.Available)
- statement.compareAndTransit(StatementState.Cancelling, StatementState.Cancelled)
- statement.updateProgress(1.0)
- }(interpreterExecutor)
-
- statementId
- }
-
- def cancel(statementId: Int): Unit = {
- val statementOpt = _statements.synchronized { _statements.get(statementId) }
- if (statementOpt.isEmpty) {
- return
- }
-
- val statement = statementOpt.get
- if (statement.state.get().isOneOf(
- StatementState.Available, StatementState.Cancelled, StatementState.Cancelling)) {
- return
- } else {
- // statement 1 is running and statement 2 is waiting. User cancels
- // statement 2 then cancels statement 1. The 2nd cancel call will loop and block the 1st
- // cancel call since cancelExecutor is single threaded. To avoid this, set the statement
- // state to cancelled when cancelling a waiting statement.
- statement.compareAndTransit(StatementState.Waiting, StatementState.Cancelled)
- statement.compareAndTransit(StatementState.Running, StatementState.Cancelling)
- }
-
- info(s"Cancelling statement $statementId...")
-
- Future {
- val deadline = livyConf.getTimeAsMs(RSCConf.Entry.JOB_CANCEL_TIMEOUT).millis.fromNow
-
- while (statement.state.get() == StatementState.Cancelling) {
- if (deadline.isOverdue()) {
- info(s"Failed to cancel statement $statementId.")
- statement.compareAndTransit(StatementState.Cancelling, StatementState.Cancelled)
- } else {
- _sc.foreach(_.cancelJobGroup(statementId.toString))
- if (statement.state.get() == StatementState.Cancelling) {
- Thread.sleep(livyConf.getTimeAsMs(RSCConf.Entry.JOB_CANCEL_TRIGGER_INTERVAL))
- }
- }
- }
-
- if (statement.state.get() == StatementState.Cancelled) {
- info(s"Statement $statementId cancelled.")
- }
- }(cancelExecutor)
- }
-
- def close(): Unit = {
- interpreterExecutor.shutdown()
- cancelExecutor.shutdown()
- interpreter.close()
- }
-
- /**
- * Get the current progress of given statement id.
- */
- def progressOfStatement(stmtId: Int): Double = {
- val jobGroup = statementIdToJobGroup(stmtId)
-
- _sc.map { sc =>
- val jobIds = sc.statusTracker.getJobIdsForGroup(jobGroup)
- val jobs = jobIds.flatMap { id => sc.statusTracker.getJobInfo(id) }
- val stages = jobs.flatMap { job =>
- job.stageIds().flatMap(sc.statusTracker.getStageInfo)
- }
-
- val taskCount = stages.map(_.numTasks).sum
- val completedTaskCount = stages.map(_.numCompletedTasks).sum
- if (taskCount == 0) {
- 0.0
- } else {
- completedTaskCount.toDouble / taskCount
- }
- }.getOrElse(0.0)
- }
-
- private def changeState(newState: SessionState): Unit = {
- synchronized {
- _state = newState
- }
- stateChangedCallback(newState)
- }
-
- private def executeCode(executionCount: Int, code: String): String = {
- changeState(SessionState.Busy())
-
- def transitToIdle() = {
- val executingLastStatement = executionCount == newStatementId.intValue() - 1
- if (_statements.isEmpty || executingLastStatement) {
- changeState(SessionState.Idle())
- }
- }
-
- val resultInJson = try {
- interpreter.execute(code) match {
- case Interpreter.ExecuteSuccess(data) =>
- transitToIdle()
-
- (STATUS -> OK) ~
- (EXECUTION_COUNT -> executionCount) ~
- (DATA -> data)
-
- case Interpreter.ExecuteIncomplete() =>
- transitToIdle()
-
- (STATUS -> ERROR) ~
- (EXECUTION_COUNT -> executionCount) ~
- (ENAME -> "Error") ~
- (EVALUE -> "incomplete statement") ~
- (TRACEBACK -> Seq.empty[String])
-
- case Interpreter.ExecuteError(ename, evalue, traceback) =>
- transitToIdle()
-
- (STATUS -> ERROR) ~
- (EXECUTION_COUNT -> executionCount) ~
- (ENAME -> ename) ~
- (EVALUE -> evalue) ~
- (TRACEBACK -> traceback)
-
- case Interpreter.ExecuteAborted(message) =>
- changeState(SessionState.Error())
-
- (STATUS -> ERROR) ~
- (EXECUTION_COUNT -> executionCount) ~
- (ENAME -> "Error") ~
- (EVALUE -> f"Interpreter died:\n$message") ~
- (TRACEBACK -> Seq.empty[String])
- }
- } catch {
- case e: Throwable =>
- error("Exception when executing code", e)
-
- transitToIdle()
-
- (STATUS -> ERROR) ~
- (EXECUTION_COUNT -> executionCount) ~
- (ENAME -> f"Internal Error: ${e.getClass.getName}") ~
- (EVALUE -> e.getMessage) ~
- (TRACEBACK -> Seq.empty[String])
- }
-
- compact(render(resultInJson))
- }
-
- private def setJobGroup(statementId: Int): String = {
- val jobGroup = statementIdToJobGroup(statementId)
- val cmd = Kind(interpreter.kind) match {
- case Spark() =>
- // A dummy value to avoid automatic value binding in scala REPL.
- s"""val _livyJobGroup$jobGroup = sc.setJobGroup("$jobGroup",""" +
- s""""Job group for statement $jobGroup")"""
- case PySpark() | PySpark3() =>
- s"""sc.setJobGroup("$jobGroup", "Job group for statement $jobGroup")"""
- case SparkR() =>
- interpreter.asInstanceOf[SparkRInterpreter].sparkMajorVersion match {
- case "1" =>
- s"""setJobGroup(sc, "$jobGroup", "Job group for statement $jobGroup", """ +
- "FALSE)"
- case "2" =>
- s"""setJobGroup("$jobGroup", "Job group for statement $jobGroup", FALSE)"""
- }
- }
- // Set the job group
- executeCode(statementId, cmd)
- }
-
- private def statementIdToJobGroup(statementId: Int): String = {
- statementId.toString
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-livy/blob/412ccc8f/repl/src/main/scala/com/cloudera/livy/repl/SparkContextInitializer.scala
----------------------------------------------------------------------
diff --git a/repl/src/main/scala/com/cloudera/livy/repl/SparkContextInitializer.scala b/repl/src/main/scala/com/cloudera/livy/repl/SparkContextInitializer.scala
deleted file mode 100644
index ba334da..0000000
--- a/repl/src/main/scala/com/cloudera/livy/repl/SparkContextInitializer.scala
+++ /dev/null
@@ -1,124 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package com.cloudera.livy.repl
-
-import org.apache.spark.{SparkConf, SparkContext}
-
-import com.cloudera.livy.Logging
-
-/**
- * A mixin trait for Spark entry point creation. This trait exists two different code path
- * separately for Spark1 and Spark2, depends on whether SparkSession exists or not.
- */
-trait SparkContextInitializer extends Logging {
- self: SparkInterpreter =>
-
- def createSparkContext(conf: SparkConf): Unit = {
- if (isSparkSessionPresent()) {
- spark2CreateContext(conf)
- } else {
- spark1CreateContext(conf)
- }
- }
-
- private def spark1CreateContext(conf: SparkConf): Unit = {
- sparkContext = SparkContext.getOrCreate(conf)
- var sqlContext: Object = null
-
- if (conf.getBoolean("spark.repl.enableHiveContext", false)) {
- try {
- val loader = Option(Thread.currentThread().getContextClassLoader)
- .getOrElse(getClass.getClassLoader)
- if (loader.getResource("hive-site.xml") == null) {
- warn("livy.repl.enable-hive-context is true but no hive-site.xml found on classpath.")
- }
-
- sqlContext = Class.forName("org.apache.spark.sql.hive.HiveContext")
- .getConstructor(classOf[SparkContext]).newInstance(sparkContext).asInstanceOf[Object]
- info("Created sql context (with Hive support).")
- } catch {
- case _: NoClassDefFoundError =>
- sqlContext = Class.forName("org.apache.spark.sql.SQLContext")
- .getConstructor(classOf[SparkContext]).newInstance(sparkContext).asInstanceOf[Object]
- info("Created sql context.")
- }
- } else {
- sqlContext = Class.forName("org.apache.spark.sql.SQLContext")
- .getConstructor(classOf[SparkContext]).newInstance(sparkContext).asInstanceOf[Object]
- info("Created sql context.")
- }
-
- bind("sc", "org.apache.spark.SparkContext", sparkContext, List("""@transient"""))
- bind("sqlContext", sqlContext.getClass.getCanonicalName, sqlContext, List("""@transient"""))
-
- execute("import org.apache.spark.SparkContext._")
- execute("import sqlContext.implicits._")
- execute("import sqlContext.sql")
- execute("import org.apache.spark.sql.functions._")
- }
-
- private def spark2CreateContext(conf: SparkConf): Unit = {
- val sparkClz = Class.forName("org.apache.spark.sql.SparkSession$")
- val sparkObj = sparkClz.getField("MODULE$").get(null)
-
- val builderMethod = sparkClz.getMethod("builder")
- val builder = builderMethod.invoke(sparkObj)
- builder.getClass.getMethod("config", classOf[SparkConf]).invoke(builder, conf)
-
- var spark: Object = null
- if (conf.get("spark.sql.catalogImplementation", "in-memory").toLowerCase == "hive") {
- if (sparkClz.getMethod("hiveClassesArePresent").invoke(sparkObj).asInstanceOf[Boolean]) {
- val loader = Option(Thread.currentThread().getContextClassLoader)
- .getOrElse(getClass.getClassLoader)
- if (loader.getResource("hive-site.xml") == null) {
- warn("livy.repl.enable-hive-context is true but no hive-site.xml found on classpath.")
- }
-
- builder.getClass.getMethod("enableHiveSupport").invoke(builder)
- spark = builder.getClass.getMethod("getOrCreate").invoke(builder)
- info("Created Spark session (with Hive support).")
- } else {
- spark = builder.getClass.getMethod("getOrCreate").invoke(builder)
- info("Created Spark session.")
- }
- } else {
- spark = builder.getClass.getMethod("getOrCreate").invoke(builder)
- info("Created Spark session.")
- }
-
- sparkContext = spark.getClass.getMethod("sparkContext").invoke(spark)
- .asInstanceOf[SparkContext]
-
- bind("spark", spark.getClass.getCanonicalName, spark, List("""@transient"""))
- bind("sc", "org.apache.spark.SparkContext", sparkContext, List("""@transient"""))
-
- execute("import org.apache.spark.SparkContext._")
- execute("import spark.implicits._")
- execute("import spark.sql")
- execute("import org.apache.spark.sql.functions._")
- }
-
- private def isSparkSessionPresent(): Boolean = {
- try {
- Class.forName("org.apache.spark.sql.SparkSession")
- true
- } catch {
- case _: ClassNotFoundException | _: NoClassDefFoundError => false
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-livy/blob/412ccc8f/repl/src/main/scala/com/cloudera/livy/repl/SparkRInterpreter.scala
----------------------------------------------------------------------
diff --git a/repl/src/main/scala/com/cloudera/livy/repl/SparkRInterpreter.scala b/repl/src/main/scala/com/cloudera/livy/repl/SparkRInterpreter.scala
deleted file mode 100644
index 42f8e17..0000000
--- a/repl/src/main/scala/com/cloudera/livy/repl/SparkRInterpreter.scala
+++ /dev/null
@@ -1,324 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package com.cloudera.livy.repl
-
-import java.io.{File, FileOutputStream}
-import java.lang.ProcessBuilder.Redirect
-import java.nio.file.Files
-import java.util.concurrent.{CountDownLatch, Semaphore, TimeUnit}
-
-import scala.annotation.tailrec
-import scala.collection.JavaConverters._
-import scala.reflect.runtime.universe
-
-import org.apache.commons.codec.binary.Base64
-import org.apache.commons.lang.StringEscapeUtils
-import org.apache.spark.{SparkConf, SparkContext}
-import org.apache.spark.util.{ChildFirstURLClassLoader, MutableURLClassLoader, Utils}
-import org.json4s._
-import org.json4s.JsonDSL._
-
-import com.cloudera.livy.client.common.ClientConf
-import com.cloudera.livy.rsc.RSCConf
-
-private case class RequestResponse(content: String, error: Boolean)
-
-// scalastyle:off println
-object SparkRInterpreter {
- private val LIVY_END_MARKER = "----LIVY_END_OF_COMMAND----"
- private val LIVY_ERROR_MARKER = "----LIVY_END_OF_ERROR----"
- private val PRINT_MARKER = f"""print("$LIVY_END_MARKER")"""
- private val EXPECTED_OUTPUT = f"""[1] "$LIVY_END_MARKER""""
-
- private val PLOT_REGEX = (
- "(" +
- "(?:bagplot)|" +
- "(?:barplot)|" +
- "(?:boxplot)|" +
- "(?:dotchart)|" +
- "(?:hist)|" +
- "(?:lines)|" +
- "(?:pie)|" +
- "(?:pie3D)|" +
- "(?:plot)|" +
- "(?:qqline)|" +
- "(?:qqnorm)|" +
- "(?:scatterplot)|" +
- "(?:scatterplot3d)|" +
- "(?:scatterplot\\.matrix)|" +
- "(?:splom)|" +
- "(?:stripchart)|" +
- "(?:vioplot)" +
- ")"
- ).r.unanchored
-
- def apply(conf: SparkConf): SparkRInterpreter = {
- val backendTimeout = sys.env.getOrElse("SPARKR_BACKEND_TIMEOUT", "120").toInt
- val mirror = universe.runtimeMirror(getClass.getClassLoader)
- val sparkRBackendClass = mirror.classLoader.loadClass("org.apache.spark.api.r.RBackend")
- val backendInstance = sparkRBackendClass.getDeclaredConstructor().newInstance()
-
- var sparkRBackendPort = 0
- val initialized = new Semaphore(0)
- // Launch a SparkR backend server for the R process to connect to
- val backendThread = new Thread("SparkR backend") {
- override def run(): Unit = {
- sparkRBackendPort = sparkRBackendClass.getMethod("init").invoke(backendInstance)
- .asInstanceOf[Int]
-
- initialized.release()
- sparkRBackendClass.getMethod("run").invoke(backendInstance)
- }
- }
-
- backendThread.setDaemon(true)
- backendThread.start()
- try {
- // Wait for RBackend initialization to finish
- initialized.tryAcquire(backendTimeout, TimeUnit.SECONDS)
- val rExec = conf.getOption("spark.r.shell.command")
- .orElse(sys.env.get("SPARKR_DRIVER_R"))
- .getOrElse("R")
-
- var packageDir = ""
- if (sys.env.getOrElse("SPARK_YARN_MODE", "") == "true") {
- packageDir = "./sparkr"
- } else {
- // local mode
- val rLibPath = new File(sys.env.getOrElse("SPARKR_PACKAGE_DIR",
- Seq(sys.env.getOrElse("SPARK_HOME", "."), "R", "lib").mkString(File.separator)))
- if (!ClientConf.TEST_MODE) {
- require(rLibPath.exists(), "Cannot find sparkr package directory.")
- packageDir = rLibPath.getAbsolutePath()
- }
- }
-
- val builder = new ProcessBuilder(Seq(rExec, "--slave @").asJava)
- val env = builder.environment()
- env.put("SPARK_HOME", sys.env.getOrElse("SPARK_HOME", "."))
- env.put("EXISTING_SPARKR_BACKEND_PORT", sparkRBackendPort.toString)
- env.put("SPARKR_PACKAGE_DIR", packageDir)
- env.put("R_PROFILE_USER",
- Seq(packageDir, "SparkR", "profile", "general.R").mkString(File.separator))
-
- builder.redirectErrorStream(true)
- val process = builder.start()
- new SparkRInterpreter(process, backendInstance, backendThread,
- conf.get("spark.livy.spark_major_version", "1"),
- conf.getBoolean("spark.repl.enableHiveContext", false))
- } catch {
- case e: Exception =>
- if (backendThread != null) {
- backendThread.interrupt()
- }
- throw e
- }
- }
-}
-
-class SparkRInterpreter(process: Process,
- backendInstance: Any,
- backendThread: Thread,
- val sparkMajorVersion: String,
- hiveEnabled: Boolean)
- extends ProcessInterpreter(process) {
- import SparkRInterpreter._
-
- implicit val formats = DefaultFormats
-
- private[this] var executionCount = 0
- override def kind: String = "sparkr"
- private[this] val isStarted = new CountDownLatch(1)
-
- final override protected def waitUntilReady(): Unit = {
- // Set the option to catch and ignore errors instead of halting.
- sendRequest("options(error = dump.frames)")
- if (!ClientConf.TEST_MODE) {
- sendRequest("library(SparkR)")
- if (sparkMajorVersion >= "2") {
- if (hiveEnabled) {
- sendRequest("spark <- SparkR::sparkR.session()")
- } else {
- sendRequest("spark <- SparkR::sparkR.session(enableHiveSupport=FALSE)")
- }
- sendRequest(
- """sc <- SparkR:::callJStatic("org.apache.spark.sql.api.r.SQLUtils",
- "getJavaSparkContext", spark)""")
- } else {
- sendRequest("sc <- sparkR.init()")
- if (hiveEnabled) {
- sendRequest("sqlContext <- sparkRHive.init(sc)")
- } else {
- sendRequest("sqlContext <- sparkRSQL.init(sc)")
- }
- }
- }
-
- isStarted.countDown()
- executionCount = 0
- }
-
- override protected def sendExecuteRequest(command: String): Interpreter.ExecuteResponse = {
- isStarted.await()
- var code = command
-
- // Create a image file if this command is trying to plot.
- val tempFile = PLOT_REGEX.findFirstIn(code).map { case _ =>
- val tempFile = Files.createTempFile("", ".png")
- val tempFileString = tempFile.toAbsolutePath
-
- code = f"""png("$tempFileString")\n$code\ndev.off()"""
-
- tempFile
- }
-
- try {
- val response = sendRequest(code)
-
- if (response.error) {
- Interpreter.ExecuteError("Error", response.content)
- } else {
- var content: JObject = TEXT_PLAIN -> response.content
-
- // If we rendered anything, pass along the last image.
- tempFile.foreach { case file =>
- val bytes = Files.readAllBytes(file)
- if (bytes.nonEmpty) {
- val image = Base64.encodeBase64String(bytes)
- content = content ~ (IMAGE_PNG -> image)
- }
- }
-
- Interpreter.ExecuteSuccess(content)
- }
-
- } catch {
- case e: Error =>
- Interpreter.ExecuteError("Error", e.output)
- case e: Exited =>
- Interpreter.ExecuteAborted(e.getMessage)
- } finally {
- tempFile.foreach(Files.delete)
- }
-
- }
-
- private def sendRequest(code: String): RequestResponse = {
- stdin.println(s"""tryCatch(eval(parse(text="${StringEscapeUtils.escapeJava(code)}"))
- |,error = function(e) sprintf("%s%s", e, "${LIVY_ERROR_MARKER}"))
- """.stripMargin)
- stdin.flush()
-
- stdin.println(PRINT_MARKER)
- stdin.flush()
-
- readTo(EXPECTED_OUTPUT, LIVY_ERROR_MARKER)
- }
-
- override protected def sendShutdownRequest() = {
- stdin.println("q()")
- stdin.flush()
-
- while (stdout.readLine() != null) {}
- }
-
- override def close(): Unit = {
- try {
- val closeMethod = backendInstance.getClass().getMethod("close")
- closeMethod.setAccessible(true)
- closeMethod.invoke(backendInstance)
-
- backendThread.interrupt()
- backendThread.join()
- } finally {
- super.close()
- }
- }
-
- @tailrec
- private def readTo(
- marker: String,
- errorMarker: String,
- output: StringBuilder = StringBuilder.newBuilder): RequestResponse = {
- var char = readChar(output)
-
- // Remove any ANSI color codes which match the pattern "\u001b\\[[0-9;]*[mG]".
- // It would be easier to do this with a regex, but unfortunately I don't see an easy way to do
- // without copying the StringBuilder into a string for each character.
- if (char == '\u001b') {
- if (readChar(output) == '[') {
- char = readDigits(output)
-
- if (char == 'm' || char == 'G') {
- output.delete(output.lastIndexOf('\u001b'), output.length)
- }
- }
- }
-
- if (output.endsWith(marker)) {
- var result = stripMarker(output.toString(), marker)
-
- if (result.endsWith(errorMarker + "\"")) {
- result = stripMarker(result, "\\n" + errorMarker)
- RequestResponse(result, error = true)
- } else {
- RequestResponse(result, error = false)
- }
- } else {
- readTo(marker, errorMarker, output)
- }
- }
-
- private def stripMarker(result: String, marker: String): String = {
- result.replace(marker, "")
- .stripPrefix("\n")
- .stripSuffix("\n")
- }
-
- private def readChar(output: StringBuilder): Char = {
- val byte = stdout.read()
- if (byte == -1) {
- throw new Exited(output.toString())
- } else {
- val char = byte.toChar
- output.append(char)
- char
- }
- }
-
- @tailrec
- private def readDigits(output: StringBuilder): Char = {
- val byte = stdout.read()
- if (byte == -1) {
- throw new Exited(output.toString())
- }
-
- val char = byte.toChar
-
- if (('0' to '9').contains(char)) {
- output.append(char)
- readDigits(output)
- } else {
- char
- }
- }
-
- private class Exited(val output: String) extends Exception {}
- private class Error(val output: String) extends Exception {}
-}
-// scalastyle:on println