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 (InvocationTargetException ite) { if (Boolean.parseBoolean(getProperty("zeppelin.spark.sql.stacktrace"))) { throw new InterpreterException(ite); } logger.error("Invocation target exception", ite); String msg = ite.getTargetException().getMessage() + "\nset zeppelin.spark.sql.stacktrace = true to see full stacktrace"; return new InterpreterResult(Code.ERROR, msg); } catch (NoSuchMethodException | SecurityException | IllegalAccessException | IllegalArgumentException 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<InterpreterCompletion> completion(String buf, int cursor);  }
@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()); ret = sql.interpret("select wrong syntax", context); assertEquals(InterpreterResult.Code.ERROR, ret.code()); assertTrue(ret.message().length() > 0); assertEquals(InterpreterResult.Code.SUCCESS, sql.interpret("select case when name==\"aa\" then name else name end from test", context).code()); }
@Test public void testStruct() { repl.interpret("case class Person(name:String, age:Int)", context); repl.interpret("case class People(group:String, person:Person)", context); repl.interpret( "val gr = sc.parallelize(Seq(People(\"g1\", Person(\"moon\",33)), People(\"g2\", Person(\"sun\",11))))", context); if (isDataFrameSupported()) { repl.interpret("gr.toDF.registerTempTable(\"gr\")", context); } else { repl.interpret("gr.registerTempTable(\"gr\")", context); } InterpreterResult ret = sql.interpret("select * from gr", context); assertEquals(InterpreterResult.Code.SUCCESS, ret.code()); }
@Test public void test_null_value_in_row() { repl.interpret("import org.apache.spark.sql._", context); if (isDataFrameSupported()) { repl.interpret( "import org.apache.spark.sql.types.{StructType,StructField,StringType,IntegerType}", context); } repl.interpret( "def toInt(s:String): Any = {try { s.trim().toInt} catch {case e:Exception => null}}", context); repl.interpret( "val schema = StructType(Seq(StructField(\"name\", StringType, false),StructField(\"age\" , IntegerType, true),StructField(\"other\" , StringType, false)))", context); repl.interpret( "val csv = sc.parallelize(Seq((\"jobs, 51, apple\"), (\"gates, , microsoft\")))", context); repl.interpret( "val raw = csv.map(_.split(\",\")).map(p => Row(p(0),toInt(p(1)),p(2)))", context); repl.interpret("val people = z.sqlContext.applySchema(raw, schema)", context); if (isDataFrameSupported()) { repl.interpret("people.toDF.registerTempTable(\"people\")", context); } else { repl.interpret("people.registerTempTable(\"people\")", context); } InterpreterResult ret = sql.interpret( "select name, age from people where name = 'gates'", context); System.err.println("RET=" + ret.message()); assertEquals(InterpreterResult.Code.SUCCESS, ret.code()); assertEquals(Type.TABLE, ret.type()); assertEquals("name\tage\ngates\tnull\n", ret.message()); }