@Test public void test() { repl.interpret("case class Test(name:String, age:Int)", context); repl.interpret("val test = sc.parallelize(Seq(Test(\"moon\", 33), Test(\"jobs\", 51), Test(\"gates\", 51), Test(\"park\", 34)))", context); if (isDataFrameSupported()) { repl.interpret("test.toDF.registerTempTable(\"test\")", context); } else { repl.interpret("test.registerTempTable(\"test\")", context); } InterpreterResult ret = sql.interpret("select name, age from test where age < 40", context); assertEquals(InterpreterResult.Code.SUCCESS, ret.code()); assertEquals(Type.TABLE, ret.type()); assertEquals("name\tage\nmoon\t33\npark\t34\n", ret.message()); try { sql.interpret("select wrong syntax", context); fail("Exception not catched"); } catch (Exception e) { } assertEquals(InterpreterResult.Code.SUCCESS, sql.interpret("select case when name==\"aa\" then name else name end from test", context).code()); }
@Override public InterpreterResult interpret(String st, InterpreterContext context) { SQLContext sqlc = null; SparkInterpreter sparkInterpreter = getSparkInterpreter(); if (sparkInterpreter.getSparkVersion().isUnsupportedVersion()) { return new InterpreterResult(Code.ERROR, "Spark " + sparkInterpreter.getSparkVersion().toString() + " is not supported"); } sqlc = getSparkInterpreter().getSQLContext(); SparkContext sc = sqlc.sparkContext(); if (concurrentSQL()) { sc.setLocalProperty("spark.scheduler.pool", "fair"); } else { sc.setLocalProperty("spark.scheduler.pool", null); } sc.setJobGroup(getJobGroup(context), "Zeppelin", false); Object rdd = null; try { Method sqlMethod = sqlc.getClass().getMethod("sql", String.class); rdd = sqlMethod.invoke(sqlc, st); } catch (NoSuchMethodException | SecurityException | IllegalAccessException | IllegalArgumentException | InvocationTargetException e) { throw new InterpreterException(e); } String msg = ZeppelinContext.showDF(sc, context, rdd, maxResult); sc.clearJobGroup(); return new InterpreterResult(Code.SUCCESS, msg); }
SparkSqlInterpreter extends Interpreter { @Override public InterpreterResult interpret(String st, InterpreterContext context) { SQLContext sqlc = null; SparkInterpreter sparkInterpreter = getSparkInterpreter(); if (sparkInterpreter.getSparkVersion().isUnsupportedVersion()) { return new InterpreterResult(Code.ERROR, "Spark " + sparkInterpreter.getSparkVersion().toString() + " is not supported"); } sqlc = getSparkInterpreter().getSQLContext(); SparkContext sc = sqlc.sparkContext(); if (concurrentSQL()) { sc.setLocalProperty("spark.scheduler.pool", "fair"); } else { sc.setLocalProperty("spark.scheduler.pool", null); } sc.setJobGroup(getJobGroup(context), "Zeppelin", false); Object rdd = null; try { Method sqlMethod = sqlc.getClass().getMethod("sql", String.class); rdd = sqlMethod.invoke(sqlc, st); } catch (NoSuchMethodException | SecurityException | IllegalAccessException | IllegalArgumentException | InvocationTargetException e) { throw new InterpreterException(e); } String msg = ZeppelinContext.showDF(sc, context, rdd, maxResult); sc.clearJobGroup(); return new InterpreterResult(Code.SUCCESS, msg); } }
SparkSqlInterpreter extends Interpreter { @Override public InterpreterResult interpret(String st, InterpreterContext context) { SQLContext sqlc = null; SparkInterpreter sparkInterpreter = getSparkInterpreter(); if (sparkInterpreter.getSparkVersion().isUnsupportedVersion()) { return new InterpreterResult(Code.ERROR, "Spark " + sparkInterpreter.getSparkVersion().toString() + " is not supported"); } sqlc = getSparkInterpreter().getSQLContext(); SparkContext sc = sqlc.sparkContext(); if (concurrentSQL()) { sc.setLocalProperty("spark.scheduler.pool", "fair"); } else { sc.setLocalProperty("spark.scheduler.pool", null); } sc.setJobGroup(getJobGroup(context), "Zeppelin", false); Object rdd = null; try { Method sqlMethod = sqlc.getClass().getMethod("sql", String.class); rdd = sqlMethod.invoke(sqlc, st); } catch (NoSuchMethodException | SecurityException | IllegalAccessException | IllegalArgumentException | InvocationTargetException e) { throw new InterpreterException(e); } String msg = ZeppelinContext.showDF(sc, context, rdd, maxResult); sc.clearJobGroup(); return new InterpreterResult(Code.SUCCESS, msg); } SparkSqlInterpreter(Properties property); }
SparkSqlInterpreter extends Interpreter { @Override public InterpreterResult interpret(String st, InterpreterContext context) { SQLContext sqlc = null; SparkInterpreter sparkInterpreter = getSparkInterpreter(); if (sparkInterpreter.getSparkVersion().isUnsupportedVersion()) { return new InterpreterResult(Code.ERROR, "Spark " + sparkInterpreter.getSparkVersion().toString() + " is not supported"); } sqlc = getSparkInterpreter().getSQLContext(); SparkContext sc = sqlc.sparkContext(); if (concurrentSQL()) { sc.setLocalProperty("spark.scheduler.pool", "fair"); } else { sc.setLocalProperty("spark.scheduler.pool", null); } sc.setJobGroup(getJobGroup(context), "Zeppelin", false); Object rdd = null; try { Method sqlMethod = sqlc.getClass().getMethod("sql", String.class); rdd = sqlMethod.invoke(sqlc, st); } catch (NoSuchMethodException | SecurityException | IllegalAccessException | IllegalArgumentException | InvocationTargetException e) { throw new InterpreterException(e); } String msg = ZeppelinContext.showDF(sc, context, rdd, maxResult); sc.clearJobGroup(); return new InterpreterResult(Code.SUCCESS, msg); } SparkSqlInterpreter(Properties property); @Override void open(); boolean concurrentSQL(); @Override void close(); @Override InterpreterResult interpret(String st, InterpreterContext context); @Override void cancel(InterpreterContext context); @Override FormType getFormType(); @Override int getProgress(InterpreterContext context); @Override Scheduler getScheduler(); @Override List<String> completion(String buf, int cursor); }
SparkSqlInterpreter extends Interpreter { @Override public InterpreterResult interpret(String st, InterpreterContext context) { SQLContext sqlc = null; SparkInterpreter sparkInterpreter = getSparkInterpreter(); if (sparkInterpreter.getSparkVersion().isUnsupportedVersion()) { return new InterpreterResult(Code.ERROR, "Spark " + sparkInterpreter.getSparkVersion().toString() + " is not supported"); } sqlc = getSparkInterpreter().getSQLContext(); SparkContext sc = sqlc.sparkContext(); if (concurrentSQL()) { sc.setLocalProperty("spark.scheduler.pool", "fair"); } else { sc.setLocalProperty("spark.scheduler.pool", null); } sc.setJobGroup(getJobGroup(context), "Zeppelin", false); Object rdd = null; try { Method sqlMethod = sqlc.getClass().getMethod("sql", String.class); rdd = sqlMethod.invoke(sqlc, st); } catch (NoSuchMethodException | SecurityException | IllegalAccessException | IllegalArgumentException | InvocationTargetException e) { throw new InterpreterException(e); } String msg = ZeppelinContext.showDF(sc, context, rdd, maxResult); sc.clearJobGroup(); return new InterpreterResult(Code.SUCCESS, msg); } SparkSqlInterpreter(Properties property); @Override void open(); boolean concurrentSQL(); @Override void close(); @Override InterpreterResult interpret(String st, InterpreterContext context); @Override void cancel(InterpreterContext context); @Override FormType getFormType(); @Override int getProgress(InterpreterContext context); @Override Scheduler getScheduler(); @Override List<String> completion(String buf, int cursor);  }