Move ZeppelinContext code to Scala

cassandra/pom.xml

@@ -35,8 +35,8 @@
 
     <properties>
         <cassandra.driver.version>2.1.7.1</cassandra.driver.version>
-        <scala.version>2.11.7</scala.version>
-        <scala.binary.version>2.11</scala.binary.version>
+        <scala.version>2.10.4</scala.version>
+        <scala.binary.version>2.10</scala.binary.version>
         <commons-lang.version>3.3.2</commons-lang.version>
         <scalate.version>1.7.1</scalate.version>
 

spark/pom.xml

@@ -48,7 +48,7 @@
 
     <akka.group>org.spark-project.akka</akka.group>
     <akka.version>2.3.4-spark</akka.version>
-
+    <mockito.version>1.9.5</mockito.version>
     <spark.download.url>http://www.apache.org/dist/spark/spark-${spark.version}/spark-${spark.version}.tgz</spark.download.url>
   </properties>
 
@@ -494,6 +494,13 @@
       <artifactId>junit</artifactId>
       <scope>test</scope>
     </dependency>
+
+    <dependency>
+      <groupId>org.mockito</groupId>
+      <artifactId>mockito-core</artifactId>
+      <version>${mockito.version}</version>
+      <scope>test</scope>
+    </dependency>
   </dependencies>
 
   <profiles>
@@ -1002,7 +1009,19 @@
           </execution>
         </executions>
       </plugin>
-
+      <plugin>
+        <groupId>org.scalatest</groupId>
+        <artifactId>scalatest-maven-plugin</artifactId>
+        <version>1.0</version>
+        <executions>
+          <execution>
+            <id>test</id>
+            <goals>
+              <goal>test</goal>
+            </goals>
+          </execution>
+        </executions>
+      </plugin>
     </plugins>
   </build>
 </project>

spark/src/main/java/org/apache/zeppelin/spark/PySparkInterpreter.java

@@ -42,6 +42,7 @@
 import org.apache.spark.SparkConf;
 import org.apache.spark.api.java.JavaSparkContext;
 import org.apache.spark.sql.SQLContext;
+import org.apache.zeppelin.context.ZeppelinContext;
 import org.apache.zeppelin.interpreter.Interpreter;
 import org.apache.zeppelin.interpreter.InterpreterContext;
 import org.apache.zeppelin.interpreter.InterpreterException;

spark/src/main/java/org/apache/zeppelin/spark/SparkInterpreter.java

@@ -33,6 +33,7 @@
 import org.apache.spark.SparkConf;
 import org.apache.spark.SparkContext;
 import org.apache.spark.SparkEnv;
+import org.apache.zeppelin.spark.display.SparkDisplayFunctionsHelper$;
 import org.apache.spark.repl.SparkCommandLine;
 import org.apache.spark.repl.SparkILoop;
 import org.apache.spark.repl.SparkIMain;
@@ -43,6 +44,7 @@
 import org.apache.spark.scheduler.Stage;
 import org.apache.spark.sql.SQLContext;
 import org.apache.spark.ui.jobs.JobProgressListener;
+import org.apache.zeppelin.context.ZeppelinContext;
 import org.apache.zeppelin.interpreter.Interpreter;
 import org.apache.zeppelin.interpreter.InterpreterContext;
 import org.apache.zeppelin.interpreter.InterpreterException;
