فهرست منبع

[livy] Migrate fake_shell.py into a resource

Erick Tryzelaar 10 سال پیش
والد
کامیت
4cf28a0

+ 0 - 0
apps/spark/java/livy-repl/src/main/python/fake_shell.py → apps/spark/java/livy-repl/src/main/resources/fake_shell.py


+ 1 - 1
apps/spark/java/livy-repl/src/main/scala/com/cloudera/hue/livy/repl/Main.scala

@@ -34,8 +34,8 @@ object Main extends Logging {
     val server = new WebServer(port)
 
     server.context.setResourceBase("src/main/com/cloudera/hue/livy/repl")
-    server.context.setInitParameter(ScalatraListener.LifeCycleKey, classOf[ScalatraBootstrap].getCanonicalName)
     server.context.addEventListener(new ScalatraListener)
+    server.context.setInitParameter(ScalatraListener.LifeCycleKey, classOf[ScalatraBootstrap].getCanonicalName)
     server.context.setInitParameter(SESSION_KIND, session_kind)
 
     server.start()

+ 24 - 5
apps/spark/java/livy-repl/src/main/scala/com/cloudera/hue/livy/repl/python/PythonSession.scala

@@ -1,11 +1,11 @@
 package com.cloudera.hue.livy.repl.python
 
 import java.io._
+import java.nio.file.Files
 
 import com.cloudera.hue.livy.ExecuteResponse
 import com.cloudera.hue.livy.repl.Session
 import org.json4s.DefaultFormats
-import org.json4s.JsonAST._
 import org.json4s.JsonDSL._
 import org.json4s.jackson.JsonMethods._
 
@@ -13,11 +13,9 @@ import scala.collection.mutable.ArrayBuffer
 import scala.concurrent.{ExecutionContext, Future}
 
 object PythonSession {
-  val LIVY_HOME = System.getenv("LIVY_HOME")
-  val FAKE_SHELL = LIVY_HOME + "/livy-repl/src/main/python/fake_shell.py"
-
   def create(): Session = {
-    val pb = new ProcessBuilder("python", FAKE_SHELL)
+    val file = createScript()
+    val pb = new ProcessBuilder("python", file.toString)
     val process = pb.start()
     val in = process.getInputStream
     val out = process.getOutputStream
@@ -25,6 +23,27 @@ object PythonSession {
     new PythonSession(process, in, out)
   }
 
+  private def createScript(): 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
+  }
+
   // Java unfortunately wraps the input stream in a buffer, so we need to hack around it so we can read the output
   // without blocking.
   private def unwrapInputStream(inputStream: InputStream) = {

+ 0 - 8
apps/spark/java/livy-yarn/src/main/assembly/dist.xml

@@ -24,14 +24,6 @@
             <outputDirectory>lib</outputDirectory>
             <useProjectArtifact>true</useProjectArtifact>
 
-            <!--
-            <includes>
-                <include>com.cloudera.hue.livy:livy-yarn</include>
-                <include>org.slf4j:slf4j-log4j12</include>
-                <include>org.apache.hadoop:hadoop-hdfs</include>
-            </includes>
-            -->
-
             <useTransitiveFiltering>true</useTransitiveFiltering>
         </dependencySet>
     </dependencySets>

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

@@ -7,11 +7,16 @@ import org.apache.hadoop.yarn.api.records.FinalApplicationStatus
 import org.apache.hadoop.yarn.client.api.AMRMClient
 import org.apache.hadoop.yarn.conf.YarnConfiguration
 import org.apache.hadoop.yarn.util.ConverterUtils
+import org.apache.spark.repl.Main
 import org.scalatra.servlet.ScalatraListener
 
 object AppMaster extends Logging {
 
+  val SESSION_KIND = "livy-repl.session.kind"
+
   def main(args: Array[String]): Unit = {
+    val lang = args(1)
+
     val containerIdString = System.getenv(ApplicationConstants.Environment.CONTAINER_ID.toString)
     info("got container id: %s" format containerIdString)
     val containerId = ConverterUtils.toContainerId(containerIdString)
@@ -26,20 +31,21 @@ object AppMaster extends Logging {
     info("got node manager port: %s" format nodePortString)
 
     val yarnConfig = new YarnConfiguration()
-    val service = new AppMasterService(yarnConfig, nodeHostString)
+    val service = new AppMasterService(yarnConfig, nodeHostString, lang)
     service.run()
   }
 
 }
 
-class AppMasterService(yarnConfig: YarnConfiguration, nodeHostString: String) extends Logging {
+class AppMasterService(yarnConfig: YarnConfiguration, nodeHostString: String, lang: String) extends Logging {
   val webServer = new WebServer(0)
   val amRMClient = AMRMClient.createAMRMClient()
   amRMClient.init(yarnConfig)
 
   webServer.context.setResourceBase("src/main/com/cloudera/hue/livy/repl")
-  webServer.context.setInitParameter(ScalatraListener.LifeCycleKey, classOf[ScalatraBootstrap].getCanonicalName)
   webServer.context.addEventListener(new ScalatraListener)
+  webServer.context.setInitParameter(ScalatraListener.LifeCycleKey, classOf[ScalatraBootstrap].getCanonicalName)
+  webServer.context.setInitParameter(AppMaster.SESSION_KIND, lang)
 
   def run(): Unit = {
     webServer.start()

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

@@ -14,6 +14,7 @@ object Client extends Logging {
 
   def main(args: Array[String]): Unit = {
     val packagePath = new Path(args(1))
+    val lang = args(2)
 
     val yarnConf = new YarnConfiguration()
     yarnConf.set("yarn.resourcemanager.am.max-attempts", "1")
@@ -24,7 +25,8 @@ object Client extends Logging {
       val job = client.submitApplication(
         packagePath,
         List(
-          "__package/bin/run-am.sh 1>%s/stdout 2>%s/stderr" format (
+          "__package/bin/run-am.sh %s 1>%s/stdout 2>%s/stderr" format (
+            lang,
             ApplicationConstants.LOG_DIR_EXPANSION_VAR,
             ApplicationConstants.LOG_DIR_EXPANSION_VAR
           )