render output

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

@@ -28,7 +28,6 @@ import org.apache.zeppelin.scheduler.SchedulerFactory;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import java.io.File;
 import java.io.IOException;
 import java.util.ArrayList;
 import java.util.List;
@@ -52,6 +51,9 @@ public class SparkRInterpreter extends Interpreter {
           .add("zeppelin.R.cmd",
               SparkInterpreter.getSystemDefault("ZEPPELIN_R_CMD", "zeppelin.R.cmd", "R"),
               "R repl path")
+          .add("zeppelin.R.knitr",
+              SparkInterpreter.getSystemDefault("ZEPPELIN_R_KNITR", "zeppelin.R.knitr", "true"),
+              "whether use knitr or not")
           .add("zeppelin.R.image.width",
               SparkInterpreter.getSystemDefault("ZEPPELIN_R_IMAGE_WIDTH",
                   "zeppelin.R.image.width", "100%"),
@@ -90,9 +92,10 @@ public class SparkRInterpreter extends Interpreter {
 
     int port = SparkRBackend.port();
 
-//    SparkInterpreter sparkInterpreter = getSparkInterpreter();
-//    ZeppelinRContext.setSparkContext(sparkInterpreter.getSparkContext());
-//    ZeppelinRContext.setSqlContext(sparkInterpreter.getSQLContext());
+    SparkInterpreter sparkInterpreter = getSparkInterpreter();
+    ZeppelinRContext.setSparkContext(sparkInterpreter.getSparkContext());
+    ZeppelinRContext.setSqlContext(sparkInterpreter.getSQLContext());
+    ZeppelinRContext.setZepplinContext(sparkInterpreter.getZeppelinContext());
 
     zeppelinR = new ZeppelinR(rCmdPath, sparkRLibPath, port);
     try {
@@ -100,6 +103,10 @@ public class SparkRInterpreter extends Interpreter {
     } catch (IOException e) {
       throw new InterpreterException(e);
     }
+
+    if (useKnitr()) {
+      zeppelinR.eval("library('knitr')");
+    }
     renderOptions = getProperty("zeppelin.R.render.options");
   }
 
@@ -125,26 +132,29 @@ public class SparkRInterpreter extends Interpreter {
       }
     }
 
-    interpreterContext.out.clear();
-    zeppelinR.setInterpreterOutput(interpreterContext.out);
+
 
     try {
-/*
-      zeppelinR.set(".zcmd", "\n```{r " + renderOptions + "}\n" + lines + "\n```");
-      zeppelinR.eval(".zres <- knit2html(text=.zcmd)");
-      String html = zeppelinR.getS0(".zres");
+      // render output with knitr
+      if (useKnitr()) {
+        zeppelinR.setInterpreterOutput(null);
+        zeppelinR.set(".zcmd", "\n```{r " + renderOptions + "}\n" + lines + "\n```");
+        zeppelinR.eval(".zres <- knit2html(text=.zcmd)");
+        String html = zeppelinR.getS0(".zres");
 
-      RDisplay rDisplay = render(html, imageWidth);
+        RDisplay rDisplay = render(html, imageWidth);
 
-      return new InterpreterResult(
-              rDisplay.code(),
-              rDisplay.type(),
-              rDisplay.content()
-      );
-*/
-      zeppelinR.eval(lines);
-
-      return new InterpreterResult(InterpreterResult.Code.SUCCESS, "");
+        return new InterpreterResult(
+            rDisplay.code(),
+            rDisplay.type(),
+            rDisplay.content()
+        );
+      } else {
+        // alternatively, stream the output (without knitr)
+        zeppelinR.setInterpreterOutput(interpreterContext.out);
+        zeppelinR.eval(lines);
+        return new InterpreterResult(InterpreterResult.Code.SUCCESS, "");
+      }
     } catch (Exception e) {
       logger.error("Exception while connecting to R", e);
       return new InterpreterResult(InterpreterResult.Code.ERROR, e.getMessage());
@@ -204,4 +214,12 @@ public class SparkRInterpreter extends Interpreter {
     return spark;
   }
 
+  private boolean useKnitr() {
+    try {
+      return Boolean.parseBoolean(getProperty("zeppelin.R.knitr"));
+    } catch (Exception e) {
+      return false;
+    }
+  }
+
 }

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

@@ -156,9 +156,12 @@ public class ZeppelinR implements ExecuteResultHandler {
       public void onUpdate(InterpreterOutput out, byte[] output) {
       }
     });
-    //outputStream.setInterpreterOutput(initialOutput);
+    outputStream.setInterpreterOutput(initialOutput);
     executor.execute(cmd, env, this);
     rScriptRunning = true;
+
+    // flush output
+    eval("cat('')");
   }
 
   /**

spark/src/main/resources/R/zeppelin_sparkr.R

@@ -1,3 +1,21 @@
+#
+# 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.
+#
+
 args <- commandArgs(trailingOnly = TRUE)
 
 hashCode <- as.integer(args[1])
@@ -37,8 +55,6 @@ z.input <- function(name, value) {
   SparkR:::callJMethod(.zeppelinContext, "input", name, value)
 }
 
-library("knitr")
-
 # notify script is initialized
 SparkR:::callJMethod(.zeppelinR, "onScriptInitialized")