@@ -335,14 +337,19 @@ public void open() {
     // https://groups.google.com/forum/#!topic/scala-user/MlVwo2xCCI0
 
     /*
-     * > val env = new nsc.Settings(errLogger) > env.usejavacp.value = true > val p = new
-     * Interpreter(env) > p.setContextClassLoader > Alternatively you can set the class path through
-     * nsc.Settings.classpath.
+     * > val env = new nsc.Settings(errLogger)
+     * > env.usejavacp.value = true
+     * > val p = new Interpreter(env)
+     * > p.setContextClassLoader
+     * > Alternatively you can set the class path through nsc.Settings.classpath.
      *
-     * >> val settings = new Settings() >> settings.usejavacp.value = true >>
-     * settings.classpath.value += File.pathSeparator + >> System.getProperty("java.class.path") >>
-     * val in = new Interpreter(settings) { >> override protected def parentClassLoader =
-     * getClass.getClassLoader >> } >> in.setContextClassLoader()
+     * >> val settings = new Settings()
+     * >> settings.usejavacp.value = true
+     * >> settings.classpath.value += File.pathSeparator + System.getProperty("java.class.path")
+     * >> val in = new Interpreter(settings) {
+     * >>   override protected def parentClassLoader = getClass.getClassLoader
+     * >> }
+     * >> in.setContextClassLoader()
      */
     Settings settings = new Settings();
     if (getProperty("args") != null) {
@@ -433,18 +440,24 @@ public void open() {
 
     dep = getDependencyResolver();
 
-    z = new ZeppelinContext(sc, sqlc, null, dep, printStream,
-        Integer.parseInt(getProperty("zeppelin.spark.maxResult")));
+    final int defaultSparkMaxResult = Integer.parseInt(getProperty("zeppelin.spark.maxResult"));
+    z = new ZeppelinContext(defaultSparkMaxResult);
+
+    SparkDisplayFunctionsHelper$.MODULE$.registerDisplayFunctions(sc, z);
 
     intp.interpret("@transient var _binder = new java.util.HashMap[String, Object]()");
     binder = (Map<String, Object>) getValue("_binder");
     binder.put("sc", sc);
     binder.put("sqlc", sqlc);
     binder.put("z", z);
+    binder.put("dep", dep);
     binder.put("out", printStream);
 
     intp.interpret("@transient val z = "
-                 + "_binder.get(\"z\").asInstanceOf[org.apache.zeppelin.spark.ZeppelinContext]");
+            + "_binder.get(\"z\").asInstanceOf[org.apache.zeppelin.context.ZeppelinContext]");
+    intp.interpret("@transient val dep = "
+            + "_binder.get(\"dep\").asInstanceOf" +
+            "[org.apache.zeppelin.spark.dep.DependencyResolver]");
     intp.interpret("@transient val sc = "
                  + "_binder.get(\"sc\").asInstanceOf[org.apache.spark.SparkContext]");
     intp.interpret("@transient val sqlc = "
@@ -467,17 +480,6 @@ public void open() {
       intp.interpret("import org.apache.spark.sql.functions._");
     }
 
-    /* Temporary disabling DisplayUtils. see https://issues.apache.org/jira/browse/ZEPPELIN-127
-     *
-    // Utility functions for display
-    intp.interpret("import org.apache.zeppelin.spark.utils.DisplayUtils._");
-
-    // Scala implicit value for spark.maxResult
-    intp.interpret("import org.apache.zeppelin.spark.utils.SparkMaxResult");
-    intp.interpret("implicit val sparkMaxResult = new SparkMaxResult(" +
-            Integer.parseInt(getProperty("zeppelin.spark.maxResult")) + ")");
-     */
-
     try {
       if (sc.version().startsWith("1.1") || sc.version().startsWith("1.2")) {
         Method loadFiles = this.interpreter.getClass().getMethod("loadFiles", Settings.class);

spark/src/main/java/org/apache/zeppelin/spark/SparkSqlInterpreter.java

@@ -17,8 +17,11 @@
 
 package org.apache.zeppelin.spark;
 
+import java.io.ByteArrayOutputStream;
+import java.io.PrintStream;
 import java.lang.reflect.InvocationTargetException;
 import java.lang.reflect.Method;
+import java.util.Arrays;
 import java.util.List;
 import java.util.Properties;
 import java.util.Set;
@@ -30,6 +33,8 @@
 import org.apache.spark.scheduler.Stage;
 import org.apache.spark.sql.SQLContext;
 import org.apache.spark.ui.jobs.JobProgressListener;
+import org.apache.zeppelin.context.ZeppelinContext;
+import org.apache.zeppelin.display.DisplayParams;
 import org.apache.zeppelin.interpreter.Interpreter;
 import org.apache.zeppelin.interpreter.InterpreterContext;
 import org.apache.zeppelin.interpreter.InterpreterException;
@@ -120,7 +125,8 @@ public void close() {}
   public InterpreterResult interpret(String st, InterpreterContext context) {
     SQLContext sqlc = null;
 
-    sqlc = getSparkInterpreter().getSQLContext();
+    final SparkInterpreter sparkInterpreter = getSparkInterpreter();
+    sqlc = sparkInterpreter.getSQLContext();
 
     SparkContext sc = sqlc.sparkContext();
     if (concurrentSQL()) {
@@ -131,8 +137,13 @@ public InterpreterResult interpret(String st, InterpreterContext context) {
 
 
     Object rdd = sqlc.sql(st);
-    String msg = ZeppelinContext.showRDD(sc, context, rdd, maxResult);
-    return new InterpreterResult(Code.SUCCESS, msg);
+    final ZeppelinContext zeppelinContext = sparkInterpreter.getZeppelinContext();
+    zeppelinContext.setInterpreterContext(context);
+    final ByteArrayOutputStream out = new ByteArrayOutputStream();
+    PrintStream stream = new PrintStream(out);
+    zeppelinContext.display(rdd, new DisplayParams(maxResult, stream, context,
+            new java.util.ArrayList<String>()));
+    return new InterpreterResult(Code.SUCCESS, out.toString());
   }
 
   @Override