SparkInterpreter extends AbstractInterpreter { @Override public void open() throws InterpreterException { try { SparkConf conf = new SparkConf(); for (Map.Entry<Object, Object> entry : getProperties().entrySet()) { if (!StringUtils.isBlank(entry.getValue().toString())) { conf.set(entry.getKey().toString(), entry.getValue().toString()); } if (entry.getKey().toString().equals("zeppelin.spark.useHiveContext")) { conf.set("spark.useHiveContext", entry.getValue().toString()); } if (entry.getKey().toString().equals("zeppelin.spark.concurrentSQL") && entry.getValue().toString().equals("true")) { conf.set(SparkStringConstants.SCHEDULER_MODE_PROP_NAME, "FAIR"); } } if (!conf.contains(SparkStringConstants.MASTER_PROP_NAME)) { if (conf.contains("master")) { conf.set(SparkStringConstants.MASTER_PROP_NAME, conf.get("master")); } else { String masterEnv = System.getenv(SparkStringConstants.MASTER_ENV_NAME); conf.set(SparkStringConstants.MASTER_PROP_NAME, masterEnv == null ? SparkStringConstants.DEFAULT_MASTER_VALUE : masterEnv); } } this.innerInterpreter = loadSparkScalaInterpreter(conf); this.innerInterpreter.open(); sc = this.innerInterpreter.getSparkContext(); jsc = JavaSparkContext.fromSparkContext(sc); sparkVersion = SparkVersion.fromVersionString(sc.version()); if (enableSupportedVersionCheck && sparkVersion.isUnsupportedVersion()) { throw new Exception("This is not officially supported spark version: " + sparkVersion + "\nYou can set zeppelin.spark.enableSupportedVersionCheck to false if you really" + " want to try this version of spark."); } sqlContext = this.innerInterpreter.getSqlContext(); sparkSession = this.innerInterpreter.getSparkSession(); SESSION_NUM.incrementAndGet(); } catch (Exception e) { LOGGER.error("Fail to open SparkInterpreter", e); throw new InterpreterException("Fail to open SparkInterpreter", e); } } SparkInterpreter(Properties properties); @Override void open(); @Override void close(); @Override InterpreterResult internalInterpret(String st,
                                             InterpreterContext context); @Override void cancel(InterpreterContext context); @Override List<InterpreterCompletion> completion(String buf,
                                                int cursor,
                                                InterpreterContext interpreterContext); @Override FormType getFormType(); @Override int getProgress(InterpreterContext context); ZeppelinContext getZeppelinContext(); SparkContext getSparkContext(); Object getSQLContext(); JavaSparkContext getJavaSparkContext(); Object getSparkSession(); SparkVersion getSparkVersion(); boolean isScala212(); ClassLoader getScalaShellClassLoader(); boolean isUnsupportedSparkVersion();  }
@Test public void testDisableReplOutput() throws InterpreterException { Properties properties = new Properties(); properties.setProperty(SparkStringConstants.MASTER_PROP_NAME, "local"); properties.setProperty(SparkStringConstants.APP_NAME_PROP_NAME, "test"); properties.setProperty("zeppelin.spark.maxResult", "100"); properties.setProperty("zeppelin.spark.printREPLOutput", "false"); properties.setProperty("zeppelin.spark.scala.color", "false"); properties.setProperty("zeppelin.spark.deprecatedMsg.show", "false"); InterpreterContext.set(getInterpreterContext()); interpreter = new SparkInterpreter(properties); interpreter.setInterpreterGroup(mock(InterpreterGroup.class)); interpreter.open(); InterpreterResult result = interpreter.interpret("val a=\"hello world\"", getInterpreterContext()); assertEquals(InterpreterResult.Code.SUCCESS, result.code()); assertEquals("", output); result = interpreter.interpret("print(a)", getInterpreterContext()); assertEquals(InterpreterResult.Code.SUCCESS, result.code()); assertEquals("hello world", output); }
@Test public void testDisableReplOutputForParagraph() throws InterpreterException { Properties properties = new Properties(); properties.setProperty("spark.master", "local"); properties.setProperty("spark.app.name", "test"); properties.setProperty("zeppelin.spark.maxResult", "100"); properties.setProperty("zeppelin.spark.printREPLOutput", "true"); properties.setProperty("zeppelin.spark.scala.color", "false"); properties.setProperty("zeppelin.spark.deprecatedMsg.show", "false"); InterpreterContext.set(getInterpreterContext()); interpreter = new SparkInterpreter(properties); interpreter.setInterpreterGroup(mock(InterpreterGroup.class)); interpreter.open(); InterpreterResult result = interpreter.interpret("val a=\"hello world\"", getInterpreterContext()); assertEquals(InterpreterResult.Code.SUCCESS, result.code()); assertEquals("a: String = hello world\n", output); result = interpreter.interpret("print(a)", getInterpreterContext()); assertEquals(InterpreterResult.Code.SUCCESS, result.code()); assertEquals("hello world", output); InterpreterContext context = getInterpreterContext(); context.getLocalProperties().put("printREPLOutput", "false"); result = interpreter.interpret("print(a)", context); assertEquals(InterpreterResult.Code.SUCCESS, result.code()); assertEquals("", output); result = interpreter.interpret("val a=\"hello world\"", getInterpreterContext()); assertEquals(InterpreterResult.Code.SUCCESS, result.code()); assertEquals("a: String = hello world\n", output); result = interpreter.interpret("print(a)", getInterpreterContext()); assertEquals(InterpreterResult.Code.SUCCESS, result.code()); assertEquals("hello world", output); }
@Test public void testDisableSparkUI_1() throws InterpreterException { Properties properties = new Properties(); properties.setProperty(SparkStringConstants.MASTER_PROP_NAME, "local"); properties.setProperty(SparkStringConstants.APP_NAME_PROP_NAME, "test"); properties.setProperty("zeppelin.spark.maxResult", "100"); properties.setProperty("spark.ui.enabled", "false"); properties.setProperty("zeppelin.spark.scala.color", "false"); properties.setProperty("zeppelin.spark.deprecatedMsg.show", "false"); interpreter = new SparkInterpreter(properties); interpreter.setInterpreterGroup(mock(InterpreterGroup.class)); InterpreterContext.set(getInterpreterContext()); interpreter.open(); InterpreterContext context = getInterpreterContext(); InterpreterResult result = interpreter.interpret("sc.range(1, 10).sum", context); assertEquals(InterpreterResult.Code.SUCCESS, result.code()); verify(mockRemoteEventClient, never()).onParaInfosReceived(any(Map.class)); }
@Test public void testDisableSparkUI_2() throws InterpreterException { Properties properties = new Properties(); properties.setProperty(SparkStringConstants.MASTER_PROP_NAME, "local"); properties.setProperty(SparkStringConstants.APP_NAME_PROP_NAME, "test"); properties.setProperty("zeppelin.spark.maxResult", "100"); properties.setProperty("zeppelin.spark.ui.hidden", "true"); properties.setProperty("zeppelin.spark.scala.color", "false"); properties.setProperty("zeppelin.spark.deprecatedMsg.show", "false"); interpreter = new SparkInterpreter(properties); interpreter.setInterpreterGroup(mock(InterpreterGroup.class)); InterpreterContext.set(getInterpreterContext()); interpreter.open(); InterpreterContext context = getInterpreterContext(); InterpreterResult result = interpreter.interpret("sc.range(1, 10).sum", context); assertEquals(InterpreterResult.Code.SUCCESS, result.code()); verify(mockRemoteEventClient, never()).onParaInfosReceived(any(Map.class)); }