KotlinSparkInterpreter extends Interpreter { @Override public InterpreterResult interpret(String st, InterpreterContext context) throws InterpreterException { if (isSparkVersionUnsupported()) { return unsupportedMessage; } z.setInterpreterContext(context); z.setGui(context.getGui()); z.setNoteGui(context.getNoteGui()); InterpreterContext.set(context); jsc.setJobGroup(buildJobGroupId(context), buildJobDesc(context), false); jsc.setLocalProperty("spark.scheduler.pool", context.getLocalProperties().get("pool")); InterpreterOutput out = context.out; PrintStream scalaOut = Console.out(); PrintStream newOut = (out != null) ? new PrintStream(out) : null; Console.setOut(newOut); InterpreterResult result = interpreter.interpret(st, context); Console.setOut(scalaOut); return result; } KotlinSparkInterpreter(Properties properties); @Override void open(); @Override void close(); @Override InterpreterResult interpret(String st, InterpreterContext context); @Override void cancel(InterpreterContext context); @Override FormType getFormType(); @Override int getProgress(InterpreterContext context); @Override List<InterpreterCompletion> completion(String buf, int cursor,
      InterpreterContext interpreterContext);  }
@Test public void dataFrameTest() throws Exception { interpreter.interpret("spark.range(100, 0, -1).sort(\"id\").show(2)", context); assertTrue(output.contains( "+---+\n" + "| id|\n" + "+---+\n" + "| 1|\n" + "| 2|\n" + "+---+")); }
@Test public void sparkPropertiesTest() throws Exception { InterpreterResult result = interpreter.interpret( "sc.conf.all.map{ it.toString() }", context); String message = result.message().get(0).getData().trim(); System.out.println("PROPS_1 = " + message); assertTrue(message.contains("(zeppelin.spark.property_1,value_1)")); }
@Test public void classWriteTest() throws Exception { interpreter.interpret("val f = { x: Any -> println(x) }", context); output = ""; InterpreterResult result = interpreter.interpret("spark.range(5).foreach(f)", context); assertEquals(SUCCESS, result.code()); assertTrue(output.contains("0")); assertTrue(output.contains("1")); assertTrue(output.contains("2")); assertTrue(output.contains("3")); assertTrue(output.contains("4")); String classOutputDir = repl.getSparkContext().getConf().get("spark.repl.class.outputDir"); System.out.println(classOutputDir); Path outPath = Paths.get(classOutputDir); Files.walk(outPath).forEach(System.out::println); assertTrue(Files.walk(outPath).anyMatch(path -> path.toString().matches( ".*Line_\\d+\\$f\\$1\\.class"))); assertTrue(Files.walk(outPath).anyMatch(path -> path.toString().matches( ".*Line_\\d+\\$sam\\$org_apache_spark_api_java_function_ForeachFunction\\$0\\.class"))); }
@Test public void zeppelinContextTest() throws Exception { InterpreterResult 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()); }