[ZEPPELIN-554] Streaming interpreter output to front-end

flink/src/test/java/org/apache/zeppelin/flink/FlinkInterpreterTest.java

@@ -40,7 +40,7 @@ public static void setUp() {
     Properties p = new Properties();
     flink = new FlinkInterpreter(p);
     flink.open();
-    context = new InterpreterContext(null, null, null, null, null, null, null, null);
+    context = new InterpreterContext(null, null, null, null, null, null, null, null, null);
   }
 
   @AfterClass

hive/src/test/java/org/apache/zeppelin/hive/HiveInterpreterTest.java

@@ -79,9 +79,9 @@ public void readTest() throws IOException {
     HiveInterpreter t = new HiveInterpreter(properties);
     t.open();
 
-    assertTrue(t.interpret("show databases", new InterpreterContext("", "1", "","", null,null,null,null)).message().contains("SCHEMA_NAME"));
+    assertTrue(t.interpret("show databases", new InterpreterContext("", "1", "","", null,null,null,null,null)).message().contains("SCHEMA_NAME"));
     assertEquals("ID\tNAME\na\ta_name\nb\tb_name\n",
-        t.interpret("select * from test_table", new InterpreterContext("", "1", "","", null,null,null,null)).message());
+        t.interpret("select * from test_table", new InterpreterContext("", "1", "","", null,null,null,null,null)).message());
   }
 
   @Test
@@ -101,7 +101,7 @@ public void readTestWithConfiguration() throws IOException {
     t.open();
 
     assertEquals("ID\tNAME\na\ta_name\nb\tb_name\n",
-        t.interpret("(h2)\n select * from test_table", new InterpreterContext("", "1", "","", null,null,null,null)).message());
+        t.interpret("(h2)\n select * from test_table", new InterpreterContext("", "1", "","", null,null,null,null,null)).message());
   }
 
   @Test
@@ -117,13 +117,13 @@ public void jdbcRestart() throws IOException, SQLException, ClassNotFoundExcepti
     t.open();
 
     InterpreterResult interpreterResult =
-        t.interpret("select * from test_table", new InterpreterContext("", "1", "","", null,null,null,null));
+        t.interpret("select * from test_table", new InterpreterContext("", "1", "","", null,null,null,null,null));
     assertEquals("ID\tNAME\na\ta_name\nb\tb_name\n", interpreterResult.message());
 
     t.getConnection("default").close();
 
     interpreterResult =
-        t.interpret("select * from test_table", new InterpreterContext("", "1", "","", null,null,null,null));
+        t.interpret("select * from test_table", new InterpreterContext("", "1", "","", null,null,null,null,null));
     assertEquals("ID\tNAME\na\ta_name\nb\tb_name\n", interpreterResult.message());
   }
 
