Bläddra i källkod

[livy] Rename Batch* to BatchSession*

Erick Tryzelaar 10 år sedan
förälder
incheckning
e1a6db3

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

@@ -20,7 +20,7 @@ package com.cloudera.hue.livy.server
 
 import javax.servlet.ServletContext
 
-import com.cloudera.hue.livy.server.batch.{BatchServlet, BatchManager, BatchYarnFactory, BatchProcessFactory}
+import com.cloudera.hue.livy.server.batch.{BatchSessionProcessFactory, BatchSessionServlet, BatchManager, BatchSessionYarnFactory}
 import com.cloudera.hue.livy.server.interactive._
 import com.cloudera.hue.livy.{Utils, Logging, LivyConf, WebServer}
 import org.scalatra._
@@ -80,18 +80,18 @@ class ScalatraBootstrap extends LifeCycle with Logging {
 
     val (sessionFactory, batchFactory) = sessionFactoryKind match {
       case LivyConf.Thread() =>
-        (new InteractiveSessionThreadFactory(livyConf), new BatchProcessFactory(livyConf) )
+        (new InteractiveSessionProcessFactory(livyConf), new BatchSessionProcessFactory(livyConf) )
       case LivyConf.Process() =>
-        (new InteractiveSessionProcessFactory(livyConf), new BatchProcessFactory(livyConf))
+        (new InteractiveSessionProcessFactory(livyConf), new BatchSessionProcessFactory(livyConf))
       case LivyConf.Yarn() =>
-        (new InteractiveSessionYarnFactory(livyConf), new BatchYarnFactory(livyConf))
+        (new InteractiveSessionYarnFactory(livyConf), new BatchSessionYarnFactory(livyConf))
     }
 
     sessionManager = new SessionManager(sessionFactory)
     batchManager = new BatchManager(batchFactory)
 
     context.mount(new InteractiveSessionServlet(sessionManager), "/sessions/*")
-    context.mount(new BatchServlet(batchManager), "/batches/*")
+    context.mount(new BatchSessionServlet(batchManager), "/batches/*")
   }
 
   override def destroy(context: ServletContext): Unit = {

+ 1 - 1
apps/spark/java/livy-server/src/main/scala/com/cloudera/hue/livy/server/batch/Batch.scala → apps/spark/java/livy-server/src/main/scala/com/cloudera/hue/livy/server/batch/BatchSession.scala

@@ -22,7 +22,7 @@ import com.cloudera.hue.livy.server.Session
 
 import scala.concurrent.Future
 
-trait Batch extends Session {
+trait BatchSession extends Session {
   def lines: IndexedSeq[String]
 
   def stop(): Future[Unit]

+ 2 - 2
apps/spark/java/livy-server/src/main/scala/com/cloudera/hue/livy/server/batch/BatchFactory.scala → apps/spark/java/livy-server/src/main/scala/com/cloudera/hue/livy/server/batch/BatchSessionFactory.scala

@@ -18,6 +18,6 @@
 
 package com.cloudera.hue.livy.server.batch
 
-abstract class BatchFactory {
-  def create(id: Int, createBatchRequest: CreateBatchRequest): Batch
+abstract class BatchSessionFactory {
+  def create(id: Int, createBatchRequest: CreateBatchRequest): BatchSession
 }

+ 7 - 7
apps/spark/java/livy-server/src/main/scala/com/cloudera/hue/livy/server/batch/BatchManager.scala → apps/spark/java/livy-server/src/main/scala/com/cloudera/hue/livy/server/batch/BatchSessionManager.scala

@@ -24,15 +24,15 @@ import java.util.concurrent.atomic.AtomicInteger
 import scala.collection.JavaConversions._
 import scala.concurrent.Future
 
-class BatchManager(batchFactory: BatchFactory) {
+class BatchManager(batchFactory: BatchSessionFactory) {
   private[this] val _idCounter = new AtomicInteger()
-  private[this] val _batches = new ConcurrentHashMap[Int, Batch]
+  private[this] val _batches = new ConcurrentHashMap[Int, BatchSession]
 
-  def getBatch(id: Int): Option[Batch] = Option(_batches.get(id))
+  def getBatch(id: Int): Option[BatchSession] = Option(_batches.get(id))
 
-  def getBatches: Array[Batch] = _batches.values().iterator().toArray
+  def getBatches: Array[BatchSession] = _batches.values().iterator().toArray
 
-  def createBatch(createBatchRequest: CreateBatchRequest): Batch = {
+  def createBatch(createBatchRequest: CreateBatchRequest): BatchSession = {
     val id = _idCounter.getAndIncrement
     val batch = batchFactory.create(id, createBatchRequest)
     _batches.put(id, batch)
@@ -40,11 +40,11 @@ class BatchManager(batchFactory: BatchFactory) {
     batch
   }
 
-  def remove(id: Int): Option[Batch] = {
+  def remove(id: Int): Option[BatchSession] = {
     Option(_batches.remove(id))
   }
 
-  def delete(batch: Batch): Future[Unit] = {
+  def delete(batch: BatchSession): Future[Unit] = {
     _batches.remove(batch.id)
     batch.stop()
   }

+ 5 - 5
apps/spark/java/livy-server/src/main/scala/com/cloudera/hue/livy/server/batch/BatchProcess.scala → apps/spark/java/livy-server/src/main/scala/com/cloudera/hue/livy/server/batch/BatchSessionProcess.scala

@@ -27,12 +27,12 @@ import com.cloudera.hue.livy.spark.SparkSubmitProcessBuilder
 
 import scala.concurrent.{Future, ExecutionContext, ExecutionContextExecutor}
 
-object BatchProcess {
-  def apply(livyConf: LivyConf, id: Int, createBatchRequest: CreateBatchRequest): Batch = {
+object BatchSessionProcess {
+  def apply(livyConf: LivyConf, id: Int, createBatchRequest: CreateBatchRequest): BatchSession = {
     val builder = sparkBuilder(livyConf, createBatchRequest)
 
     val process = builder.start(RelativePath(createBatchRequest.file), createBatchRequest.args)
-    new BatchProcess(id, new LineBufferedProcess(process))
+    new BatchSessionProcess(id, new LineBufferedProcess(process))
   }
 
   private def sparkBuilder(livyConf: LivyConf, createBatchRequest: CreateBatchRequest): SparkSubmitProcessBuilder = {
@@ -54,8 +54,8 @@ object BatchProcess {
   }
 }
 
-private class BatchProcess(val id: Int,
-                           process: LineBufferedProcess) extends Batch {
+private class BatchSessionProcess(val id: Int,
+                           process: LineBufferedProcess) extends BatchSession {
   protected implicit def executor: ExecutionContextExecutor = ExecutionContext.global
 
   private[this] var isAlive = true

+ 3 - 3
apps/spark/java/livy-server/src/main/scala/com/cloudera/hue/livy/server/batch/BatchProcessFactory.scala → apps/spark/java/livy-server/src/main/scala/com/cloudera/hue/livy/server/batch/BatchSessionProcessFactory.scala

@@ -20,7 +20,7 @@ package com.cloudera.hue.livy.server.batch
 
 import com.cloudera.hue.livy.LivyConf
 
-class BatchProcessFactory(livyConf: LivyConf) extends BatchFactory {
-  def create(id: Int, createBatchRequest: CreateBatchRequest): Batch =
-    BatchProcess(livyConf, id, createBatchRequest)
+class BatchSessionProcessFactory(livyConf: LivyConf) extends BatchSessionFactory {
+  def create(id: Int, createBatchRequest: CreateBatchRequest): BatchSession =
+    BatchSessionProcess(livyConf, id, createBatchRequest)
 }

+ 7 - 7
apps/spark/java/livy-server/src/main/scala/com/cloudera/hue/livy/server/batch/BatchServlet.scala → apps/spark/java/livy-server/src/main/scala/com/cloudera/hue/livy/server/batch/BatchSessionServlet.scala

@@ -27,9 +27,9 @@ import org.scalatra.json.JacksonJsonSupport
 
 import scala.concurrent.{Future, ExecutionContext, ExecutionContextExecutor}
 
-object BatchServlet extends Logging
+object BatchSessionServlet extends Logging
 
-class BatchServlet(batchManager: BatchManager)
+class BatchSessionServlet(batchManager: BatchManager)
   extends ScalatraServlet
   with FutureSupport
   with MethodOverride
@@ -119,7 +119,7 @@ class BatchServlet(batchManager: BatchManager)
     case e: JsonParseException => BadRequest(e.getMessage)
     case e: MappingException => BadRequest(e.getMessage)
     case e =>
-      BatchServlet.error("internal error", e)
+      BatchSessionServlet.error("internal error", e)
       InternalServerError(e.toString)
   }
 }
@@ -129,7 +129,7 @@ private object Serializers {
 
   def Formats: List[CustomSerializer[_]] = List(BatchSerializer)
 
-  def serializeBatch(batch: Batch,
+  def serializeBatch(batch: BatchSession,
                      from: Option[Int],
                      size: Option[Int]): JValue = {
 
@@ -138,7 +138,7 @@ private object Serializers {
       ("log", getLogs(batch, from, size))
   }
 
-  def getLogs(batch: Batch, fromOpt: Option[Int], sizeOpt: Option[Int]): JValue = {
+  def getLogs(batch: BatchSession, fromOpt: Option[Int], sizeOpt: Option[Int]): JValue = {
     val lines = batch.lines
 
     val size = sizeOpt.getOrElse(100)
@@ -151,12 +151,12 @@ private object Serializers {
     lines.view(from, until)
   }
 
-  case object BatchSerializer extends CustomSerializer[Batch](
+  case object BatchSerializer extends CustomSerializer[BatchSession](
     implicit formats => ( {
     // We don't support deserialization.
     PartialFunction.empty
   }, {
-    case batch: Batch =>
+    case batch: BatchSession =>
       serializeBatch(batch, None, None)
   }
     )

+ 4 - 4
apps/spark/java/livy-server/src/main/scala/com/cloudera/hue/livy/server/batch/BatchYarn.scala → apps/spark/java/livy-server/src/main/scala/com/cloudera/hue/livy/server/batch/BatchSessionYarn.scala

@@ -31,18 +31,18 @@ import scala.annotation.tailrec
 import scala.concurrent.{ExecutionContextExecutor, ExecutionContext, Future}
 import scala.util
 
-object BatchYarn {
+object BatchSessionYarn {
 
   implicit def executor: ExecutionContextExecutor = ExecutionContext.global
 
-  def apply(livyConf: LivyConf, client: Client, id: Int, createBatchRequest: CreateBatchRequest): Batch = {
+  def apply(livyConf: LivyConf, client: Client, id: Int, createBatchRequest: CreateBatchRequest): BatchSession = {
     val builder = sparkBuilder(livyConf, createBatchRequest)
 
     val process = new LineBufferedProcess(builder.start(RelativePath(createBatchRequest.file), createBatchRequest.args))
     val job = Future {
       client.getJobFromProcess(process)
     }
-    new BatchYarn(id, process, job)
+    new BatchSessionYarn(id, process, job)
   }
 
   private def sparkBuilder(livyConf: LivyConf, createBatchRequest: CreateBatchRequest): SparkSubmitProcessBuilder = {
@@ -67,7 +67,7 @@ object BatchYarn {
   }
 }
 
-private class BatchYarn(val id: Int, process: LineBufferedProcess, jobFuture: Future[Job]) extends Batch {
+private class BatchSessionYarn(val id: Int, process: LineBufferedProcess, jobFuture: Future[Job]) extends BatchSession {
 
   implicit def executor: ExecutionContextExecutor = ExecutionContext.global
 

+ 3 - 3
apps/spark/java/livy-server/src/main/scala/com/cloudera/hue/livy/server/batch/BatchYarnFactory.scala → apps/spark/java/livy-server/src/main/scala/com/cloudera/hue/livy/server/batch/BatchSessionYarnFactory.scala

@@ -21,10 +21,10 @@ package com.cloudera.hue.livy.server.batch
 import com.cloudera.hue.livy.LivyConf
 import com.cloudera.hue.livy.yarn.Client
 
-class BatchYarnFactory(livyConf: LivyConf) extends BatchFactory {
+class BatchSessionYarnFactory(livyConf: LivyConf) extends BatchSessionFactory {
 
   val client = new Client(livyConf)
 
-  def create(id: Int, createBatchRequest: CreateBatchRequest): Batch =
-    BatchYarn(livyConf, client, id, createBatchRequest)
+  def create(id: Int, createBatchRequest: CreateBatchRequest): BatchSession =
+    BatchSessionYarn(livyConf, client, id, createBatchRequest)
 }

+ 2 - 2
apps/spark/java/livy-server/src/test/scala/com/cloudera/hue/livy/server/batches/BatchProcessSpec.scala

@@ -24,7 +24,7 @@ import java.util.concurrent.TimeUnit
 
 import com.cloudera.hue.livy.sessions.Success
 import com.cloudera.hue.livy.{LivyConf, Utils}
-import com.cloudera.hue.livy.server.batch.{CreateBatchRequest, BatchProcess}
+import com.cloudera.hue.livy.server.batch.{CreateBatchRequest, BatchSessionProcess}
 import org.scalatest.{ShouldMatchers, BeforeAndAfterAll, FunSpec}
 
 import scala.concurrent.duration.Duration
@@ -54,7 +54,7 @@ class BatchProcessSpec
       val req = CreateBatchRequest(
         file = script.toString
       )
-      val batch = BatchProcess(new LivyConf(), 0, req)
+      val batch = BatchSessionProcess(new LivyConf(), 0, req)
 
       Utils.waitUntil({ () =>
         batch.state == Success()

+ 3 - 3
apps/spark/java/livy-server/src/test/scala/com/cloudera/hue/livy/server/batches/BatchServletSpec.scala

@@ -53,9 +53,9 @@ class BatchServletSpec extends ScalatraSuite with FunSpecLike with BeforeAndAfte
     script
   }
 
-  val batchFactory = new BatchProcessFactory(new LivyConf())
+  val batchFactory = new BatchSessionProcessFactory(new LivyConf())
   val batchManager = new BatchManager(batchFactory)
-  val servlet = new BatchServlet(batchManager)
+  val servlet = new BatchSessionServlet(batchManager)
 
   addServlet(servlet, "/*")
 
@@ -89,7 +89,7 @@ class BatchServletSpec extends ScalatraSuite with FunSpecLike with BeforeAndAfte
 
       // Wait for the process to finish.
       {
-        val batch: Batch = batchManager.getBatch(0).get
+        val batch: BatchSession = batchManager.getBatch(0).get
         Utils.waitUntil({ () =>
           batch.state == Success()
         }, Duration(10, TimeUnit.SECONDS))