瀏覽代碼

[livy] Thread sessions aren't being used and weren't worthwhile

Erick Tryzelaar 10 年之前
父節點
當前提交
16fa5704ac

+ 1 - 1
apps/spark/java/conf/livy-defaults.conf.tmpl

@@ -5,7 +5,7 @@
 # What port to start the server on. Defaults to 8998.
 # livy.server.port = 8998
 
-# What session factory to use. The options are `thread`, `process`, and `yarn`.
+# What session factory to use. The options are `process` and `yarn`.
 # livy.server.session.factory = process
 
 # What spark-submit executable path to use to submit spark applications. Defaults to

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

@@ -27,7 +27,6 @@ object LivyConf {
   val SPARK_SUBMIT_KEY = "livy.server.spark-submit"
 
   sealed trait SessionKind
-  case class Thread() extends SessionKind
   case class Process() extends SessionKind
   case class Yarn() extends SessionKind
 }
@@ -95,14 +94,13 @@ class LivyConf(loadDefaults: Boolean) {
 
   def sessionKind(): SessionKind = getOption(SESSION_FACTORY_KEY).getOrElse("process") match {
     case "process" => Process()
-    case "thread" => Thread()
     case "yarn" => Yarn()
     case kind => throw new IllegalStateException(f"unknown kind $kind")
   }
 
   /** Return the filesystem root. Defaults to the local filesystem. */
   def filesystemRoot(): String = sessionKind() match {
-    case Process() | Thread() => "file://"
+    case Process() => "file://"
     case Yarn() => "hdfs://"
   }
 }

+ 0 - 3
apps/spark/java/livy-server/src/main/scala/com/cloudera/hue/livy/server/Main.scala

@@ -32,7 +32,6 @@ import org.slf4j.LoggerFactory
 object Main {
 
   val SESSION_KIND = "livy-server.session.kind"
-  val THREAD_SESSION = "thread"
   val PROCESS_SESSION = "process"
   val YARN_SESSION = "yarn"
   lazy val logger = LoggerFactory.getLogger(this.getClass)
@@ -136,8 +135,6 @@ class ScalatraBootstrap extends LifeCycle with Logging {
     info(f"Using $sessionFactoryKind sessions")
 
     val (sessionFactory, batchFactory) = sessionFactoryKind match {
-      case LivyConf.Thread() =>
-        (new InteractiveSessionProcessFactory(livyConf), new BatchSessionProcessFactory(livyConf) )
       case LivyConf.Process() =>
         (new InteractiveSessionProcessFactory(livyConf), new BatchSessionProcessFactory(livyConf))
       case LivyConf.Yarn() =>

+ 0 - 85
apps/spark/java/livy-server/src/main/scala/com/cloudera/hue/livy/server/interactive/InteractiveSessionThread.scala

@@ -1,85 +0,0 @@
-/*
- * Licensed to Cloudera, Inc. under one
- * or more contributor license agreements.  See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership.  Cloudera, Inc. 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.hue.livy.server.interactive
-
-import java.net.URL
-
-import com.cloudera.hue.livy.msgs.ExecuteRequest
-import com.cloudera.hue.livy.repl.python.PythonSession
-import com.cloudera.hue.livy.repl.scala.SparkSession
-import com.cloudera.hue.livy.sessions.{PySpark, Spark, State}
-
-import scala.collection.mutable.ArrayBuffer
-import scala.concurrent.{ExecutionContext, ExecutionContextExecutor, Future}
-
-object InteractiveSessionThread {
-  val LIVY_HOME = System.getenv("LIVY_HOME")
-  val LIVY_REPL = LIVY_HOME + "/bin/livy-repl"
-
-  def create(id: Int, createInteractiveRequest: CreateInteractiveRequest): InteractiveSession = {
-    val session = createInteractiveRequest.kind match {
-      case Spark() =>
-        SparkSession.create()
-      case PySpark() =>
-        PythonSession.createPySpark()
-    }
-    new InteractiveSessionThread(id, createInteractiveRequest, session)
-  }
-}
-
-private class InteractiveSessionThread(val id: Int,
-                                       createInteractiveRequest: CreateInteractiveRequest,
-                                       session: com.cloudera.hue.livy.repl.Session) extends InteractiveSession {
-
-  protected implicit def executor: ExecutionContextExecutor = ExecutionContext.global
-
-  private var executedStatements = 0
-  private var statements_ = new ArrayBuffer[Statement]
-
-  override def kind = createInteractiveRequest.kind
-
-  override def proxyUser: Option[String] = None
-
-  override def lastActivity: Long = 0
-
-  override def state: State = session.state
-
-  override def url: Option[URL] = None
-
-  override def url_=(url: URL): Unit = {}
-
-  override def executeStatement(content: ExecuteRequest): Statement = {
-    val statement = new Statement(executedStatements, content, session.execute(content.code))
-
-    executedStatements += 1
-    statements_ += statement
-
-    statement
-  }
-
-  override def statements: IndexedSeq[Statement] = statements_
-
-  override def interrupt(): Future[Unit] = {
-    stop()
-  }
-
-  override def stop(): Future[Unit] = Future {
-    session.close()
-  }
-}

+ 0 - 34
apps/spark/java/livy-server/src/main/scala/com/cloudera/hue/livy/server/interactive/InteractiveSessionThreadFactory.scala

@@ -1,34 +0,0 @@
-/*
- * Licensed to Cloudera, Inc. under one
- * or more contributor license agreements.  See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership.  Cloudera, Inc. 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.hue.livy.server.interactive
-
-import com.cloudera.hue.livy.LivyConf
-
-import scala.concurrent.{ExecutionContext, Future}
-
-class InteractiveSessionThreadFactory(livyConf: LivyConf) extends InteractiveSessionFactory {
-
-   implicit def executor: ExecutionContext = ExecutionContext.global
-
-   override def createSession(id: Int, createInteractiveRequest: CreateInteractiveRequest): Future[InteractiveSession] = {
-     Future {
-       InteractiveSessionThread.create(id, createInteractiveRequest)
-     }
-   }
- }

+ 0 - 27
apps/spark/java/livy-server/src/test/scala/com/cloudera/hue/livy/server/interactive/InteractiveSessionThreadSpec.scala

@@ -1,27 +0,0 @@
-/*
- * Licensed to Cloudera, Inc. under one
- * or more contributor license agreements.  See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership.  Cloudera, Inc. 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.hue.livy.server.interactive
-
-import com.cloudera.hue.livy.sessions.Spark
-import org.scalatest.{BeforeAndAfter, FunSpecLike, Matchers}
-
-class InteractiveSessionThreadSpec extends BaseSessionSpec with FunSpecLike with Matchers with BeforeAndAfter {
-
-  def createSession() = InteractiveSessionThread.create(0, CreateInteractiveRequest(kind = Spark()))
-}