Bladeren bron

[livy] Cache the session results in livy-server

Erick Tryzelaar 10 jaren geleden
bovenliggende
commit
ece0313

+ 42 - 18
apps/spark/java/livy-server/src/main/scala/com/cloudera/hue/livy/server/sessions/SparkWebSession.scala

@@ -6,6 +6,7 @@ import org.json4s.jackson.Serialization.write
 import org.json4s.{DefaultFormats, Formats}
 
 import scala.annotation.tailrec
+import scala.collection.mutable.ArrayBuffer
 import scala.concurrent.{Future, _}
 
 abstract class SparkWebSession(val id: String, hostname: String, port: Int) extends Session with Logging {
@@ -17,54 +18,77 @@ abstract class SparkWebSession(val id: String, hostname: String, port: Int) exte
   private[this] var _state: State = Idle()
   private[this] val svc = host(hostname, port)
 
+  private[this] var executedStatements = 0
+  private[this] var statements_ = new ArrayBuffer[ExecuteResponse]
+
   override def lastActivity: Long = _lastActivity
 
   override def state: State = _state
 
-  override def executeStatement(statement: String): Future[ExecuteResponse] = {
+  override def executeStatement(statement: String): (Int, Future[ExecuteResponse]) = {
     ensureIdle {
       _state = Busy()
       touchLastActivity()
 
+      executedStatements += 1
+
       var req = (svc / "statements").setContentType("application/json", "UTF-8")
       req = req << write(ExecuteRequest(statement))
 
-      Http(req OK as.json4s.Json).map {
-        transition(Idle())
-        _.extract[ExecuteResponse]
+      val future = Http(req OK as.json4s.Json).map { case (resp) =>
+        synchronized {
+          transition(Idle())
+          val response: ExecuteResponse = resp.extract[ExecuteResponse]
+          statements_ += response
+          response
+        }
       }
+
+      (executedStatements - 1, future)
     }
   }
 
   override def statement(statementId: Int): Future[ExecuteResponse] = {
     ensureRunning {
-      val req = svc / "statements" / statementId
+      if (statementId < statements_.length) {
+        Future.successful(statements_(statementId))
+      } else {
+        val req = svc / "statements" / statementId
 
-      for {
-        body <- Http(req OK as.json4s.Json)
-      } yield body.extract[ExecuteResponse]
+        for {
+          body <- Http(req OK as.json4s.Json)
+        } yield body.extract[ExecuteResponse]
+      }
     }
   }
 
   override def statements(): Future[List[ExecuteResponse]] = {
     ensureRunning {
-      val req = svc / "statements"
+      if (_state == Idle()) {
+        Future.successful(statements_.toList)
+      } else {
+        val req = svc / "statements"
 
-      for {
-        body <- Http(req OK as.json4s.Json)
-      } yield body.extract[List[ExecuteResponse]]
+        for {
+          body <- Http(req OK as.json4s.Json)
+        } yield body.extract[List[ExecuteResponse]]
+      }
     }
   }
 
   override def statements(fromIndex: Integer, toIndex: Integer): Future[List[ExecuteResponse]] = {
     ensureRunning {
-      val req = (svc / "statements")
-        .addQueryParameter("from", fromIndex.toString)
-        .addQueryParameter("to", toIndex.toString)
+      if (_state == Idle()) {
+        Future.successful(statements_.slice(fromIndex, toIndex).toList)
+      } else {
+        val req = (svc / "statements")
+          .addQueryParameter("from", fromIndex.toString)
+          .addQueryParameter("to", toIndex.toString)
 
-      for {
-        body <- Http(req OK as.json4s.Json)
-      } yield body.extract[List[ExecuteResponse]]
+        for {
+          body <- Http(req OK as.json4s.Json)
+        } yield body.extract[List[ExecuteResponse]]
+      }
     }
   }
   override def interrupt(): Future[Unit] = {