SparkRInterpreter extends RInterpreter { public SparkRInterpreter(Properties property) { super(property); } SparkRInterpreter(Properties property); @Override void open(); @Override InterpreterResult internalInterpret(String lines, InterpreterContext interpreterContext); @Override void close(); @Override void cancel(InterpreterContext context); @Override FormType getFormType(); @Override int getProgress(InterpreterContext context); @Override Scheduler getScheduler(); @Override ZeppelinContext getZeppelinContext(); @Override List<InterpreterCompletion> completion(String buf, int cursor,
                                                InterpreterContext interpreterContext); boolean isSpark1();  }
@Test public void testSparkRInterpreter() throws InterpreterException, InterruptedException { InterpreterResult result = sparkRInterpreter.interpret("1+1", getInterpreterContext()); assertEquals(InterpreterResult.Code.SUCCESS, result.code()); assertTrue(result.message().get(0).getData().contains("2")); result = sparkRInterpreter.interpret("sparkR.version()", getInterpreterContext()); assertEquals(InterpreterResult.Code.SUCCESS, result.code()); if (sparkRInterpreter.isSpark1()) { result = sparkRInterpreter.interpret("df <- createDataFrame(sqlContext, faithful)\nhead(df)", getInterpreterContext()); assertEquals(InterpreterResult.Code.SUCCESS, result.code()); assertTrue(result.message().get(0).getData().contains("eruptions waiting")); verify(mockRemoteIntpEventClient, atLeastOnce()).onParaInfosReceived(any(Map.class)); } else { result = sparkRInterpreter.interpret("df <- as.DataFrame(faithful)\nhead(df)", getInterpreterContext()); assertEquals(InterpreterResult.Code.SUCCESS, result.code()); assertTrue(result.message().get(0).getData().contains("eruptions waiting")); verify(mockRemoteIntpEventClient, atLeastOnce()).onParaInfosReceived(any(Map.class)); final InterpreterContext context = getInterpreterContext(); Thread thread = new Thread() { @Override public void run() { try { InterpreterResult result = sparkRInterpreter.interpret("ldf <- dapplyCollect(\n" + " df,\n" + " function(x) {\n" + " Sys.sleep(3)\n" + " x <- cbind(x, \"waiting_secs\" = x$waiting * 60)\n" + " })\n" + "head(ldf, 3)", context); assertTrue(result.message().get(0).getData().contains("cancelled")); } catch (InterpreterException e) { fail("Should not throw InterpreterException"); } } }; thread.setName("Cancel-Thread"); thread.start(); Thread.sleep(1000); sparkRInterpreter.cancel(context); } InterpreterContext context = getInterpreterContext(); context.getLocalProperties().put("imageWidth", "100"); result = sparkRInterpreter.interpret("hist(mtcars$mpg)", context); assertEquals(InterpreterResult.Code.SUCCESS, result.code()); assertEquals(1, result.message().size()); assertEquals(InterpreterResult.Type.HTML, result.message().get(0).getType()); assertTrue(result.message().get(0).getData().contains("<img src=")); assertTrue(result.message().get(0).getData().contains("width=\"100\"")); result = sparkRInterpreter.interpret("library(ggplot2)\n" + "ggplot(diamonds, aes(x=carat, y=price, color=cut)) + geom_point()", getInterpreterContext()); assertEquals(InterpreterResult.Code.SUCCESS, result.code()); assertEquals(1, result.message().size()); assertEquals(InterpreterResult.Type.HTML, result.message().get(0).getType()); assertTrue(result.message().get(0).getData().contains("<img src=")); Thread.sleep(15 * 1000); result = sparkRInterpreter.interpret("1+1", getInterpreterContext()); assertEquals(InterpreterResult.Code.ERROR, result.code()); assertTrue(result.message().get(0).getData().contains("sparkR backend is dead")); }