mirror of
https://github.com/apache/zeppelin
synced 2026-05-24 09:38:26 +00:00
change version from 2.3 to 2.6 in .travis.yml
This commit is contained in:
Jeff Zhang
parent
6c9795f3d9
commit
2dd73e4969
3 changed files with 141 additions and 145 deletions
30
.travis.yml
30
.travis.yml
|
|
@ -15,7 +15,7 @@
|
|||
|
||||
language: java
|
||||
|
||||
sudo: false
|
||||
-sudo: false
|
||||
|
||||
cache:
|
||||
apt: true
|
||||
|
|
@ -40,43 +40,39 @@ matrix:
|
|||
include:
|
||||
# Test License compliance using RAT tool
|
||||
- jdk: "oraclejdk7"
|
||||
env: SCALA_VER="2.11" SPARK_VER="2.0.1" HADOOP_VER="2.3" PROFILE="-Prat" BUILD_FLAG="clean" TEST_FLAG="org.apache.rat:apache-rat-plugin:check" TEST_PROJECTS=""
|
||||
env: SCALA_VER="2.11" SPARK_VER="2.0.2" HADOOP_VER="2.6" PROFILE="-Prat" BUILD_FLAG="clean" TEST_FLAG="org.apache.rat:apache-rat-plugin:check" TEST_PROJECTS=""
|
||||
|
||||
# Test all modules with spark 2.0.2 and scala 2.11
|
||||
- jdk: "oraclejdk7"
|
||||
env: SCALA_VER="2.11" SPARK_VER="2.0.2" HADOOP_VER="2.3" PROFILE="-Pspark-2.0 -Phadoop-2.3 -Ppyspark -Psparkr -Pscalding -Pexamples -Pscala-2.11" BUILD_FLAG="package -Pbuild-distr -DskipRat" TEST_FLAG="verify -Pusing-packaged-distr -DskipRat" TEST_PROJECTS=""
|
||||
env: SCALA_VER="2.11" SPARK_VER="2.0.2" HADOOP_VER="2.6" PROFILE="-Pspark-2.0 -Phadoop-2.6 -Ppyspark -Psparkr -Pscalding -Pexamples -Pscala-2.11" BUILD_FLAG="package -Pbuild-distr -DskipRat" TEST_FLAG="verify -Pusing-packaged-distr -DskipRat" TEST_PROJECTS=""
|
||||
|
||||
# Test all modules with scala 2.10
|
||||
- jdk: "oraclejdk7"
|
||||
env: SCALA_VER="2.10" SPARK_VER="1.6.3" HADOOP_VER="2.3" PROFILE="-Pspark-1.6 -Pr -Phadoop-2.3 -Ppyspark -Psparkr -Pscalding -Pbeam -Pexamples -Pscala-2.10" BUILD_FLAG="package -Pbuild-distr -DskipRat" TEST_FLAG="verify -Pusing-packaged-distr -DskipRat" TEST_PROJECTS=""
|
||||
env: SCALA_VER="2.10" SPARK_VER="1.6.3" HADOOP_VER="2.6" PROFILE="-Pspark-1.6 -Pr -Phadoop-2.6 -Ppyspark -Psparkr -Pscalding -Pbeam -Pexamples -Pscala-2.10" BUILD_FLAG="package -Pbuild-distr -DskipRat" TEST_FLAG="verify -Pusing-packaged-distr -DskipRat" TEST_PROJECTS=""
|
||||
|
||||
# Test all modules with scala 2.11
|
||||
- jdk: "oraclejdk7"
|
||||
env: SCALA_VER="2.11" SPARK_VER="1.6.3" HADOOP_VER="2.3" PROFILE="-Pspark-1.6 -Pr -Phadoop-2.3 -Ppyspark -Psparkr -Pscalding -Pexamples -Pscala-2.11" BUILD_FLAG="package -Pbuild-distr -DskipRat" TEST_FLAG="verify -Pusing-packaged-distr -DskipRat" TEST_PROJECTS=""
|
||||
env: SCALA_VER="2.11" SPARK_VER="1.6.3" HADOOP_VER="2.6" PROFILE="-Pspark-1.6 -Pr -Phadoop-2.6 -Ppyspark -Psparkr -Pscalding -Pexamples -Pscala-2.11" BUILD_FLAG="package -Pbuild-distr -DskipRat" TEST_FLAG="verify -Pusing-packaged-distr -DskipRat" TEST_PROJECTS=""
|
||||
|
||||
# Test spark module for 1.5.2
|
||||
- jdk: "oraclejdk7"
|
||||
env: SCALA_VER="2.10" SPARK_VER="1.5.2" HADOOP_VER="2.3" PROFILE="-Pspark-1.5 -Pr -Phadoop-2.3 -Ppyspark -Psparkr" BUILD_FLAG="package -DskipTests -DskipRat" TEST_FLAG="verify -DskipRat" TEST_PROJECTS="-pl zeppelin-interpreter,zeppelin-zengine,zeppelin-server,zeppelin-display,spark-dependencies,spark,r -Dtest=org.apache.zeppelin.rest.*Test,org.apache.zeppelin.spark.* -DfailIfNoTests=false"
|
||||
env: SCALA_VER="2.10" SPARK_VER="1.5.2" HADOOP_VER="2.6" LIVY_VER="0.2.0" PROFILE="-Pspark-1.5 -Pr -Phadoop-2.6 -Ppyspark -Psparkr" BUILD_FLAG="package -DskipTests -DskipRat" TEST_FLAG="verify -DskipRat" TEST_PROJECTS="-pl zeppelin-interpreter,zeppelin-zengine,zeppelin-server,zeppelin-display,spark-dependencies,spark,r,livy -Dtest=org.apache.zeppelin.rest.*Test,org.apache.zeppelin.spark.* -DfailIfNoTests=false"
|
||||
|
||||
# Test spark module for 1.4.1
|
||||
- jdk: "oraclejdk7"
|
||||
env: SCALA_VER="2.10" SPARK_VER="1.4.1" HADOOP_VER="2.3" PROFILE="-Pspark-1.4 -Pr -Phadoop-2.3 -Ppyspark -Psparkr" BUILD_FLAG="package -DskipTests -DskipRat" TEST_FLAG="verify -DskipRat" TEST_PROJECTS="-pl zeppelin-interpreter,zeppelin-zengine,zeppelin-server,zeppelin-display,spark-dependencies,spark,r -Dtest=org.apache.zeppelin.rest.*Test,org.apache.zeppelin.spark.* -DfailIfNoTests=false"
|
||||
env: SCALA_VER="2.10" SPARK_VER="1.4.1" HADOOP_VER="2.6" PROFILE="-Pspark-1.4 -Pr -Phadoop-2.6 -Ppyspark -Psparkr" BUILD_FLAG="package -DskipTests -DskipRat" TEST_FLAG="verify -DskipRat" TEST_PROJECTS="-pl zeppelin-interpreter,zeppelin-zengine,zeppelin-server,zeppelin-display,spark-dependencies,spark,r -Dtest=org.apache.zeppelin.rest.*Test,org.apache.zeppelin.spark.* -DfailIfNoTests=false"
|
||||
|
||||
# Test selenium with spark module for 1.6.3
|
||||
- jdk: "oraclejdk7"
|
||||
env: TEST_SELENIUM="true" SCALA_VER="2.10" SPARK_VER="1.6.3" HADOOP_VER="2.3" PROFILE="-Pspark-1.6 -Phadoop-2.3 -Ppyspark -Pexamples" BUILD_FLAG="package -DskipTests -DskipRat" TEST_FLAG="verify -DskipRat" TEST_PROJECTS="-pl zeppelin-interpreter,zeppelin-zengine,zeppelin-server,zeppelin-display,spark-dependencies,spark -Dtest=org.apache.zeppelin.AbstractFunctionalSuite -DfailIfNoTests=false"
|
||||
env: TEST_SELENIUM="true" SCALA_VER="2.10" SPARK_VER="1.6.3" HADOOP_VER="2.6" PROFILE="-Pspark-1.6 -Phadoop-2.6 -Ppyspark -Pexamples" BUILD_FLAG="package -DskipTests -DskipRat" TEST_FLAG="verify -DskipRat" TEST_PROJECTS="-pl zeppelin-interpreter,zeppelin-zengine,zeppelin-server,zeppelin-display,spark-dependencies,spark -Dtest=org.apache.zeppelin.AbstractFunctionalSuite -DfailIfNoTests=false"
|
||||
|
||||
# Test python/pyspark with python 2
|
||||
- jdk: "oraclejdk7"
|
||||
env: PYTHON="2" SCALA_VER="2.10" SPARK_VER="1.6.1" HADOOP_VER="2.3" PROFILE="-Pspark-1.6 -Phadoop-2.3 -Ppyspark" BUILD_FLAG="package -pl spark,python -am -DskipTests -DskipRat" TEST_FLAG="verify -DskipRat" TEST_PROJECTS="-pl zeppelin-interpreter,zeppelin-display,spark-dependencies,spark,python -Dtest=org.apache.zeppelin.spark.PySpark*Test,org.apache.zeppelin.python.* -Dpyspark.test.exclude='' -DfailIfNoTests=false"
|
||||
env: PYTHON="2" SCALA_VER="2.10" SPARK_VER="1.6.1" HADOOP_VER="2.6" PROFILE="-Pspark-1.6 -Phadoop-2.6 -Ppyspark" BUILD_FLAG="package -pl spark,python -am -DskipTests -DskipRat" TEST_FLAG="verify -DskipRat" TEST_PROJECTS="-pl zeppelin-interpreter,zeppelin-display,spark-dependencies,spark,python -Dtest=org.apache.zeppelin.spark.PySpark*Test,org.apache.zeppelin.python.* -Dpyspark.test.exclude='' -DfailIfNoTests=false"
|
||||
|
||||
# Test python/pyspark with python 3
|
||||
- jdk: "oraclejdk7"
|
||||
env: PYTHON="3" SCALA_VER="2.11" SPARK_VER="2.0.0" HADOOP_VER="2.3" PROFILE="-Pspark-2.0 -Phadoop-2.3 -Ppyspark -Pscala-2.11" BUILD_FLAG="package -pl spark,python -am -DskipTests -DskipRat" TEST_FLAG="verify -DskipRat" TEST_PROJECTS="-pl zeppelin-interpreter,zeppelin-display,spark-dependencies,spark,python -Dtest=org.apache.zeppelin.spark.PySpark*Test,org.apache.zeppelin.python.* -Dpyspark.test.exclude='' -DfailIfNoTests=false"
|
||||
|
||||
# Test livy with spark 1.5.2 and hadoop 2.6
|
||||
- jdk: "oraclejdk7"
|
||||
env: SCALA_VER="2.10" SPARK_VER="1.5.2" HADOOP_VER="2.6" LIVY_VER="0.2.0" PROFILE="-Pspark-1.5 -Phadoop-2.6" BUILD_FLAG="package -DskipTests -DskipRat" TEST_FLAG="verify -DskipRat" TEST_PROJECTS="-pl zeppelin-interpreter,livy -DfailIfNoTests=false"
|
||||
env: PYTHON="3" SCALA_VER="2.11" SPARK_VER="2.0.0" HADOOP_VER="2.6" PROFILE="-Pspark-2.0 -Phadoop-2.6 -Ppyspark -Pscala-2.11" BUILD_FLAG="package -pl spark,python -am -DskipTests -DskipRat" TEST_FLAG="verify -DskipRat" TEST_PROJECTS="-pl zeppelin-interpreter,zeppelin-display,spark-dependencies,spark,python -Dtest=org.apache.zeppelin.spark.PySpark*Test,org.apache.zeppelin.python.* -Dpyspark.test.exclude='' -DfailIfNoTests=false"
|
||||
|
||||
before_install:
|
||||
- echo "MAVEN_OPTS='-Xms1024M -Xmx2048M -XX:MaxPermSize=1024m -XX:-UseGCOverheadLimit -Dorg.slf4j.simpleLogger.defaultLogLevel=warn'" >> ~/.mavenrc
|
||||
|
|
@ -106,6 +102,7 @@ after_success:
|
|||
|
||||
after_failure:
|
||||
- echo "Travis exited with ${TRAVIS_TEST_RESULT}"
|
||||
- cat /home/travis/build.sh
|
||||
- cat target/rat.txt
|
||||
- cat zeppelin-server/target/rat.txt
|
||||
- cat zeppelin-distribution/target/zeppelin-*-SNAPSHOT/zeppelin-*-SNAPSHOT/logs/zeppelin*.log
|
||||
|
|
@ -113,9 +110,6 @@ after_failure:
|
|||
- cat zeppelin-web/npm-debug.log
|
||||
- cat spark-*/logs/*
|
||||
- cat livy/target/tmp/*/output.log
|
||||
- ls -l livy/target/tmp/MiniYarnMain/target/com.cloudera.livy.test.framework.MiniYarnMain/*
|
||||
- ls -l livy/target/tmp/MiniYarnMain/target/com.cloudera.livy.test.framework.MiniYarnMain/*/*
|
||||
- ls -l livy/target/tmp/MiniYarnMain/target/com.cloudera.livy.test.framework.MiniYarnMain/*/*/*
|
||||
- ls -l livy/target/tmp/MiniYarnMain/target/com.cloudera.livy.test.framework.MiniYarnMain/*/*/*/*
|
||||
- ls -R livy/target/tmp/MiniYarnMain/target/com.cloudera.livy.test.framework.MiniYarnMain/*
|
||||
- cat livy/target/tmp/MiniYarnMain/target/com.cloudera.livy.test.framework.MiniYarnMain/*/*/*/stdout
|
||||
- cat livy/target/tmp/MiniYarnMain/target/com.cloudera.livy.test.framework.MiniYarnMain/*/*/*/stderr
|
||||
|
|
|
|||
|
|
@ -59,6 +59,8 @@ public abstract class BaseLivyInterprereter extends Interpreter {
|
|||
public void close() {
|
||||
if (sessionId != -1) {
|
||||
livyHelper.closeSession(sessionId);
|
||||
// reset sessionId to -1
|
||||
sessionId = -1;
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -66,14 +68,14 @@ public abstract class BaseLivyInterprereter extends Interpreter {
|
|||
sessionId = livyHelper.createSession(context, getSessionKind());
|
||||
if (displayAppInfo) {
|
||||
this.appId = extractStatementResult(
|
||||
livyHelper.interpret("sc.applicationId", context, sessionId).message());
|
||||
livyHelper.interpret("sc.applicationId", context, sessionId).message().get(0).getData());
|
||||
livyHelper.interpret(
|
||||
"val webui=sc.getClass.getMethod(\"ui\").invoke(sc).asInstanceOf[Some[_]].get",
|
||||
context, sessionId);
|
||||
this.webUIAddress = extractStatementResult(
|
||||
livyHelper.interpret(
|
||||
"webui.getClass.getMethod(\"appUIAddress\").invoke(webui)",
|
||||
context, sessionId).message());
|
||||
context, sessionId).message().get(0).getData());
|
||||
LOGGER.info("Create livy session with sessionId: {}, appId: {}, webUI: {}",
|
||||
sessionId, appId, webUIAddress);
|
||||
}
|
||||
|
|
|
|||
|
|
@ -88,74 +88,71 @@ public class LivyInterpreterIT {
|
|||
InterpreterContext context = new InterpreterContext("noteId", "paragraphId", "livy.spark",
|
||||
"title", "text", authInfo, null, null, null, null, null, output);
|
||||
sparkInterpreter.open();
|
||||
InterpreterResult result = sparkInterpreter.interpret("sc.version", context);
|
||||
assertEquals(InterpreterResult.Code.SUCCESS, result.code());
|
||||
assertEquals(InterpreterResult.Type.TEXT, result.type());
|
||||
assertNull(result.message());
|
||||
assertTrue(outputListener.getOutputAppended().contains("1.5.2"));
|
||||
|
||||
// test RDD api
|
||||
outputListener.reset();
|
||||
result = sparkInterpreter.interpret("sc.parallelize(1 to 10).sum()", context);
|
||||
assertEquals(InterpreterResult.Code.SUCCESS, result.code());
|
||||
assertEquals(InterpreterResult.Type.TEXT, result.type());
|
||||
assertNull(result.message());
|
||||
assertTrue(outputListener.getOutputAppended().contains("Double = 55.0"));
|
||||
|
||||
// single line comment
|
||||
outputListener.reset();
|
||||
String singleLineComment = "// my comment";
|
||||
result = sparkInterpreter.interpret(singleLineComment, context);
|
||||
assertEquals(InterpreterResult.Code.SUCCESS, result.code());
|
||||
assertEquals(InterpreterResult.Type.TEXT, result.message().get(0).getType());
|
||||
assertNull(result.message().get(0).getData());
|
||||
try {
|
||||
InterpreterResult result = sparkInterpreter.interpret("sc.version", context);
|
||||
assertEquals(InterpreterResult.Code.SUCCESS, result.code());
|
||||
assertEquals(0, result.message().size());
|
||||
assertTrue(outputListener.getOutputAppended().contains("1.5.2"));
|
||||
|
||||
// multiple line comment
|
||||
outputListener.reset();
|
||||
String multipleLineComment = "/* multiple \n" + "line \n" + "comment */";
|
||||
result = sparkInterpreter.interpret(multipleLineComment, context);
|
||||
assertEquals(InterpreterResult.Code.SUCCESS, result.code());
|
||||
assertEquals(InterpreterResult.Type.TEXT, result.message().get(0).getType());
|
||||
assertNull(result.message().get(0).getData());
|
||||
// test RDD api
|
||||
outputListener.reset();
|
||||
result = sparkInterpreter.interpret("sc.parallelize(1 to 10).sum()", context);
|
||||
assertEquals(InterpreterResult.Code.SUCCESS, result.code());
|
||||
assertEquals(0, result.message().size());
|
||||
assertTrue(outputListener.getOutputAppended().contains("Double = 55.0"));
|
||||
|
||||
// multi-line string
|
||||
outputListener.reset();
|
||||
String multiLineString = "val str = \"\"\"multiple\n" +
|
||||
"line\"\"\"\n" +
|
||||
"println(str)";
|
||||
result = sparkInterpreter.interpret(multiLineString, context);
|
||||
assertEquals(InterpreterResult.Code.SUCCESS, result.code());
|
||||
assertEquals(InterpreterResult.Type.TEXT, result.message().get(0).getType());
|
||||
assertNull(result.message().get(0).getData());
|
||||
assertTrue(outputListener.getOutputAppended().contains("multiple\nline"));
|
||||
// single line comment
|
||||
outputListener.reset();
|
||||
String singleLineComment = "// my comment";
|
||||
result = sparkInterpreter.interpret(singleLineComment, context);
|
||||
assertEquals(InterpreterResult.Code.SUCCESS, result.code());
|
||||
assertEquals(0, result.message().size());
|
||||
|
||||
// case class
|
||||
outputListener.reset();
|
||||
String caseClassCode = "case class Person(id:Int, \n" +
|
||||
"name:String)\n" +
|
||||
"val p=Person(1, \"name_a\")";
|
||||
result = sparkInterpreter.interpret(caseClassCode, context);
|
||||
assertEquals(InterpreterResult.Code.SUCCESS, result.code());
|
||||
assertEquals(InterpreterResult.Type.TEXT, result.message().get(0).getType());
|
||||
assertNull(result.message().get(0).getData());
|
||||
assertTrue(outputListener.getOutputAppended().contains("defined class Person"));
|
||||
// multiple line comment
|
||||
outputListener.reset();
|
||||
String multipleLineComment = "/* multiple \n" + "line \n" + "comment */";
|
||||
result = sparkInterpreter.interpret(multipleLineComment, context);
|
||||
assertEquals(InterpreterResult.Code.SUCCESS, result.code());
|
||||
assertEquals(0, result.message().size());
|
||||
|
||||
// object class
|
||||
outputListener.reset();
|
||||
String objectClassCode = "object Person {}";
|
||||
result = sparkInterpreter.interpret(objectClassCode, context);
|
||||
assertEquals(InterpreterResult.Code.SUCCESS, result.code());
|
||||
assertEquals(InterpreterResult.Type.TEXT, result.message().get(0).getType());
|
||||
assertNull(result.message().get(0).getData());
|
||||
assertTrue(outputListener.getOutputAppended().contains("defined module Person"));
|
||||
// multi-line string
|
||||
outputListener.reset();
|
||||
String multiLineString = "val str = \"\"\"multiple\n" +
|
||||
"line\"\"\"\n" +
|
||||
"println(str)";
|
||||
result = sparkInterpreter.interpret(multiLineString, context);
|
||||
assertEquals(InterpreterResult.Code.SUCCESS, result.code());
|
||||
assertEquals(0, result.message().size());
|
||||
assertTrue(outputListener.getOutputAppended().contains("multiple\nline"));
|
||||
|
||||
// error
|
||||
result = sparkInterpreter.interpret("println(a)", context);
|
||||
assertEquals(InterpreterResult.Code.ERROR, result.code());
|
||||
assertEquals(InterpreterResult.Type.TEXT, result.type());
|
||||
assertTrue(result.message().contains("error: not found: value a"));
|
||||
// case class
|
||||
outputListener.reset();
|
||||
String caseClassCode = "case class Person(id:Int, \n" +
|
||||
"name:String)\n" +
|
||||
"val p=Person(1, \"name_a\")";
|
||||
result = sparkInterpreter.interpret(caseClassCode, context);
|
||||
assertEquals(InterpreterResult.Code.SUCCESS, result.code());
|
||||
assertEquals(0, result.message().size());
|
||||
assertTrue(outputListener.getOutputAppended().contains("defined class Person"));
|
||||
|
||||
sparkInterpreter.close();
|
||||
// object class
|
||||
outputListener.reset();
|
||||
String objectClassCode = "object Person {}";
|
||||
result = sparkInterpreter.interpret(objectClassCode, context);
|
||||
assertEquals(InterpreterResult.Code.SUCCESS, result.code());
|
||||
assertEquals(0, result.message().size());
|
||||
assertTrue(outputListener.getOutputAppended().contains("defined module Person"));
|
||||
|
||||
// error
|
||||
result = sparkInterpreter.interpret("println(a)", context);
|
||||
assertEquals(InterpreterResult.Code.ERROR, result.code());
|
||||
assertEquals(InterpreterResult.Type.TEXT, result.message().get(0).getType());
|
||||
assertTrue(result.message().get(0).getData().contains("error: not found: value a"));
|
||||
|
||||
} finally {
|
||||
sparkInterpreter.close();
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
|
|
@ -174,34 +171,35 @@ public class LivyInterpreterIT {
|
|||
InterpreterContext context = new InterpreterContext("noteId", "paragraphId", "livy.spark",
|
||||
"title", "text", authInfo, null, null, null, null, null, output);
|
||||
sparkInterpreter.open();
|
||||
|
||||
// test DataFrame api
|
||||
outputListener.reset();
|
||||
sparkInterpreter.interpret("val sqlContext = new org.apache.spark.sql.SQLContext(sc)\n"
|
||||
+ "import sqlContext.implicits._", context);
|
||||
InterpreterResult result = sparkInterpreter.interpret("val df=sqlContext.createDataFrame(Seq((\"hello\",20)))\n"
|
||||
+ "df.collect()" , context);
|
||||
assertEquals(InterpreterResult.Code.SUCCESS, result.code());
|
||||
assertEquals(InterpreterResult.Type.TEXT, result.type());
|
||||
assertNull(result.message());
|
||||
assertTrue(outputListener.getOutputAppended()
|
||||
.contains("Array[org.apache.spark.sql.Row] = Array([hello,20])"));
|
||||
sparkInterpreter.interpret("df.registerTempTable(\"df\")", context);
|
||||
|
||||
// test LivySparkSQLInterpreter which share the same SparkContext with LivySparkInterpreter
|
||||
outputListener.reset();
|
||||
LivySparkSQLInterpreter sqlInterpreter = new LivySparkSQLInterpreter(properties);
|
||||
interpreterGroup.get("session_1").add(sqlInterpreter);
|
||||
sqlInterpreter.setInterpreterGroup(interpreterGroup);
|
||||
interpreterGroup.get("session_1").add(sqlInterpreter);
|
||||
sqlInterpreter.open();
|
||||
result = sqlInterpreter.interpret("select * from df", context);
|
||||
assertEquals(InterpreterResult.Code.SUCCESS, result.code());
|
||||
assertEquals(InterpreterResult.Type.TABLE, result.type());
|
||||
// TODO(zjffdu), \t at the end of each line is not necessary, it is a bug of LivySparkSQLInterpreter
|
||||
assertEquals("_1\t_2\t\nhello\t20\t\n", result.message());
|
||||
|
||||
sparkInterpreter.close();
|
||||
sqlInterpreter.close();
|
||||
try {
|
||||
// test DataFrame api
|
||||
outputListener.reset();
|
||||
sparkInterpreter.interpret("val sqlContext = new org.apache.spark.sql.SQLContext(sc)\n"
|
||||
+ "import sqlContext.implicits._", context);
|
||||
InterpreterResult result = sparkInterpreter.interpret("val df=sqlContext.createDataFrame(Seq((\"hello\",20)))\n"
|
||||
+ "df.collect()", context);
|
||||
assertEquals(InterpreterResult.Code.SUCCESS, result.code());
|
||||
assertEquals(0, result.message().size());
|
||||
assertTrue(outputListener.getOutputAppended()
|
||||
.contains("Array[org.apache.spark.sql.Row] = Array([hello,20])"));
|
||||
sparkInterpreter.interpret("df.registerTempTable(\"df\")", context);
|
||||
|
||||
// test LivySparkSQLInterpreter which share the same SparkContext with LivySparkInterpreter
|
||||
outputListener.reset();
|
||||
result = sqlInterpreter.interpret("select * from df", context);
|
||||
assertEquals(InterpreterResult.Code.SUCCESS, result.code());
|
||||
assertEquals(InterpreterResult.Type.TABLE, result.message().get(0).getType());
|
||||
// TODO(zjffdu), \t at the end of each line is not necessary, it is a bug of LivySparkSQLInterpreter
|
||||
assertEquals("_1\t_2\t\nhello\t20\t\n", result.message().get(0).getData());
|
||||
} finally {
|
||||
sparkInterpreter.close();
|
||||
sqlInterpreter.close();
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
|
|
@ -219,17 +217,19 @@ public class LivyInterpreterIT {
|
|||
sqlInterpreter.setInterpreterGroup(interpreterGroup);
|
||||
sqlInterpreter.open();
|
||||
|
||||
AuthenticationInfo authInfo = new AuthenticationInfo("user1");
|
||||
MyInterpreterOutputListener outputListener = new MyInterpreterOutputListener();
|
||||
InterpreterOutput output = new InterpreterOutput(outputListener);
|
||||
InterpreterContext context = new InterpreterContext("noteId", "paragraphId", "livy.sql",
|
||||
"title", "text", authInfo, null, null, null, null, null, output);
|
||||
InterpreterResult result = sqlInterpreter.interpret("show tables", context);
|
||||
assertEquals(InterpreterResult.Code.SUCCESS, result.code());
|
||||
assertEquals(InterpreterResult.Type.TABLE, result.type());
|
||||
assertTrue(result.message().contains("tableName"));
|
||||
|
||||
sqlInterpreter.close();
|
||||
try {
|
||||
AuthenticationInfo authInfo = new AuthenticationInfo("user1");
|
||||
MyInterpreterOutputListener outputListener = new MyInterpreterOutputListener();
|
||||
InterpreterOutput output = new InterpreterOutput(outputListener);
|
||||
InterpreterContext context = new InterpreterContext("noteId", "paragraphId", "livy.sql",
|
||||
"title", "text", authInfo, null, null, null, null, null, output);
|
||||
InterpreterResult result = sqlInterpreter.interpret("show tables", context);
|
||||
assertEquals(InterpreterResult.Code.SUCCESS, result.code());
|
||||
assertEquals(InterpreterResult.Type.TABLE, result.message().get(0).getType());
|
||||
assertTrue(result.message().get(0).getData().contains("tableName"));
|
||||
} finally {
|
||||
sqlInterpreter.close();
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
|
|
@ -245,41 +245,41 @@ public class LivyInterpreterIT {
|
|||
InterpreterContext context = new InterpreterContext("noteId", "paragraphId", "livy.pyspark",
|
||||
"title", "text", authInfo, null, null, null, null, null, output);
|
||||
pysparkInterpreter.open();
|
||||
InterpreterResult result = pysparkInterpreter.interpret("sc.version", context);
|
||||
assertEquals(InterpreterResult.Code.SUCCESS, result.code());
|
||||
assertEquals(InterpreterResult.Type.TEXT, result.type());
|
||||
assertNull(result.message());
|
||||
assertTrue(outputListener.getOutputAppended().contains("1.5.2"));
|
||||
|
||||
// test RDD api
|
||||
outputListener.reset();
|
||||
result = pysparkInterpreter.interpret("sc.range(1, 10).sum()", context);
|
||||
assertEquals(InterpreterResult.Code.SUCCESS, result.code());
|
||||
assertEquals(InterpreterResult.Type.TEXT, result.type());
|
||||
assertNull(result.message());
|
||||
assertTrue(outputListener.getOutputAppended().contains("45"));
|
||||
try {
|
||||
InterpreterResult result = pysparkInterpreter.interpret("sc.version", context);
|
||||
assertEquals(InterpreterResult.Code.SUCCESS, result.code());
|
||||
assertEquals(0, result.message().size());
|
||||
assertTrue(outputListener.getOutputAppended().contains("1.5.2"));
|
||||
|
||||
// test DataFrame api
|
||||
outputListener.reset();
|
||||
pysparkInterpreter.interpret("from pyspark.sql import SQLContext\n"
|
||||
+ "sqlContext = SQLContext(sc)", context);
|
||||
result = pysparkInterpreter.interpret("df=sqlContext.createDataFrame([(\"hello\",20)])\n"
|
||||
+ "df.collect()" , context);
|
||||
assertEquals(InterpreterResult.Type.TEXT, result.type());
|
||||
assertEquals(InterpreterResult.Code.SUCCESS, result.code());
|
||||
assertNull(result.message());
|
||||
assertTrue(outputListener.getOutputAppended().contains("[Row(_1=u'hello', _2=20)]"));
|
||||
// test RDD api
|
||||
outputListener.reset();
|
||||
result = pysparkInterpreter.interpret("sc.range(1, 10).sum()", context);
|
||||
assertEquals(InterpreterResult.Code.SUCCESS, result.code());
|
||||
assertEquals(0, result.message().size());
|
||||
assertTrue(outputListener.getOutputAppended().contains("45"));
|
||||
|
||||
// error
|
||||
result = pysparkInterpreter.interpret("print(a)", context);
|
||||
assertEquals(InterpreterResult.Code.ERROR, result.code());
|
||||
assertEquals(InterpreterResult.Type.TEXT, result.type());
|
||||
assertTrue(result.message().contains("name 'a' is not defined"));
|
||||
// test DataFrame api
|
||||
outputListener.reset();
|
||||
pysparkInterpreter.interpret("from pyspark.sql import SQLContext\n"
|
||||
+ "sqlContext = SQLContext(sc)", context);
|
||||
result = pysparkInterpreter.interpret("df=sqlContext.createDataFrame([(\"hello\",20)])\n"
|
||||
+ "df.collect()", context);
|
||||
assertEquals(InterpreterResult.Code.SUCCESS, result.code());
|
||||
assertEquals(0, result.message().size());
|
||||
assertTrue(outputListener.getOutputAppended().contains("[Row(_1=u'hello', _2=20)]"));
|
||||
|
||||
pysparkInterpreter.close();
|
||||
// error
|
||||
result = pysparkInterpreter.interpret("print(a)", context);
|
||||
assertEquals(InterpreterResult.Code.ERROR, result.code());
|
||||
assertEquals(InterpreterResult.Type.TEXT, result.message().get(0).getType());
|
||||
assertTrue(result.message().get(0).getData().contains("name 'a' is not defined"));
|
||||
} finally {
|
||||
pysparkInterpreter.close();
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
// @Test
|
||||
public void testSparkRInterpreter() {
|
||||
if (!checkPreCondition()) {
|
||||
return;
|
||||
|
|
|
|||
Loading…
Reference in a new issue