SparkInterpreter extends AbstractInterpreter { public SparkInterpreter(Properties properties) { properties = mergeZeppelinInterpreterProperties(properties); this.zeppelinInterpreter = new org.apache.zeppelin.spark.SparkInterpreter(properties); this.setInterpreterGroup(new InterpreterGroup()); } SparkInterpreter(Properties properties);  SparkInterpreter(); @Override boolean test(); SparkContext getSparkContext(); void setSchedulerPool(String pool);  }
@Test public void testSparkInterpreter() throws InterruptedException, InterpreterException { Properties properties = new Properties(); properties.setProperty("spark.master", "local"); properties.setProperty("spark.app.name", "test"); properties.setProperty("submarine.spark.scala.color", "false"); properties.setProperty("submarine.spark.test", "true"); properties.setProperty("submarine.spark.uiWebUrl", "fake_spark_weburl"); properties.setProperty("submarine.spark.deprecatedMsg.show", "false"); interpreter = new SparkInterpreter(properties); interpreter.open(); InterpreterResult result = interpreter.interpret("val a=\"hello world\""); assertEquals(InterpreterResult.Code.SUCCESS, result.code()); assertEquals("a: String = hello world\n", result.message().get(0).getData()); result = interpreter.interpret("print(a)"); assertEquals(InterpreterResult.Code.SUCCESS, result.code()); assertEquals("hello world", result.message().get(0).getData()); result = interpreter.interpret("System.out.print(a)"); assertEquals(InterpreterResult.Code.SUCCESS, result.code()); assertEquals("hello world", result.message().get(0).getData()); result = interpreter.interpret("println(a"); assertEquals(InterpreterResult.Code.INCOMPLETE, result.code()); result = interpreter.interpret("println(b)"); assertEquals(InterpreterResult.Code.ERROR, result.code()); assertTrue(result.message().get(0).getData().contains("not found: value b")); result = interpreter.interpret("\"123\".\ntoInt"); assertEquals(InterpreterResult.Code.SUCCESS, result.code()); result = interpreter.interpret("print(\"hello world\")"); assertEquals(InterpreterResult.Code.SUCCESS, result.code()); assertEquals("hello world", result.message().get(0).getData()); result = interpreter.interpret("\nprint(\"hello world\")"); assertEquals(InterpreterResult.Code.SUCCESS, result.code()); result = interpreter.interpret(""); assertEquals(InterpreterResult.Code.SUCCESS, result.code()); result = interpreter.interpret("def add(x:Int, y:Int)\n{ return x+y }"); assertEquals(InterpreterResult.Code.SUCCESS, result.code()); result = interpreter.interpret("print(add(1,2))"); assertEquals(InterpreterResult.Code.SUCCESS, result.code()); result = interpreter.interpret("print(\"hello world\")"); 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)"); assertEquals(InterpreterResult.Code.SUCCESS, result.code()); result = interpreter.interpret("import java.util.ArrayList"); assertEquals(InterpreterResult.Code.SUCCESS, result.code()); result = interpreter.interpret("sc\n.range(1, 10)\n.sum"); assertEquals(InterpreterResult.Code.SUCCESS, result.code()); assertTrue(result.message().get(0).getData().contains("45")); result = interpreter.interpret("sc\n.range(1, 10)\n.sum"); assertEquals(InterpreterResult.Code.SUCCESS, result.code()); assertTrue(result.message().get(0).getData().contains("45")); result = interpreter.interpret("val bankText = sc.textFile(\"bank.csv\")"); assertEquals(InterpreterResult.Code.SUCCESS, result.code()); result = interpreter.interpret( "case class Bank(age:Integer, job:String, marital : String, edu : 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()"); assertEquals(InterpreterResult.Code.SUCCESS, result.code()); result = interpreter.interpret("sc.version"); assertEquals(InterpreterResult.Code.SUCCESS, result.code()); String version = result.message().get(0).getData().trim(); if (version.contains("String = 1.")) { result = interpreter.interpret("sqlContext"); assertEquals(InterpreterResult.Code.SUCCESS, result.code()); result = interpreter.interpret( "val df = sqlContext.createDataFrame(Seq((1,\"a\"),(2, null)))\n" + "df.show()"); assertEquals(InterpreterResult.Code.SUCCESS, result.code()); assertTrue(result.message().get(0).getData().contains( "+---+----+\n" + "| _1| _2|\n" + "+---+----+\n" + "| 1| a|\n" + "| 2|null|\n" + "+---+----+")); } else if (version.contains("String = 2.")) { result = interpreter.interpret("spark"); assertEquals(InterpreterResult.Code.SUCCESS, result.code()); result = interpreter.interpret( "val df = spark.createDataFrame(Seq((1,\"a\"),(2, null)))\n" + "df.show()"); assertEquals(InterpreterResult.Code.SUCCESS, result.code()); assertTrue(result.message().get(0).getData().contains( "+---+----+\n" + "| _1| _2|\n" + "+---+----+\n" + "| 1| a|\n" + "| 2|null|\n" + "+---+----+")); } result = interpreter.interpret("z.show(df)"); assertEquals(InterpreterResult.Code.SUCCESS, result.code()); assertEquals(InterpreterResult.Type.TABLE, result.message().get(0).getType()); assertEquals("_1\t_2\n1\ta\n2\tnull\n", result.message().get(0).getData()); result = interpreter.interpret("z.input(\"name\", \"default_name\")"); assertEquals(InterpreterResult.Code.SUCCESS, result.code()); Thread interpretThread = new Thread(() -> { try { InterpreterResult result1 = interpreter.interpret( "val df = sc.parallelize(1 to 10, 5).foreach(e=>Thread.sleep(1000))"); assertEquals(InterpreterResult.Code.SUCCESS, result1.code()); } catch (InterpreterException e) { e.printStackTrace(); fail(); } }); interpretThread.start(); boolean nonZeroProgress = false; int progress = 0; while (interpretThread.isAlive()) { progress = interpreter.getProgress(); assertTrue(progress >= 0); if (progress != 0 && progress != 100) { nonZeroProgress = true; } Thread.sleep(100); } assertTrue(nonZeroProgress); interpretThread = new Thread(() -> { try { InterpreterResult result12 = interpreter.interpret( "val df = sc.parallelize(1 to 10, 2).foreach(e=>Thread.sleep(1000))"); assertEquals(InterpreterResult.Code.ERROR, result12.code()); assertTrue(result12.message().get(0).getData().contains("cancelled")); } catch (org.apache.submarine.interpreter.InterpreterException e) { e.printStackTrace(); fail(); } }); interpretThread.start(); Thread.sleep(1000); interpreter.cancel(); interpretThread.join(); }