You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@toree.apache.org by lr...@apache.org on 2016/01/11 22:01:46 UTC
[04/50] [abbrv] incubator-toree git commit: InterpreterManager for
Interpreter initialization and management
InterpreterManager for Interpreter initialization and management
Project: http://git-wip-us.apache.org/repos/asf/incubator-toree/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-toree/commit/502372d4
Tree: http://git-wip-us.apache.org/repos/asf/incubator-toree/tree/502372d4
Diff: http://git-wip-us.apache.org/repos/asf/incubator-toree/diff/502372d4
Branch: refs/heads/master
Commit: 502372d47ddd940969149f96936746ba8f6f253a
Parents: d2172c1
Author: Brian Burns <bb...@us.ibm.com>
Authored: Mon Nov 9 15:51:37 2015 -0500
Committer: Brian Burns <bb...@us.ibm.com>
Committed: Mon Nov 9 15:51:37 2015 -0500
----------------------------------------------------------------------
.../com/ibm/spark/interpreter/Interpreter.scala | 9 +++
.../com/ibm/spark/kernel/api/KernelLike.scala | 3 +
.../com/ibm/spark/boot/CommandLineOptions.scala | 17 +++-
.../boot/layer/ComponentInitialization.scala | 19 +++--
.../spark/boot/layer/InterpreterManager.scala | 59 ++++++++++++++
.../scala/com/ibm/spark/kernel/api/Kernel.scala | 29 +++++--
.../StandardComponentInitializationSpec.scala | 84 --------------------
.../com/ibm/spark/kernel/api/KernelSpec.scala | 10 ++-
.../scala/test/utils/DummyInterpreter.scala | 7 ++
.../scala/test/utils/SparkKernelDeployer.scala | 1 +
.../pyspark/PySparkInterpreter.scala | 22 ++++-
.../com/ibm/spark/magic/builtin/PySpark.scala | 2 +-
resources/compile/reference.conf | 5 +-
resources/test/reference.conf | 2 +-
.../interpreter/scala/ScalaInterpreter.scala | 13 ++-
.../com/ibm/spark/magic/builtin/Scala.scala | 2 +-
.../interpreter/sparkr/SparkRInterpreter.scala | 12 ++-
.../com/ibm/spark/magic/builtin/SparkR.scala | 2 +-
.../kernel/interpreter/sql/SqlInterpreter.scala | 13 ++-
.../scala/com/ibm/spark/magic/builtin/Sql.scala | 2 +-
20 files changed, 192 insertions(+), 121 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-toree/blob/502372d4/kernel-api/src/main/scala/com/ibm/spark/interpreter/Interpreter.scala
----------------------------------------------------------------------
diff --git a/kernel-api/src/main/scala/com/ibm/spark/interpreter/Interpreter.scala b/kernel-api/src/main/scala/com/ibm/spark/interpreter/Interpreter.scala
index 24fe9dc..76d4432 100644
--- a/kernel-api/src/main/scala/com/ibm/spark/interpreter/Interpreter.scala
+++ b/kernel-api/src/main/scala/com/ibm/spark/interpreter/Interpreter.scala
@@ -18,11 +18,20 @@ package com.ibm.spark.interpreter
import java.net.URL
+import com.ibm.spark.kernel.api.KernelLike
import org.apache.spark.SparkContext
import scala.tools.nsc.interpreter._
trait Interpreter {
+
+ /**
+ * Initializes the interpreter.
+ * @param kernel The kernel
+ * @return The newly initialized interpreter
+ */
+ def init(kernel: KernelLike): Interpreter
+
/**
* Starts the interpreter, initializing any internal state.
* @return A reference to the interpreter
http://git-wip-us.apache.org/repos/asf/incubator-toree/blob/502372d4/kernel-api/src/main/scala/com/ibm/spark/kernel/api/KernelLike.scala
----------------------------------------------------------------------
diff --git a/kernel-api/src/main/scala/com/ibm/spark/kernel/api/KernelLike.scala b/kernel-api/src/main/scala/com/ibm/spark/kernel/api/KernelLike.scala
index 4098bfb..8fb5d80 100644
--- a/kernel-api/src/main/scala/com/ibm/spark/kernel/api/KernelLike.scala
+++ b/kernel-api/src/main/scala/com/ibm/spark/kernel/api/KernelLike.scala
@@ -90,6 +90,9 @@ trait KernelLike {
*/
val data: java.util.Map[String, Any]
+
+ def interpreter(name: String): Option[com.ibm.spark.interpreter.Interpreter]
+
def sparkContext: SparkContext
def sparkConf: SparkConf
http://git-wip-us.apache.org/repos/asf/incubator-toree/blob/502372d4/kernel/src/main/scala/com/ibm/spark/boot/CommandLineOptions.scala
----------------------------------------------------------------------
diff --git a/kernel/src/main/scala/com/ibm/spark/boot/CommandLineOptions.scala b/kernel/src/main/scala/com/ibm/spark/boot/CommandLineOptions.scala
index 27c5c68..9539404 100644
--- a/kernel/src/main/scala/com/ibm/spark/boot/CommandLineOptions.scala
+++ b/kernel/src/main/scala/com/ibm/spark/boot/CommandLineOptions.scala
@@ -96,7 +96,7 @@ class CommandLineOptions(args: Seq[String]) {
private val _nosparkcontext =
parser.accepts("nosparkcontext", "kernel should not create a spark context")
- private val _plugins = parser.accepts(
+ private val _interpreter_plugin = parser.accepts(
"interpreter-plugin"
).withRequiredArg().ofType(classOf[String])
@@ -178,8 +178,19 @@ class CommandLineOptions(args: Seq[String]) {
}
private def interpreterPlugins: Option[java.util.List[String]] = {
- val p = getAll(_plugins)
- p.map(_.asJava)
+ val defaults = List[String](
+ "PySpark:com.ibm.spark.kernel.interpreter.pyspark.PySparkInterpreter",
+ "SparkR:com.ibm.spark.kernel.interpreter.sparkr.SparkRInterpreter",
+ "SQL:com.ibm.spark.kernel.interpreter.sql.SqlInterpreter"
+ )
+
+ val userDefined = getAll(_interpreter_plugin) match {
+ case Some(l) => l
+ case _ => List[String]()
+ }
+
+ val p = defaults ++ userDefined
+ Some(p.asJava)
}
/**
http://git-wip-us.apache.org/repos/asf/incubator-toree/blob/502372d4/kernel/src/main/scala/com/ibm/spark/boot/layer/ComponentInitialization.scala
----------------------------------------------------------------------
diff --git a/kernel/src/main/scala/com/ibm/spark/boot/layer/ComponentInitialization.scala b/kernel/src/main/scala/com/ibm/spark/boot/layer/ComponentInitialization.scala
index df09028..e403e9b 100644
--- a/kernel/src/main/scala/com/ibm/spark/boot/layer/ComponentInitialization.scala
+++ b/kernel/src/main/scala/com/ibm/spark/boot/layer/ComponentInitialization.scala
@@ -86,14 +86,15 @@ trait StandardComponentInitialization extends ComponentInitialization {
val dependencyDownloader = initializeDependencyDownloader(config)
val magicLoader = initializeMagicLoader(
config, interpreter, dependencyDownloader)
- //val kernel = initializeKernel(
- // config, actorLoader, interpreter, commManager, magicLoader
- //)
+ val manager = InterpreterManager(config)
+ .addInterpreter("Scala",interpreter)
val kernel = initializeKernel(
- config, actorLoader, null, commManager, magicLoader
+ config, actorLoader, manager, commManager, magicLoader
)
val responseMap = initializeResponseMap()
+
+ /*
// NOTE: Tested via initializing the following and returning this
// interpreter instead of the Scala one
val pySparkInterpreter = new PySparkInterpreter(kernel)
@@ -144,10 +145,12 @@ trait StandardComponentInitialization extends ComponentInitialization {
interpreter
}
- kernel.interpreter = defaultInterpreter
+ */
+ //kernel.interpreter = defaultInterpreter
initializeSparkContext(config, kernel, appName)
- (commStorage, commRegistrar, commManager, defaultInterpreter, kernel,
+ (commStorage, commRegistrar, commManager,
+ manager.defaultInterpreter.getOrElse(null), kernel,
dependencyDownloader, magicLoader, responseMap)
}
@@ -278,14 +281,14 @@ trait StandardComponentInitialization extends ComponentInitialization {
private def initializeKernel(
config: Config,
actorLoader: ActorLoader,
- interpreter: Interpreter,
+ interpreterManager: InterpreterManager,
commManager: CommManager,
magicLoader: MagicLoader
) = {
val kernel = new Kernel(
config,
actorLoader,
- interpreter,
+ interpreterManager,
commManager,
magicLoader
)
http://git-wip-us.apache.org/repos/asf/incubator-toree/blob/502372d4/kernel/src/main/scala/com/ibm/spark/boot/layer/InterpreterManager.scala
----------------------------------------------------------------------
diff --git a/kernel/src/main/scala/com/ibm/spark/boot/layer/InterpreterManager.scala b/kernel/src/main/scala/com/ibm/spark/boot/layer/InterpreterManager.scala
new file mode 100644
index 0000000..e07af8e
--- /dev/null
+++ b/kernel/src/main/scala/com/ibm/spark/boot/layer/InterpreterManager.scala
@@ -0,0 +1,59 @@
+package com.ibm.spark.boot.layer
+
+import com.ibm.spark.kernel.api.KernelLike
+import com.typesafe.config.Config
+import com.ibm.spark.interpreter._
+import scala.collection.JavaConverters._
+
+case class InterpreterManager(
+ default: String = "Scala",
+ interpreters: Map[String, Interpreter] = Map[String, Interpreter]()
+) {
+
+ def initializeInterpreters(kernel: KernelLike): Unit = {
+ interpreters.values.foreach(interpreter =>
+ interpreter.init(kernel)
+ )
+ }
+
+ def addInterpreter(
+ name:String,
+ interpreter: Interpreter
+ ): InterpreterManager = {
+ copy(interpreters = interpreters + (name -> interpreter))
+ }
+
+ def defaultInterpreter(): Option[Interpreter] = {
+ interpreters.get(default)
+ }
+}
+
+object InterpreterManager {
+
+ def apply(config: Config): InterpreterManager = {
+ val p = config
+ .getStringList("interpreter_plugins")
+ .listIterator().asScala
+
+ val m = p.foldLeft(Map[String, Interpreter]())( (acc, v) => {
+ v.split(":") match {
+ case Array(name, className) =>
+ try {
+ val i = Class
+ .forName(className)
+ .newInstance()
+ .asInstanceOf[Interpreter]
+ acc + (name -> i)
+ }
+ catch {
+ case _:Throwable => acc
+ }
+ case _ => acc
+ }
+ })
+
+ val default = config.getString("default_interpreter")
+
+ InterpreterManager(interpreters = m, default = default)
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-toree/blob/502372d4/kernel/src/main/scala/com/ibm/spark/kernel/api/Kernel.scala
----------------------------------------------------------------------
diff --git a/kernel/src/main/scala/com/ibm/spark/kernel/api/Kernel.scala b/kernel/src/main/scala/com/ibm/spark/kernel/api/Kernel.scala
index 58676c3..f070004 100644
--- a/kernel/src/main/scala/com/ibm/spark/kernel/api/Kernel.scala
+++ b/kernel/src/main/scala/com/ibm/spark/kernel/api/Kernel.scala
@@ -20,6 +20,7 @@ import java.io.{OutputStream, InputStream, PrintStream}
import java.util.concurrent.ConcurrentHashMap
import com.ibm.spark.annotations.Experimental
+import com.ibm.spark.boot.layer.InterpreterManager
import com.ibm.spark.comm.CommManager
import com.ibm.spark.global
import com.ibm.spark.interpreter.Results.Result
@@ -46,7 +47,7 @@ import com.ibm.spark.global.ExecuteRequestState
* Represents the main kernel API to be used for interaction.
*
* @param config The configuration used when starting the kernel
- * @param interpreter The interpreter to expose in this instance
+ * @param interpreterManager The interpreter manager to expose in this instance
* @param comm The Comm manager to expose in this instance
* @param actorLoader The actor loader to use for message relaying
*/
@@ -54,7 +55,7 @@ import com.ibm.spark.global.ExecuteRequestState
class Kernel (
private val config: Config,
private val actorLoader: ActorLoader,
- var interpreter: Interpreter,
+ val interpreterManager: InterpreterManager,
val comm: CommManager,
val magicLoader: MagicLoader
) extends KernelLike with LogLike {
@@ -107,6 +108,11 @@ class Kernel (
*/
val data: java.util.Map[String, Any] = new ConcurrentHashMap[String, Any]()
+
+ interpreterManager.initializeInterpreters(this)
+
+ val interpreter = interpreterManager.defaultInterpreter().get
+
/**
* Handles the output of interpreting code.
* @param output the output of the interpreter
@@ -424,9 +430,12 @@ class Kernel (
@inline def getJarPathFor(klass: Class[_]): String =
klass.getProtectionDomain.getCodeSource.getLocation.getPath
+ val interpreterC = interpreterManager.interpreters.values.map(_.getClass)
+
// TODO: Provide less hard-coded solution in case additional dependencies
// are added or classes are refactored to different projects
- val jarPaths = Seq(
+ val classDep = Seq(
+
// Macro project
classOf[com.ibm.spark.annotations.Experimental],
@@ -440,17 +449,19 @@ class Kernel (
classOf[com.ibm.spark.kernel.api.KernelLike],
// Scala-interpreter project
- classOf[com.ibm.spark.kernel.interpreter.scala.ScalaInterpreter],
+ //classOf[com.ibm.spark.kernel.interpreter.scala.ScalaInterpreter],
// PySpark-interpreter project
- classOf[com.ibm.spark.kernel.interpreter.pyspark.PySparkInterpreter],
+ //classOf[com.ibm.spark.kernel.interpreter.pyspark.PySparkInterpreter],
// SparkR-interpreter project
- classOf[com.ibm.spark.kernel.interpreter.sparkr.SparkRInterpreter],
+ //classOf[com.ibm.spark.kernel.interpreter.sparkr.SparkRInterpreter],
// Kernel project
classOf[com.ibm.spark.boot.KernelBootstrap]
- ).map(getJarPathFor)
+ )
+
+ val jarPaths = (interpreterC ++ classDep).map(getJarPathFor)
logger.info("Adding kernel jars to cluster:\n- " +
jarPaths.mkString("\n- "))
@@ -460,6 +471,10 @@ class Kernel (
}
}
+ override def interpreter(name: String): Option[Interpreter] = {
+ interpreterManager.interpreters.get(name)
+ }
+
override def sparkContext: SparkContext = _sparkContext
override def sparkConf: SparkConf = _sparkConf
override def javaSparkContext: JavaSparkContext = _javaSparkContext
http://git-wip-us.apache.org/repos/asf/incubator-toree/blob/502372d4/kernel/src/test/scala/com/ibm/spark/boot/layer/StandardComponentInitializationSpec.scala
----------------------------------------------------------------------
diff --git a/kernel/src/test/scala/com/ibm/spark/boot/layer/StandardComponentInitializationSpec.scala b/kernel/src/test/scala/com/ibm/spark/boot/layer/StandardComponentInitializationSpec.scala
deleted file mode 100644
index 68ee0cc..0000000
--- a/kernel/src/test/scala/com/ibm/spark/boot/layer/StandardComponentInitializationSpec.scala
+++ /dev/null
@@ -1,84 +0,0 @@
-/*
- * Copyright 2014 IBM Corp.
- *
- * Licensed 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.ibm.spark.boot.layer
-
-import com.ibm.spark.boot.{CommandLineOptions, KernelBootstrap}
-import com.ibm.spark.interpreter.Interpreter
-import com.ibm.spark.kernel.api.KernelLike
-import com.ibm.spark.kernel.protocol.v5.KMBuilder
-import com.ibm.spark.kernel.protocol.v5.kernel.ActorLoader
-import com.ibm.spark.utils.LogLike
-import com.typesafe.config.Config
-import org.apache.spark.{SparkConf, SparkContext}
-import org.mockito.ArgumentCaptor
-import org.mockito.Matchers._
-import org.mockito.Mockito._
-import org.scalatest.mock.MockitoSugar
-import org.scalatest.{BeforeAndAfter, FunSpec, Matchers}
-
-import scala.collection.mutable
-import scala.collection.JavaConverters._
-class StandardComponentInitializationSpec extends FunSpec with Matchers
- with MockitoSugar with BeforeAndAfter
-{
- private val TestAppName = "test app"
-
- private var mockConfig: Config = _
- private var mockActorLoader: ActorLoader = _
- private var mockSparkContext: SparkContext = _
- private var mockInterpreter: Interpreter = _
- private var mockKernel: KernelLike = _
- private var spyComponentInitialization: StandardComponentInitialization = _
-
- private class TestComponentInitialization
- extends StandardComponentInitialization with LogLike
-
- before {
- mockConfig = mock[Config]
- mockActorLoader = mock[ActorLoader]
- mockSparkContext = mock[SparkContext]
- mockInterpreter = mock[Interpreter]
- mockKernel = mock[KernelLike]
-
- spyComponentInitialization = spy(new TestComponentInitialization())
- }
-
- describe("StandardComponentInitialization") {
- describe("#initializeInterpreterPlugins") {
- it("should return a map with the DummyInterpreter") {
- val conf = new CommandLineOptions(List(
- "--interpreter-plugin", "dummy:test.utils.DummyInterpreter",
- "--interpreter-plugin", "dummy2:test.utils.DummyInterpreter"
- )).toConfig
-
- val m = spyComponentInitialization
- .initializeInterpreterPlugins(mockKernel, conf)
-
- m.get("dummy") should not be None
- m.get("dummy2") should not be None
- }
- it("should return an empty map") {
- val conf = new CommandLineOptions(List()).toConfig
-
- val m = spyComponentInitialization
- .initializeInterpreterPlugins(mockKernel, conf)
-
- m.isEmpty shouldBe true
- }
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-toree/blob/502372d4/kernel/src/test/scala/com/ibm/spark/kernel/api/KernelSpec.scala
----------------------------------------------------------------------
diff --git a/kernel/src/test/scala/com/ibm/spark/kernel/api/KernelSpec.scala b/kernel/src/test/scala/com/ibm/spark/kernel/api/KernelSpec.scala
index 22cd08f..f5d5517 100644
--- a/kernel/src/test/scala/com/ibm/spark/kernel/api/KernelSpec.scala
+++ b/kernel/src/test/scala/com/ibm/spark/kernel/api/KernelSpec.scala
@@ -2,6 +2,7 @@ package com.ibm.spark.kernel.api
import java.io.{InputStream, PrintStream}
+import com.ibm.spark.boot.layer.InterpreterManager
import com.ibm.spark.comm.CommManager
import com.ibm.spark.interpreter._
import com.ibm.spark.kernel.protocol.v5._
@@ -31,6 +32,7 @@ class KernelSpec extends FunSpec with Matchers with MockitoSugar
private var mockSparkConf: SparkConf = _
private var mockActorLoader: ActorLoader = _
private var mockInterpreter: Interpreter = _
+ private var mockInterpreterManager: InterpreterManager = _
private var mockCommManager: CommManager = _
private var mockMagicLoader: MagicLoader = _
private var kernel: Kernel = _
@@ -39,8 +41,13 @@ class KernelSpec extends FunSpec with Matchers with MockitoSugar
before {
mockConfig = mock[Config]
mockInterpreter = mock[Interpreter]
+ mockInterpreterManager = mock[InterpreterManager]
mockSparkContext = mock[SparkContext]
mockSparkConf = mock[SparkConf]
+ when(mockInterpreterManager.defaultInterpreter())
+ .thenReturn(Some(mockInterpreter))
+ when(mockInterpreterManager.interpreters)
+ .thenReturn(Map[String, com.ibm.spark.interpreter.Interpreter]())
when(mockInterpreter.interpret(BadCode.get))
.thenReturn((Results.Incomplete, null))
when(mockInterpreter.interpret(GoodCode.get))
@@ -48,12 +55,13 @@ class KernelSpec extends FunSpec with Matchers with MockitoSugar
when(mockInterpreter.interpret(ErrorCode.get))
.thenReturn((Results.Error, Right(ExecuteError("error","bad", List("1")))))
+
mockCommManager = mock[CommManager]
mockActorLoader = mock[ActorLoader]
mockMagicLoader = mock[MagicLoader]
kernel = new Kernel(
- mockConfig, mockActorLoader, mockInterpreter, mockCommManager,
+ mockConfig, mockActorLoader, mockInterpreterManager, mockCommManager,
mockMagicLoader
)
http://git-wip-us.apache.org/repos/asf/incubator-toree/blob/502372d4/kernel/src/test/scala/test/utils/DummyInterpreter.scala
----------------------------------------------------------------------
diff --git a/kernel/src/test/scala/test/utils/DummyInterpreter.scala b/kernel/src/test/scala/test/utils/DummyInterpreter.scala
index 1ab8cac..ee7c096 100644
--- a/kernel/src/test/scala/test/utils/DummyInterpreter.scala
+++ b/kernel/src/test/scala/test/utils/DummyInterpreter.scala
@@ -102,4 +102,11 @@ class DummyInterpreter(kernel: KernelLike) extends Interpreter {
* @return The cursor position and list of possible completions
*/
override def completion(code: String, pos: Int): (Int, List[String]) = ???
+
+ /**
+ * Initializes the interpreter.
+ * @param kernel The kernel
+ * @return The newly initialized interpreter
+ */
+ override def init(kernel: KernelLike): Interpreter = ???
}
http://git-wip-us.apache.org/repos/asf/incubator-toree/blob/502372d4/kernel/src/test/scala/test/utils/SparkKernelDeployer.scala
----------------------------------------------------------------------
diff --git a/kernel/src/test/scala/test/utils/SparkKernelDeployer.scala b/kernel/src/test/scala/test/utils/SparkKernelDeployer.scala
index fe97463..4f0412a 100644
--- a/kernel/src/test/scala/test/utils/SparkKernelDeployer.scala
+++ b/kernel/src/test/scala/test/utils/SparkKernelDeployer.scala
@@ -83,6 +83,7 @@ object SparkKernelDeployer extends LogLike with MockitoSugar {
}
+
/*
def reallyInitializeSparkContext(
config: Config,
http://git-wip-us.apache.org/repos/asf/incubator-toree/blob/502372d4/pyspark-interpreter/src/main/scala/com/ibm/spark/kernel/interpreter/pyspark/PySparkInterpreter.scala
----------------------------------------------------------------------
diff --git a/pyspark-interpreter/src/main/scala/com/ibm/spark/kernel/interpreter/pyspark/PySparkInterpreter.scala b/pyspark-interpreter/src/main/scala/com/ibm/spark/kernel/interpreter/pyspark/PySparkInterpreter.scala
index ab1d061..38e1d68 100644
--- a/pyspark-interpreter/src/main/scala/com/ibm/spark/kernel/interpreter/pyspark/PySparkInterpreter.scala
+++ b/pyspark-interpreter/src/main/scala/com/ibm/spark/kernel/interpreter/pyspark/PySparkInterpreter.scala
@@ -33,12 +33,11 @@ import scala.tools.nsc.interpreter.{InputStream, OutputStream}
* SPARK_HOME, PYTHONPATH pointing to Spark's Python source, and py4j installed
* where it is accessible to the Spark Kernel.
*
- * @param _kernel The kernel API to expose to the PySpark instance
*/
class PySparkInterpreter(
- private val _kernel: KernelLike
) extends Interpreter {
private val logger = LoggerFactory.getLogger(this.getClass)
+ private var _kernel:KernelLike = _
// TODO: Replace hard-coded maximum queue count
/** Represents the state used by this interpreter's Python instance. */
@@ -50,6 +49,7 @@ class PySparkInterpreter(
_kernel
)
+
/** Represents the interface for Python to talk to JVM Spark components. */
private lazy val gatewayServer = new GatewayServer(pySparkBridge, 0)
@@ -69,6 +69,21 @@ class PySparkInterpreter(
private lazy val pySparkTransformer = new PySparkTransformer
/**
+ * Initializes the interpreter.
+ * @param kernel The kernel
+ * @return The newly initialized interpreter
+ */
+ override def init(kernel: KernelLike): Interpreter = {
+ _kernel = kernel
+ this
+ }
+
+
+ override def bindSparkContext(sparkContext: SparkContext) = {
+
+ }
+
+ /**
* Executes the provided code with the option to silence output.
* @param code The code to execute
* @param silent Whether or not to execute the code silently (no output)
@@ -128,7 +143,7 @@ class PySparkInterpreter(
override def updatePrintStreams(in: InputStream, out: OutputStream, err: OutputStream): Unit = ???
// Unsupported
- override def classServerURI: String = ???
+ override def classServerURI: String = ""
// Unsupported
override def interrupt(): Interpreter = ???
@@ -141,4 +156,5 @@ class PySparkInterpreter(
// Unsupported
override def doQuietly[T](body: => T): T = ???
+
}
http://git-wip-us.apache.org/repos/asf/incubator-toree/blob/502372d4/pyspark-interpreter/src/main/scala/com/ibm/spark/magic/builtin/PySpark.scala
----------------------------------------------------------------------
diff --git a/pyspark-interpreter/src/main/scala/com/ibm/spark/magic/builtin/PySpark.scala b/pyspark-interpreter/src/main/scala/com/ibm/spark/magic/builtin/PySpark.scala
index 5fed927..a0a79b5 100644
--- a/pyspark-interpreter/src/main/scala/com/ibm/spark/magic/builtin/PySpark.scala
+++ b/pyspark-interpreter/src/main/scala/com/ibm/spark/magic/builtin/PySpark.scala
@@ -26,7 +26,7 @@ import com.ibm.spark.magic.dependencies.IncludeKernel
*/
class PySpark extends CellMagic with IncludeKernel {
override def execute(code: String): CellMagicOutput = {
- val pySpark = Option(kernel.data.get("PySpark"))
+ val pySpark = kernel.interpreter("PySpark")
if (pySpark.isEmpty || pySpark.get == null)
throw new PySparkException("PySpark is not available!")
http://git-wip-us.apache.org/repos/asf/incubator-toree/blob/502372d4/resources/compile/reference.conf
----------------------------------------------------------------------
diff --git a/resources/compile/reference.conf b/resources/compile/reference.conf
index c9d12c9..07a49ac 100644
--- a/resources/compile/reference.conf
+++ b/resources/compile/reference.conf
@@ -57,8 +57,5 @@ send_empty_output = ${?SEND_EMPTY_OUTPUT}
jar_dir = ${?JAR_DIR}
-default_interpreter = "scala"
+default_interpreter = "Scala"
default_interpreter = ${?DEFAULT_INTERPRETER}
-
-sparkcontext = "yes"
-
http://git-wip-us.apache.org/repos/asf/incubator-toree/blob/502372d4/resources/test/reference.conf
----------------------------------------------------------------------
diff --git a/resources/test/reference.conf b/resources/test/reference.conf
index d506c8a..407f344 100644
--- a/resources/test/reference.conf
+++ b/resources/test/reference.conf
@@ -55,5 +55,5 @@ max_interpreter_threads = ${?MAX_INTERPRETER_THREADS}
send_empty_output = false
send_empty_output = ${?SEND_EMPTY_OUTPUT}
-default_interpreter = "scala"
+default_interpreter = "Scala"
default_interpreter = ${?DEFAULT_INTERPRETER}
http://git-wip-us.apache.org/repos/asf/incubator-toree/blob/502372d4/scala-interpreter/src/main/scala/com/ibm/spark/kernel/interpreter/scala/ScalaInterpreter.scala
----------------------------------------------------------------------
diff --git a/scala-interpreter/src/main/scala/com/ibm/spark/kernel/interpreter/scala/ScalaInterpreter.scala b/scala-interpreter/src/main/scala/com/ibm/spark/kernel/interpreter/scala/ScalaInterpreter.scala
index f774a5a..bccdf62 100644
--- a/scala-interpreter/src/main/scala/com/ibm/spark/kernel/interpreter/scala/ScalaInterpreter.scala
+++ b/scala-interpreter/src/main/scala/com/ibm/spark/kernel/interpreter/scala/ScalaInterpreter.scala
@@ -27,7 +27,7 @@ import com.ibm.spark.global.StreamState
import com.ibm.spark.interpreter
import com.ibm.spark.interpreter._
import com.ibm.spark.interpreter.imports.printers.{WrapperConsole, WrapperSystem}
-import com.ibm.spark.kernel.api.KernelOptions
+import com.ibm.spark.kernel.api.{KernelLike, KernelOptions}
import com.ibm.spark.utils.{MultiOutputStream, TaskManager}
import org.apache.spark.SparkContext
import org.apache.spark.repl.{SparkIMain, SparkJLineCompletion}
@@ -190,6 +190,17 @@ class ScalaInterpreter(
)
}
+ override def init(kernel: KernelLike): Interpreter = {
+ doQuietly {
+ bind(
+ "kernel", "com.ibm.spark.kernel.api.Kernel",
+ kernel, List( """@transient implicit""")
+ )
+ }
+
+ this
+ }
+
override def interrupt(): Interpreter = {
require(sparkIMain != null && taskManager != null)
http://git-wip-us.apache.org/repos/asf/incubator-toree/blob/502372d4/scala-interpreter/src/main/scala/com/ibm/spark/magic/builtin/Scala.scala
----------------------------------------------------------------------
diff --git a/scala-interpreter/src/main/scala/com/ibm/spark/magic/builtin/Scala.scala b/scala-interpreter/src/main/scala/com/ibm/spark/magic/builtin/Scala.scala
index c739bc6..c850926 100644
--- a/scala-interpreter/src/main/scala/com/ibm/spark/magic/builtin/Scala.scala
+++ b/scala-interpreter/src/main/scala/com/ibm/spark/magic/builtin/Scala.scala
@@ -26,7 +26,7 @@ import com.ibm.spark.magic.{CellMagic, CellMagicOutput}
*/
class Scala extends CellMagic with IncludeKernel {
override def execute(code: String): CellMagicOutput = {
- val scala = Option(kernel.data.get("Scala"))
+ val scala = kernel.interpreter("Scala")
if (scala.isEmpty || scala.get == null)
throw new ScalaException("Scala is not available!")
http://git-wip-us.apache.org/repos/asf/incubator-toree/blob/502372d4/sparkr-interpreter/src/main/scala/com/ibm/spark/kernel/interpreter/sparkr/SparkRInterpreter.scala
----------------------------------------------------------------------
diff --git a/sparkr-interpreter/src/main/scala/com/ibm/spark/kernel/interpreter/sparkr/SparkRInterpreter.scala b/sparkr-interpreter/src/main/scala/com/ibm/spark/kernel/interpreter/sparkr/SparkRInterpreter.scala
index 0e52f9d..a950da0 100644
--- a/sparkr-interpreter/src/main/scala/com/ibm/spark/kernel/interpreter/sparkr/SparkRInterpreter.scala
+++ b/sparkr-interpreter/src/main/scala/com/ibm/spark/kernel/interpreter/sparkr/SparkRInterpreter.scala
@@ -32,12 +32,11 @@ import scala.tools.nsc.interpreter.{InputStream, OutputStream}
* SPARK_HOME pointing to a binary distribution (needs packaged SparkR library)
* and an implementation of R on the path.
*
- * @param _kernel The kernel API to expose to the SparkR instance
*/
class SparkRInterpreter(
- private val _kernel: KernelLike
) extends Interpreter {
private val logger = LoggerFactory.getLogger(this.getClass)
+ private var _kernel: KernelLike = _
// TODO: Replace hard-coded maximum queue count
/** Represents the state used by this interpreter's R instance. */
@@ -67,6 +66,11 @@ class SparkRInterpreter(
)
private lazy val sparkRTransformer = new SparkRTransformer
+ override def init(kernel: KernelLike): Interpreter = {
+ _kernel = kernel
+ this
+ }
+
/**
* Executes the provided code with the option to silence output.
* @param code The code to execute
@@ -127,7 +131,9 @@ class SparkRInterpreter(
override def updatePrintStreams(in: InputStream, out: OutputStream, err: OutputStream): Unit = ???
// Unsupported
- override def classServerURI: String = ???
+ override def classServerURI: String = ""
+
+ override def bindSparkContext(sparkContext: SparkContext): Unit = {}
// Unsupported
override def interrupt(): Interpreter = ???
http://git-wip-us.apache.org/repos/asf/incubator-toree/blob/502372d4/sparkr-interpreter/src/main/scala/com/ibm/spark/magic/builtin/SparkR.scala
----------------------------------------------------------------------
diff --git a/sparkr-interpreter/src/main/scala/com/ibm/spark/magic/builtin/SparkR.scala b/sparkr-interpreter/src/main/scala/com/ibm/spark/magic/builtin/SparkR.scala
index cbbd8e2..7eba136 100644
--- a/sparkr-interpreter/src/main/scala/com/ibm/spark/magic/builtin/SparkR.scala
+++ b/sparkr-interpreter/src/main/scala/com/ibm/spark/magic/builtin/SparkR.scala
@@ -26,7 +26,7 @@ import com.ibm.spark.magic.dependencies.IncludeKernel
*/
class SparkR extends CellMagic with IncludeKernel {
override def execute(code: String): CellMagicOutput = {
- val sparkR = Option(kernel.data.get("SparkR"))
+ val sparkR = kernel.interpreter("SparkR")
if (sparkR.isEmpty || sparkR.get == null)
throw new SparkRException("SparkR is not available!")
http://git-wip-us.apache.org/repos/asf/incubator-toree/blob/502372d4/sql-interpreter/src/main/scala/com/ibm/spark/kernel/interpreter/sql/SqlInterpreter.scala
----------------------------------------------------------------------
diff --git a/sql-interpreter/src/main/scala/com/ibm/spark/kernel/interpreter/sql/SqlInterpreter.scala b/sql-interpreter/src/main/scala/com/ibm/spark/kernel/interpreter/sql/SqlInterpreter.scala
index ad06fbd..76ae262 100644
--- a/sql-interpreter/src/main/scala/com/ibm/spark/kernel/interpreter/sql/SqlInterpreter.scala
+++ b/sql-interpreter/src/main/scala/com/ibm/spark/kernel/interpreter/sql/SqlInterpreter.scala
@@ -20,6 +20,7 @@ import java.net.URL
import com.ibm.spark.interpreter.{ExecuteFailure, ExecuteOutput, Interpreter}
import com.ibm.spark.interpreter.Results.Result
import com.ibm.spark.kernel.api.KernelLike
+import org.apache.spark.SparkContext
import org.apache.spark.sql.SQLContext
import scala.concurrent.duration._
@@ -30,9 +31,15 @@ import scala.tools.nsc.interpreter.{OutputStream, InputStream}
* Represents an interpreter interface to Spark SQL.
*/
class SqlInterpreter(private val kernel: KernelLike) extends Interpreter {
- private lazy val sqlService = new SqlService(kernel)
+ private var _kernel: KernelLike = _
+ private lazy val sqlService = new SqlService(_kernel)
private lazy val sqlTransformer = new SqlTransformer
+ override def init(kernel: KernelLike): Interpreter = {
+ _kernel = kernel
+ this
+ }
+
/**
* Executes the provided code with the option to silence output.
* @param code The code to execute
@@ -93,7 +100,9 @@ class SqlInterpreter(private val kernel: KernelLike) extends Interpreter {
override def updatePrintStreams(in: InputStream, out: OutputStream, err: OutputStream): Unit = ???
// Unsupported
- override def classServerURI: String = ???
+ override def classServerURI: String = ""
+
+ override def bindSparkContext(sparkContext: SparkContext): Unit = {}
// Unsupported
override def interrupt(): Interpreter = ???
http://git-wip-us.apache.org/repos/asf/incubator-toree/blob/502372d4/sql-interpreter/src/main/scala/com/ibm/spark/magic/builtin/Sql.scala
----------------------------------------------------------------------
diff --git a/sql-interpreter/src/main/scala/com/ibm/spark/magic/builtin/Sql.scala b/sql-interpreter/src/main/scala/com/ibm/spark/magic/builtin/Sql.scala
index a15817b..a8f439c 100644
--- a/sql-interpreter/src/main/scala/com/ibm/spark/magic/builtin/Sql.scala
+++ b/sql-interpreter/src/main/scala/com/ibm/spark/magic/builtin/Sql.scala
@@ -26,7 +26,7 @@ import com.ibm.spark.magic.dependencies.IncludeKernel
*/
class Sql extends CellMagic with IncludeKernel {
override def execute(code: String): CellMagicOutput = {
- val sparkR = Option(kernel.data.get("SQL"))
+ val sparkR = kernel.interpreter("SQL")
if (sparkR.isEmpty || sparkR.get == null)
throw new SqlException("SQL is not available!")