@@ -139,7 +139,7 @@ public void test() throws IOException {
     HiveInterpreter t = new HiveInterpreter(properties);
     t.open();
 
-    InterpreterContext interpreterContext = new InterpreterContext(null, "a", null, null, null, null, null, null);
+    InterpreterContext interpreterContext = new InterpreterContext(null, "a", null, null, null, null, null, null, null);
 
     //simple select test
     InterpreterResult result = t.interpret("select * from test_table", interpreterContext);

ignite/src/test/java/org/apache/zeppelin/ignite/IgniteInterpreterTest.java

@@ -40,7 +40,7 @@ public class IgniteInterpreterTest {
   private static final String HOST = "127.0.0.1:47500..47509";
 
   private static final InterpreterContext INTP_CONTEXT =
-          new InterpreterContext(null, null, null, null, null, null, null, null);
+          new InterpreterContext(null, null, null, null, null, null, null, null, null);
 
   private IgniteInterpreter intp;
   private Ignite ignite;

ignite/src/test/java/org/apache/zeppelin/ignite/IgniteSqlInterpreterTest.java

@@ -44,7 +44,7 @@ public class IgniteSqlInterpreterTest {
   private static final String HOST = "127.0.0.1:47500..47509";
 
   private static final InterpreterContext INTP_CONTEXT =
-          new InterpreterContext(null, null, null, null, null, null, null, null);
+          new InterpreterContext(null, null, null, null, null, null, null, null, null);
 
   private Ignite ignite;
   private IgniteSqlInterpreter intp;

scalding/src/test/java/org/apache/zeppelin/scalding/ScaldingInterpreterTest.java

@@ -65,7 +65,7 @@ public void setUp() throws Exception {
     context = new InterpreterContext("note", "id", "title", "text",
         new HashMap<String, Object>(), new GUI(), new AngularObjectRegistry(
             intpGroup.getId(), null),
-        new LinkedList<InterpreterContextRunner>());
+        new LinkedList<InterpreterContextRunner>(), null);
   }
 
   @After

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

@@ -73,8 +73,7 @@ public class PySparkInterpreter extends Interpreter implements ExecuteResultHand
   private GatewayServer gatewayServer;
   private DefaultExecutor executor;
   private int port;
-  private ByteArrayOutputStream outputStream;
-  private ByteArrayOutputStream errStream;
+  private SparkOutputStream outputStream;
   private BufferedWriter ins;
   private PipedInputStream in;
   private ByteArrayOutputStream input;
@@ -173,7 +172,7 @@ private void createGatewayServerAndStartScript() {
     cmd.addArgument(Integer.toString(port), false);
     cmd.addArgument(Integer.toString(getSparkInterpreter().getSparkVersion().toNumber()), false);
     executor = new DefaultExecutor();
-    outputStream = new ByteArrayOutputStream();
+    outputStream = new SparkOutputStream();
     PipedOutputStream ps = new PipedOutputStream();
     in = null;
     try {
@@ -274,7 +273,6 @@ public void setStatementsFinished(String out, boolean error) {
       statementError = error;
       statementFinishedNotifier.notify();
     }
-
   }
 
   boolean pythonScriptInitialized = false;
@@ -287,6 +285,10 @@ public void onPythonScriptInitialized() {
     }
   }
 
+  public void appendOutput(String message) throws IOException {
+    outputStream.getInterpreterOutput().write(message);
+  }
+
   @Override
   public InterpreterResult interpret(String st, InterpreterContext context) {
     SparkInterpreter sparkInterpreter = getSparkInterpreter();
@@ -300,7 +302,7 @@ public InterpreterResult interpret(String st, InterpreterContext context) {
           + outputStream.toString());
     }
 
-    outputStream.reset();
+    outputStream.setInterpreterOutput(context.out);
 
     synchronized (pythonScriptInitializeNotifier) {
       long startTime = System.currentTimeMillis();
@@ -314,15 +316,24 @@ public InterpreterResult interpret(String st, InterpreterContext context) {
       }
     }
 
+    String errorMessage = "";
+    try {
+      context.out.flush();
+      errorMessage = new String(context.out.toByteArray());
+    } catch (IOException e) {
+      throw new InterpreterException(e);
+    }
+
+
     if (pythonscriptRunning == false) {
       // python script failed to initialize and terminated
       return new InterpreterResult(Code.ERROR, "failed to start pyspark"
-          + outputStream.toString());
+          + errorMessage);
     }
     if (pythonScriptInitialized == false) {
       // timeout. didn't get initialized message
       return new InterpreterResult(Code.ERROR, "pyspark is not responding "
-          + outputStream.toString());
+          + errorMessage);
     }
 
     if (!sparkInterpreter.getSparkVersion().isPysparkSupported()) {
@@ -352,7 +363,14 @@ public InterpreterResult interpret(String st, InterpreterContext context) {
     if (statementError) {
       return new InterpreterResult(Code.ERROR, statementOutput);
     } else {
-      return new InterpreterResult(Code.SUCCESS, statementOutput);
+
+      try {
+        context.out.flush();
+      } catch (IOException e) {
+        throw new InterpreterException(e);
+      }
+
+      return new InterpreterResult(Code.SUCCESS);
     }
   }
 
@@ -389,8 +407,6 @@ public List<String> completion(String buf, int cursor) {
       return new LinkedList<String>();
     }
 
-    outputStream.reset();
-
     pythonInterpretRequest = new PythonInterpretRequest(completionCommand, "");
     statementOutput = null;
 

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

@@ -17,9 +17,7 @@
 
 package org.apache.zeppelin.spark;
 
-import java.io.ByteArrayOutputStream;
 import java.io.File;
-import java.io.PrintStream;
 import java.io.PrintWriter;
 import java.lang.reflect.Constructor;
 import java.lang.reflect.InvocationTargetException;
@@ -41,7 +39,6 @@
 import org.apache.spark.scheduler.ActiveJob;
 import org.apache.spark.scheduler.DAGScheduler;
 import org.apache.spark.scheduler.Pool;
-import org.apache.spark.scheduler.SparkListener;
 import org.apache.spark.sql.SQLContext;
 import org.apache.spark.ui.jobs.JobProgressListener;
 import org.apache.zeppelin.interpreter.Interpreter;
@@ -115,7 +112,7 @@ public class SparkInterpreter extends Interpreter {
   private SparkILoop interpreter;
   private SparkIMain intp;
   private SparkContext sc;
-  private ByteArrayOutputStream out;
+  private SparkOutputStream out;
   private SQLContext sqlc;
   private SparkDependencyResolver dep;
   private SparkJLineCompletion completor;
@@ -129,7 +126,7 @@ public class SparkInterpreter extends Interpreter {
 
   public SparkInterpreter(Properties property) {
     super(property);
-    out = new ByteArrayOutputStream();
+    out = new SparkOutputStream();
   }
 
   public SparkInterpreter(Properties property, SparkContext sc) {
@@ -452,10 +449,9 @@ public void open() {
     b.v_$eq(true);
     settings.scala$tools$nsc$settings$StandardScalaSettings$_setter_$usejavacp_$eq(b);
 
-    PrintStream printStream = new PrintStream(out);
-
     /* spark interpreter */
     this.interpreter = new SparkILoop(null, new PrintWriter(out));
+
     interpreter.settings_$eq(settings);
 
     interpreter.createInterpreter();
@@ -481,15 +477,14 @@ public void open() {
 
     dep = getDependencyResolver();
 
-    z = new ZeppelinContext(sc, sqlc, null, dep, printStream,
+    z = new ZeppelinContext(sc, sqlc, null, dep,
         Integer.parseInt(getProperty("zeppelin.spark.maxResult")));
 
     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("out", printStream);
 
     intp.interpret("@transient val z = "
                  + "_binder.get(\"z\").asInstanceOf[org.apache.zeppelin.spark.ZeppelinContext]");
@@ -675,13 +670,13 @@ public InterpreterResult interpret(String[] lines, InterpreterContext context) {
     synchronized (this) {
       z.setGui(context.getGui());
       sc.setJobGroup(getJobGroup(context), "Zeppelin", false);
-      InterpreterResult r = interpretInput(lines);
+      InterpreterResult r = interpretInput(lines, context);
       sc.clearJobGroup();
       return r;
     }
   }
 
-  public InterpreterResult interpretInput(String[] lines) {
+  public InterpreterResult interpretInput(String[] lines, InterpreterContext context) {
     SparkEnv.set(env);
 
     // add print("") to make sure not finishing with comment
@@ -692,8 +687,9 @@ public InterpreterResult interpretInput(String[] lines) {
     }
     linesToRun[lines.length] = "print(\"\")";
 
-    Console.setOut((java.io.PrintStream) binder.get("out"));
-    out.reset();
+    Console.setOut(context.out);
+    out.setInterpreterOutput(context.out);
+    context.out.clear();
     Code r = null;
     String incomplete = "";
 
@@ -713,6 +709,7 @@ public InterpreterResult interpretInput(String[] lines) {
         res = intp.interpret(incomplete + s);
       } catch (Exception e) {
         sc.clearJobGroup();
+        out.setInterpreterOutput(null);
         logger.info("Interpreter exception", e);
         return new InterpreterResult(Code.ERROR, InterpreterUtils.getMostRelevantMessage(e));
       }
@@ -721,7 +718,8 @@ public InterpreterResult interpretInput(String[] lines) {
 
       if (r == Code.ERROR) {
         sc.clearJobGroup();
-        return new InterpreterResult(r, out.toString());
+        out.setInterpreterOutput(null);
+        return new InterpreterResult(r, "");
       } else if (r == Code.INCOMPLETE) {
         incomplete += s + "\n";
       } else {
@@ -730,9 +728,13 @@ public InterpreterResult interpretInput(String[] lines) {
     }
 
     if (r == Code.INCOMPLETE) {
+      sc.clearJobGroup();
+      out.setInterpreterOutput(null);
       return new InterpreterResult(r, "Incomplete expression");
     } else {
-      return new InterpreterResult(r, out.toString());
+      sc.clearJobGroup();
+      out.setInterpreterOutput(null);
+      return new InterpreterResult(Code.SUCCESS);
     }
   }
 

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

@@ -0,0 +1,75 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF 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 org.apache.zeppelin.spark;
+
+import org.apache.zeppelin.interpreter.InterpreterOutput;
+
+import java.io.IOException;
+import java.io.OutputStream;
+
+/**
+ * InterpreterOutput can be attached / detached.
+ */
+public class SparkOutputStream extends OutputStream {
+  InterpreterOutput interpreterOutput;
+
+  public SparkOutputStream() {
+  }
+
+  public InterpreterOutput getInterpreterOutput() {
+    return interpreterOutput;
+  }
+
+  public void setInterpreterOutput(InterpreterOutput interpreterOutput) {
+    this.interpreterOutput = interpreterOutput;
+  }
+
+  @Override
+  public void write(int b) throws IOException {
+    if (interpreterOutput != null) {
+      interpreterOutput.write(b);
+    }
+  }
+
+  @Override
+  public void write(byte [] b) throws IOException {
+    if (interpreterOutput != null) {
+      interpreterOutput.write(b);
+    }
+  }
+
+  @Override
+  public void write(byte [] b, int offset, int len) throws IOException {
+    if (interpreterOutput != null) {
+      interpreterOutput.write(b, offset, len);
+    }
+  }
+
+  @Override
+  public void close() throws IOException {
+    if (interpreterOutput != null) {
+      interpreterOutput.close();
+    }
+  }
+
+  @Override
+  public void flush() throws IOException {
+    if (interpreterOutput != null) {
+      interpreterOutput.flush();
+    }
+  }
+}