PythonInterpreterPandasSql extends Interpreter { @Override public InterpreterResult interpret(String st, InterpreterContext context) throws InterpreterException { LOG.info("Running SQL query: '{}' over Pandas DataFrame", st); return pythonInterpreter.interpret( "__zeppelin__.show(pysqldf('" + st + "'))\n__zeppelin__._displayhook()", context); } PythonInterpreterPandasSql(Properties property); @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);  }
@Test public void dependenciesAreInstalled() throws InterpreterException { InterpreterResult ret = python.interpret("import pandas\nimport pandasql\nimport numpy\n", context); assertEquals(ret.message().toString(), InterpreterResult.Code.SUCCESS, ret.code()); }
@Test public void errorMessageIfDependenciesNotInstalled() throws InterpreterException { InterpreterResult ret; ret = sql.interpret("SELECT * from something", context); assertNotNull(ret); assertEquals(ret.message().get(0).getData(), InterpreterResult.Code.ERROR, ret.code()); assertTrue(ret.message().get(0).getData().contains("no such table: something")); }
@Test public void sqlOverTestDataPrintsTable() throws IOException, InterpreterException { InterpreterResult ret; ret = python.interpret("import pandas as pd", context); ret = python.interpret("import numpy as np", context); ret = python.interpret("df2 = pd.DataFrame({ 'age' : np.array([33, 51, 51, 34]), " + "'name' : pd.Categorical(['moon','jobs','gates','park'])})", context); assertEquals(ret.message().toString(), InterpreterResult.Code.SUCCESS, ret.code()); ret = sql.interpret("select name, age from df2 where age < 40", context); assertEquals(new String(out.getOutputAt(1).toByteArray()), InterpreterResult.Code.SUCCESS, ret.code()); assertEquals(new String(out.getOutputAt(1).toByteArray()), Type.TABLE, out.getOutputAt(1).getType()); assertTrue(new String(out.getOutputAt(1).toByteArray()).indexOf("moon\t33") > 0); assertTrue(new String(out.getOutputAt(1).toByteArray()).indexOf("park\t34") > 0); assertEquals(InterpreterResult.Code.SUCCESS, sql.interpret("select case when name==\"aa\" then name else name end from df2", context).code()); }
@Test public void badSqlSyntaxFails() throws IOException, InterpreterException { InterpreterResult ret = sql.interpret("select wrong syntax", context); assertNotNull("Interpreter returned 'null'", ret); assertEquals(ret.toString(), InterpreterResult.Code.ERROR, ret.code()); }
@Test public void showDataFrame() throws IOException, InterpreterException { InterpreterResult ret; ret = python.interpret("import pandas as pd", context); ret = python.interpret("import numpy as np", context); ret = python.interpret("index = pd.Index([10, 11, 12, 13], name='index_name')", context); ret = python.interpret("d1 = {1 : [np.nan, 1, 2, 3], 'two' : [3., 4., 5., 6.7]}", context); ret = python.interpret("df1 = pd.DataFrame(d1, index=index)", context); assertEquals(ret.message().toString(), InterpreterResult.Code.SUCCESS, ret.code()); ret = python.interpret("z.show(df1, show_index=True)", context); assertEquals(new String(out.getOutputAt(0).toByteArray()), InterpreterResult.Code.SUCCESS, ret.code()); assertEquals(new String(out.getOutputAt(1).toByteArray()), Type.TABLE, out.getOutputAt(1).getType()); assertTrue(new String(out.getOutputAt(1).toByteArray()).contains("index_name")); assertTrue(new String(out.getOutputAt(1).toByteArray()).contains("nan")); assertTrue(new String(out.getOutputAt(1).toByteArray()).contains("6.7")); }