SparkInterpreter extends AbstractInterpreter { public SparkInterpreter(Properties properties) { super(properties); if (Boolean.parseBoolean(properties.getProperty("zeppelin.spark.scala.color", "true"))) { System.setProperty("scala.color", "true"); } this.enableSupportedVersionCheck = java.lang.Boolean.parseBoolean( properties.getProperty("zeppelin.spark.enableSupportedVersionCheck", "true")); innerInterpreterClassMap.put("2.10", "org.apache.zeppelin.spark.SparkScala210Interpreter"); innerInterpreterClassMap.put("2.11", "org.apache.zeppelin.spark.SparkScala211Interpreter"); innerInterpreterClassMap.put("2.12", "org.apache.zeppelin.spark.SparkScala212Interpreter"); } 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 testSparkInterpreter() throws IOException, InterruptedException, 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.uiWebUrl", "fake_spark_weburl/{{applicationId}}"); properties.setProperty("zeppelin.spark.scala.color", "false"); properties.setProperty("zeppelin.spark.deprecatedMsg.show", "false"); InterpreterContext context = InterpreterContext.builder() .setInterpreterOut(new InterpreterOutput(null)) .setIntpEventClient(mockRemoteEventClient) .setAngularObjectRegistry(new AngularObjectRegistry("spark", null)) .build(); InterpreterContext.set(context); 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); result = interpreter.interpret("System.out.print(a)", getInterpreterContext()); assertEquals(InterpreterResult.Code.SUCCESS, result.code()); assertEquals("hello world", output); result = interpreter.interpret("println(a", getInterpreterContext()); assertEquals(InterpreterResult.Code.INCOMPLETE, result.code()); result = interpreter.interpret("println(b)", getInterpreterContext()); assertEquals(InterpreterResult.Code.ERROR, result.code()); assertTrue(output.contains("not found: value b")); result = interpreter.interpret("\"123\".\ntoInt", getInterpreterContext()); assertEquals(InterpreterResult.Code.SUCCESS, result.code()); result = interpreter.interpret("print(\"hello world\")", getInterpreterContext()); assertEquals(InterpreterResult.Code.SUCCESS, result.code()); assertEquals("hello world", output); result = interpreter.interpret("\nprint(\"hello world\")", getInterpreterContext()); assertEquals(InterpreterResult.Code.SUCCESS, result.code()); result = interpreter.interpret("", getInterpreterContext()); assertEquals(InterpreterResult.Code.SUCCESS, result.code()); result = interpreter.interpret("def add(x:Int, y:Int)\n{ return x+y }", getInterpreterContext()); assertEquals(InterpreterResult.Code.SUCCESS, result.code()); result = interpreter.interpret("print(add(1,2))", getInterpreterContext()); assertEquals(InterpreterResult.Code.SUCCESS, result.code()); result = interpreter.interpret("print(\"hello world\")", getInterpreterContext()); assertEquals(InterpreterResult.Code.SUCCESS, result.code()); result = interpreter.interpret("import scala.math._\n" + "object Circle {\n" + " private def calculateArea(radius: Double): Double = Pi * pow(radius, 2.0)\n" + "}\n" + "case class Circle(radius: Double) {\n" + " import Circle._\n" + " def area: Double = calculateArea(radius)\n" + "}\n" + "\n" + "val circle1 = new Circle(5.0)", getInterpreterContext()); assertEquals(InterpreterResult.Code.SUCCESS, result.code()); result = interpreter.interpret("import java.util.ArrayList", getInterpreterContext()); assertEquals(InterpreterResult.Code.SUCCESS, result.code()); result = interpreter.interpret("class MyArrayList extends ArrayList{}", getInterpreterContext()); assertEquals(InterpreterResult.Code.SUCCESS, result.code()); context = getInterpreterContext(); context.setParagraphId("pid_1"); result = interpreter.interpret("sc\n.range(1, 10)\n.sum", context); assertEquals(InterpreterResult.Code.SUCCESS, result.code()); assertTrue(output.contains("45")); ArgumentCaptor<Map> captorEvent = ArgumentCaptor.forClass(Map.class); verify(mockRemoteEventClient).onParaInfosReceived(captorEvent.capture()); assertEquals("pid_1", captorEvent.getValue().get("paraId")); reset(mockRemoteEventClient); context = getInterpreterContext(); context.setParagraphId("pid_2"); result = interpreter.interpret("sc\n.range(1, 10)\n.sum", context); assertEquals(InterpreterResult.Code.SUCCESS, result.code()); assertTrue(output.contains("45")); captorEvent = ArgumentCaptor.forClass(Map.class); verify(mockRemoteEventClient).onParaInfosReceived(captorEvent.capture()); assertEquals("pid_2", captorEvent.getValue().get("paraId")); ArgumentCaptor<Map> onParaInfosReceivedArg = ArgumentCaptor.forClass(Map.class); verify(mockRemoteEventClient).onParaInfosReceived(onParaInfosReceivedArg.capture()); assertTrue(((String) onParaInfosReceivedArg.getValue().get("jobUrl")).startsWith("fake_spark_weburl/" + interpreter.getJavaSparkContext().sc().applicationId())); result = interpreter.interpret("val bankText = sc.textFile(\"bank.csv\")", getInterpreterContext()); assertEquals(InterpreterResult.Code.SUCCESS, result.code()); result = interpreter.interpret( "case class Bank(age:Integer, job:String, marital : String, education : String, balance : Integer)\n" + "val bank = bankText.map(s=>s.split(\";\")).filter(s => s(0)!=\"\\\"age\\\"\").map(\n" + " s => Bank(s(0).toInt, \n" + " s(1).replaceAll(\"\\\"\", \"\"),\n" + " s(2).replaceAll(\"\\\"\", \"\"),\n" + " s(3).replaceAll(\"\\\"\", \"\"),\n" + " s(5).replaceAll(\"\\\"\", \"\").toInt\n" + " )\n" + ").toDF()", getInterpreterContext()); assertEquals(InterpreterResult.Code.SUCCESS, result.code()); result = interpreter.interpret("sc.version", getInterpreterContext()); assertEquals(InterpreterResult.Code.SUCCESS, result.code()); String version = output.trim(); if (version.contains("String = 1.")) { result = interpreter.interpret("sqlContext", getInterpreterContext()); assertEquals(InterpreterResult.Code.SUCCESS, result.code()); result = interpreter.interpret( "val df = sqlContext.createDataFrame(Seq((1,\"a\"),(2, null)))\n" + "df.show()", getInterpreterContext()); assertEquals(InterpreterResult.Code.SUCCESS, result.code()); assertTrue(output.contains( "+---+----+\n" + "| _1| _2|\n" + "+---+----+\n" + "| 1| a|\n" + "| 2|null|\n" + "+---+----+")); } else if (version.contains("String = 2.")) { context = getInterpreterContext(); result = interpreter.interpret("case class Person(id:Int, name:String, age:Int, country:String)\n" + "val df2 = spark.createDataFrame(Seq(Person(1, \"andy\", 20, \"USA\"), " + "Person(2, \"jeff\", 23, \"China\"), Person(3, \"james\", 18, \"USA\")))\n" + "df2.printSchema\n" + "df2.show() ", context); assertEquals(InterpreterResult.Code.SUCCESS, result.code()); result = interpreter.interpret("spark", getInterpreterContext()); assertEquals(InterpreterResult.Code.SUCCESS, result.code()); result = interpreter.interpret( "val df = spark.createDataFrame(Seq((1,\"a\"),(2, null)))\n" + "df.show()", getInterpreterContext()); assertEquals(InterpreterResult.Code.SUCCESS, result.code()); assertTrue(output.contains( "+---+----+\n" + "| _1| _2|\n" + "+---+----+\n" + "| 1| a|\n" + "| 2|null|\n" + "+---+----+")); } result = interpreter.interpret("z.show(df)", getInterpreterContext()); assertEquals(InterpreterResult.Code.SUCCESS, result.code()); assertEquals(InterpreterResult.Type.TABLE, messageOutput.getType()); messageOutput.flush(); assertEquals("_1\t_2\n1\ta\n2\tnull\n", messageOutput.toInterpreterResultMessage().getData()); context = getInterpreterContext(); result = interpreter.interpret("z.input(\"name\", \"default_name\")", context); assertEquals(InterpreterResult.Code.SUCCESS, result.code()); assertEquals(1, context.getGui().getForms().size()); assertTrue(context.getGui().getForms().get("name") instanceof TextBox); TextBox textBox = (TextBox) context.getGui().getForms().get("name"); assertEquals("name", textBox.getName()); assertEquals("default_name", textBox.getDefaultValue()); context = getInterpreterContext(); result = interpreter.interpret("z.password(\"pwd\")", context); assertEquals(InterpreterResult.Code.SUCCESS, result.code()); assertEquals(1, context.getGui().getForms().size()); assertTrue(context.getGui().getForms().get("pwd") instanceof Password); Password pwd = (Password) context.getGui().getForms().get("pwd"); assertEquals("pwd", pwd.getName()); context = getInterpreterContext(); result = interpreter.interpret("z.checkbox(\"checkbox_1\", Seq((\"value_1\", \"name_1\"), (\"value_2\", \"name_2\")), Seq(\"value_2\"))", context); assertEquals(InterpreterResult.Code.SUCCESS, result.code()); assertEquals(1, context.getGui().getForms().size()); assertTrue(context.getGui().getForms().get("checkbox_1") instanceof CheckBox); CheckBox checkBox = (CheckBox) context.getGui().getForms().get("checkbox_1"); assertEquals("checkbox_1", checkBox.getName()); assertEquals(1, checkBox.getDefaultValue().length); assertEquals("value_2", checkBox.getDefaultValue()[0]); assertEquals(2, checkBox.getOptions().length); assertEquals("value_1", checkBox.getOptions()[0].getValue()); assertEquals("name_1", checkBox.getOptions()[0].getDisplayName()); assertEquals("value_2", checkBox.getOptions()[1].getValue()); assertEquals("name_2", checkBox.getOptions()[1].getDisplayName()); context = getInterpreterContext(); result = interpreter.interpret("z.select(\"select_1\", Seq((\"value_1\", \"name_1\"), (\"value_2\", \"name_2\")), Seq(\"value_2\"))", context); assertEquals(InterpreterResult.Code.SUCCESS, result.code()); assertEquals(1, context.getGui().getForms().size()); assertTrue(context.getGui().getForms().get("select_1") instanceof Select); Select select = (Select) context.getGui().getForms().get("select_1"); assertEquals("select_1", select.getName()); assertEquals(2, select.getOptions().length); assertEquals("value_1", select.getOptions()[0].getValue()); assertEquals("name_1", select.getOptions()[0].getDisplayName()); assertEquals("value_2", select.getOptions()[1].getValue()); assertEquals("name_2", select.getOptions()[1].getDisplayName()); List<InterpreterCompletion> completions = interpreter.completion("a.", 2, getInterpreterContext()); assertTrue(completions.size() > 0); completions = interpreter.completion("a.isEm", 6, getInterpreterContext()); assertEquals(1, completions.size()); assertEquals("isEmpty", completions.get(0).name); completions = interpreter.completion("sc.ra", 5, getInterpreterContext()); assertEquals(1, completions.size()); assertEquals("range", completions.get(0).name); completions = interpreter.completion("sc.ra\n1+1", 5, getInterpreterContext()); assertEquals(1, completions.size()); assertEquals("range", completions.get(0).name); result = interpreter.interpret("import org.apache.zeppelin.display.angular.notebookscope._\n" + "import AngularElem._", getInterpreterContext()); assertEquals(InterpreterResult.Code.SUCCESS, result.code()); result = interpreter.interpret("<div style=\"color:blue\">\n" + "<h4>Hello Angular Display System</h4>\n" + "</div>.display", getInterpreterContext()); assertEquals(InterpreterResult.Code.SUCCESS, result.code()); assertEquals(InterpreterResult.Type.ANGULAR, messageOutput.getType()); assertTrue(messageOutput.toInterpreterResultMessage().getData().contains("Hello Angular Display System")); result = interpreter.interpret("<div class=\"btn btn-success\">\n" + " Click me\n" + "</div>.onClick{() =>\n" + " println(\"hello world\")\n" + "}.display", getInterpreterContext()); assertEquals(InterpreterResult.Code.SUCCESS, result.code()); assertEquals(InterpreterResult.Type.ANGULAR, messageOutput.getType()); assertTrue(messageOutput.toInterpreterResultMessage().getData().contains("Click me")); final InterpreterContext context2 = getInterpreterContext(); Thread interpretThread = new Thread() { @Override public void run() { InterpreterResult result = null; try { result = interpreter.interpret( "val df = sc.parallelize(1 to 10, 5).foreach(e=>Thread.sleep(1000))", context2); } catch (InterpreterException e) { e.printStackTrace(); } assertEquals(InterpreterResult.Code.SUCCESS, result.code()); } }; interpretThread.start(); boolean nonZeroProgress = false; int progress = 0; while (interpretThread.isAlive()) { progress = interpreter.getProgress(context2); assertTrue(progress >= 0); if (progress != 0 && progress != 100) { nonZeroProgress = true; } Thread.sleep(100); } assertTrue(nonZeroProgress); final InterpreterContext context3 = getInterpreterContext(); interpretThread = new Thread() { @Override public void run() { InterpreterResult result = null; try { result = interpreter.interpret( "val df = sc.parallelize(1 to 10, 2).foreach(e=>Thread.sleep(1000))", context3); } catch (InterpreterException e) { e.printStackTrace(); } assertEquals(InterpreterResult.Code.ERROR, result.code()); assertTrue(output.contains("cancelled")); } }; interpretThread.start(); Thread.sleep(1000); interpreter.cancel(context3); interpretThread.join(); }