- *
- * To run manually on such environment, use:
- *
- * mvn -Dpython.test.exclude='' test -pl python -am
- *
- */
-@RunWith(value = Parameterized.class)
-public class PythonInterpreterPandasSqlTest {
-
- private static final Logger LOGGER =
- LoggerFactory.getLogger(PythonInterpreterPandasSqlTest.class);
-
- @Parameterized.Parameters
- public static List data() {
- return Arrays.asList(new Object[][]{
- {true},
- {false}
- });
- }
-
- private InterpreterGroup intpGroup;
- private PythonInterpreterPandasSql pandasSqlInterpreter;
- private PythonInterpreter pythonInterpreter;
-
- private InterpreterContext context;
-
- public PythonInterpreterPandasSqlTest(boolean ignored) {
- }
-
- @Before
- public void setUp() throws Exception {
- Properties p = new Properties();
- p.setProperty("zeppelin.python", "python");
- p.setProperty("zeppelin.python.maxResult", "100");
- p.setProperty("zeppelin.python.gatewayserver_address", "127.0.0.1");
-
- intpGroup = new InterpreterGroup();
-
- context = getInterpreterContext();
- InterpreterContext.set(context);
-
- pythonInterpreter = new PythonInterpreter(p);
- pandasSqlInterpreter = new PythonInterpreterPandasSql(p);
-
- pythonInterpreter.setInterpreterGroup(intpGroup);
- pandasSqlInterpreter.setInterpreterGroup(intpGroup);
-
- List interpreters =
- Arrays.asList(pythonInterpreter, pandasSqlInterpreter);
-
-
- intpGroup.put("session_1", interpreters);
-
- pythonInterpreter.open();
-
- // to make sure python is running.
- InterpreterResult ret = pythonInterpreter.interpret("print(\"python initialized\")\n", context);
- assertEquals(ret.message().toString(), InterpreterResult.Code.SUCCESS, ret.code());
- pandasSqlInterpreter.open();
- }
-
- @After
- public void afterTest() throws InterpreterException {
- pandasSqlInterpreter.close();
- }
-
- @Test
- public void dependenciesAreInstalled() throws InterpreterException {
- InterpreterResult ret =
- pythonInterpreter.interpret("import pandas\nimport pandasql\nimport numpy\n", context);
- assertEquals(ret.message().toString(), InterpreterResult.Code.SUCCESS, ret.code());
- }
-
- @Test
- public void errorMessageIfDependenciesNotInstalled() throws InterpreterException {
- context = getInterpreterContext();
- InterpreterResult ret = pandasSqlInterpreter.interpret("SELECT * from something", context);
-
- assertNotNull(ret);
- assertEquals(context.out.toString(), InterpreterResult.Code.ERROR, ret.code());
- assertTrue(ret.toString(), ret.toString().contains("no such table: something"));
- }
-
- @Test
- public void sqlOverTestDataPrintsTable() throws IOException, InterpreterException {
- InterpreterResult ret = pythonInterpreter.interpret("import pandas as pd\nimport numpy as np", context);
- assertEquals(ret.message().toString(), InterpreterResult.Code.SUCCESS, ret.code());
-
- // DataFrame df2 \w test data
- ret = pythonInterpreter.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());
-
- //when
- context = getInterpreterContext();
- ret = pandasSqlInterpreter.interpret("select name, age from df2 where age < 40", context);
-
- //then
- assertEquals(context.out.toString(), InterpreterResult.Code.SUCCESS, ret.code());
- assertEquals(context.out.toString(), Type.TABLE,
- context.out.toInterpreterResultMessage().get(0).getType());
- assertTrue(context.out.toString().indexOf("moon\t33") > 0);
- assertTrue(context.out.toString().indexOf("park\t34") > 0);
-
- assertEquals(InterpreterResult.Code.SUCCESS,
- pandasSqlInterpreter.interpret(
- "select case when name==\"aa\" then name else name end from df2",
- context).code());
- }
-
- @Test
- public void badSqlSyntaxFails() throws InterpreterException {
- //when
- context = getInterpreterContext();
- InterpreterResult ret = pandasSqlInterpreter.interpret("select wrong syntax", context);
-
- //then
- assertNotNull("Interpreter returned 'null'", ret);
- assertEquals(context.out.toString(), InterpreterResult.Code.ERROR, ret.code());
- }
-
- @Test
- public void showDataFrame() throws IOException, InterpreterException {
- pythonInterpreter.interpret("import pandas as pd", context);
- pythonInterpreter.interpret("import numpy as np", context);
-
- // given a Pandas DataFrame with an index and non-text data
- pythonInterpreter.interpret(
- "index = pd.Index([10, 11, 12, 13], name='index_name')", context);
- pythonInterpreter.interpret(
- "d1 = {1 : [np.nan, 1, 2, 3], 'two' : [3., 4., 5., 6.7]}", context);
- InterpreterResult ret = pythonInterpreter.interpret(
- "df1 = pd.DataFrame(d1, index=index)", context);
- assertEquals(ret.message().toString(), InterpreterResult.Code.SUCCESS, ret.code());
-
- // when
- context = getInterpreterContext();
- ret = pythonInterpreter.interpret("z.show(df1, show_index=True)", context);
-
- // then
- assertEquals(context.out.toString(), InterpreterResult.Code.SUCCESS, ret.code());
- assertEquals(context.out.toString(), Type.TABLE,
- context.out.toInterpreterResultMessage().get(0).getType());
- assertTrue(context.out.toString().contains("index_name"));
- assertTrue(context.out.toString().contains("nan"));
- assertTrue(context.out.toString().contains("6.7"));
- }
-
- private InterpreterContext getInterpreterContext() {
- return InterpreterContext.builder()
- .setNoteId("noteId")
- .setParagraphId("paragraphId")
- .setInterpreterOut(new InterpreterOutput())
- .setIntpEventClient(mock(RemoteInterpreterEventClient.class))
- .build();
- }
-}
diff --git a/python/src/test/java/org/apache/zeppelin/python/PythonInterpreterTest.java b/python/src/test/java/org/apache/zeppelin/python/PythonInterpreterTest.java
index 8f6b1bdf5..54d68c93a 100644
--- a/python/src/test/java/org/apache/zeppelin/python/PythonInterpreterTest.java
+++ b/python/src/test/java/org/apache/zeppelin/python/PythonInterpreterTest.java
@@ -25,7 +25,16 @@
import org.apache.zeppelin.interpreter.InterpreterGroup;
import org.apache.zeppelin.interpreter.InterpreterResult;
import org.apache.zeppelin.interpreter.LazyOpenInterpreter;
-import org.junit.Test;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Disabled;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.junit.jupiter.api.Assertions.fail;
import java.io.IOException;
import java.util.LinkedList;
@@ -34,16 +43,16 @@
import java.util.regex.Matcher;
import java.util.regex.Pattern;
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertFalse;
-import static org.junit.Assert.assertNotNull;
-import static org.junit.Assert.assertTrue;
-import static org.junit.Assert.fail;
-
-
+/**
+ * This test class is also used in spark interpreter module
+ *
+ * @author pdallig
+ */
+@SuppressWarnings("java:S5786")
public class PythonInterpreterTest extends BasePythonInterpreterTest {
-
+
@Override
+ @BeforeEach
public void setUp() throws InterpreterException {
intpGroup = new InterpreterGroup();
@@ -64,6 +73,7 @@ public void setUp() throws InterpreterException {
}
@Override
+ @AfterEach
public void tearDown() throws InterpreterException {
intpGroup.close();
}
@@ -99,7 +109,7 @@ public void run() {
}
//@Test
- public void testCancelIntp() throws InterruptedException, InterpreterException {
+ void testCancelIntp() throws InterruptedException, InterpreterException {
assertEquals(InterpreterResult.Code.SUCCESS,
interpreter.interpret("a = 1\n", getInterpreterContext()).code());
Thread t = new Thread(new infinityPythonJob());
@@ -111,8 +121,9 @@ public void testCancelIntp() throws InterruptedException, InterpreterException {
assertFalse(t.isAlive());
}
+ @Disabled(value="Contains sleep")
@Test
- public void testPythonProcessKilled() throws InterruptedException, TimeoutException {
+ void testPythonProcessKilled() throws InterruptedException, TimeoutException {
final Waiter waiter = new Waiter();
Thread thread = new Thread() {
@Override
@@ -162,7 +173,7 @@ public void testFailtoLaunchPythonProcess() throws InterpreterException {
fail("Should fail to open PythonInterpreter");
} catch (InterpreterException e) {
String stacktrace = ExceptionUtils.getStackTrace(e);
- assertTrue(stacktrace, stacktrace.contains("No such file or directory"));
+ assertTrue(stacktrace.contains("No such file or directory"), stacktrace);
}
}
}
diff --git a/rpm.pom.xml b/rpm.pom.xml
index 2c142d24e..e5a45450d 100644
--- a/rpm.pom.xml
+++ b/rpm.pom.xml
@@ -21,6 +21,7 @@
org.apache.maven.plugins
maven-surefire-plugin
+ 3.5.1
true
@@ -55,10 +56,7 @@
srv-zpln
0644
0755
-
-
- __os_install_post %(echo '%{__os_install_post}' | sed -e 's!/usr/lib[^[:space:]]*/brp-python-bytecompile[[:space:]].*$!!g')
_build_id_links none
__provides_exclude ^osgi\\(.*$
__requires_exclude ^osgi\\(.*$
@@ -83,7 +81,7 @@
true
- ${project.basedir}/zeppelin-distribution/target/zeppelin-0.10.1/zeppelin-0.10.1/bin
+ ${project.basedir}/bin
${project.basedir}/rpm/sudo-wrapper.sh
@@ -100,26 +98,19 @@
true
- ${project.basedir}/zeppelin-distribution/target/zeppelin-0.10.1/zeppelin-0.10.1
-
-
- **/conf/*
- **/bin/*
-
- **/notebook/**
- **/*.zpln
-
- **/*.jar
- **/*.war
-
- **/interpreter/kotlin/**
- **/interpreter/spark/**
-
+ ${project.basedir}/
+
+ README.md
+ NOTICE
+ README.md
+ zeppelin-server/target/zeppelin-server-0.10.1.jar
+
+
- /opt/teragrep/${project.artifactId}
+ /opt/teragrep/${project.artifactId}/interpreter
true
755
755
@@ -128,16 +119,11 @@
true
- ${project.basedir}/zeppelin-distribution/target/zeppelin-0.10.1/zeppelin-0.10.1
+ ${project.basedir}/interpreter
- **/interpreter/kotlin/**
**/interpreter/spark/**
-
- **/*.jar
- **/*.war
-
@@ -151,7 +137,7 @@
true
- ${project.basedir}/zeppelin-distribution/target/zeppelin-0.10.1/zeppelin-0.10.1/interpreter/spark
+ ${project.basedir}/interpreter/spark
@@ -167,7 +153,7 @@
noreplace
- ${project.basedir}/zeppelin-distribution/target/zeppelin-0.10.1/zeppelin-0.10.1/conf
+ ${project.basedir}/conf
**/zeppelin-env.sh*
**/*.properties*
@@ -187,7 +173,7 @@
noreplace
- ${project.basedir}/zeppelin-distribution/target/zeppelin-0.10.1/zeppelin-0.10.1/conf
+ ${project.basedir}/conf
**/zeppelin-env.sh*
**/*.properties*
@@ -195,12 +181,57 @@
+
+
+ /opt/teragrep/${project.artifactId}/licenses
+ true
+ 644
+ 755
+ srv-zpln
+ srv-zpln
+ true
+
+
+ ${project.basedir}/licenses
+
+
+
+
+
+ /opt/teragrep/${project.artifactId}/plugins
+ true
+ 644
+ 755
+ srv-zpln
+ srv-zpln
+ true
+
+
+ ${project.basedir}/plugins
+
+
+
+
+
+ /opt/teragrep/${project.artifactId}/lib
+ true
+ 755
+ 755
+ srv-zpln
+ srv-zpln
+ true
+
+
+ ${project.basedir}/zeppelin-distribution/target/zeppelin-0.10.1/zeppelin-0.10.1/lib/
+
+
+
java-1.8.0-openjdk >= 1.9.0
java-1.8.0-openjdk-headless >= 1.9.0
java-1.8.0-openjdk-devel >= 1.9.0
- python3
+ python3.11
git
sudo
pam
diff --git a/spark/interpreter/pom.xml b/spark/interpreter/pom.xml
index 773db74f2..ad67c6d48 100644
--- a/spark/interpreter/pom.xml
+++ b/spark/interpreter/pom.xml
@@ -40,57 +40,42 @@
3.0.3
2.7
+ 4.1.19
+ 4.2.4
+ 4.1.17
+
+
+ 3.4.1
+ 3.21.12
+ 0.10.9.7
+ 2.12.17
+ 2.12
+
+ spark-${spark.version}
+
+ https://archive.apache.org/dist/spark/${spark.archive}/${spark.archive}.tgz
+
+
+ https://archive.apache.org/dist/spark/${spark.archive}/${spark.archive}-bin-without-hadoop.tgz
+
+
${spark.scala.version}
**/PySparkInterpreterMatplotlibTest.java
**/*Test.*
+ 5.7.1
+ 3.12.4
+ 1.7.0
+ 4.2.0
-
- org.apache.zeppelin
- zeppelin-display
- ${project.version}
-
-
- org.scala-lang
- scala-library
-
-
- org.scala-lang
- scala-compiler
-
-
- org.scala-lang
- scalap
-
-
-
-
-
- org.apache.zeppelin
- spark1-shims
- ${project.version}
-
-
-
- org.apache.zeppelin
- spark2-shims
- ${project.version}
-
-
org.apache.zeppelin
spark3-shims
${project.version}
-
- org.apache.zeppelin
- zeppelin-kotlin
- ${project.version}
-
-
org.apache.zeppelin
zeppelin-python
@@ -144,17 +129,34 @@
org.apache.hadoop
hadoop-client
- 2.6.0
+ ${hadoop.version}
provided
+
+ org.apache.hadoop
+ hadoop-common
+ ${hadoop.version}
+ provided
+
+
+ com.google.protobuf
+ protobuf-java
+
+
+ commons-lang
+ commons-lang
+
+
+
+
org.apache.spark
spark-hive_${spark.scala.binary.version}
${spark.version}
provided
-
+
org.apache.commons
commons-exec
@@ -202,13 +204,6 @@
-
- org.scalatest
- scalatest_${spark.scala.binary.version}
- ${scalatest.version}
- test
-
-
org.datanucleus
datanucleus-core
@@ -233,56 +228,54 @@
org.mockito
mockito-core
+ ${mockito.version}
+ test
+
+
+
+ net.jodah
+ concurrentunit
+ 0.4.4
test
- org.powermock
- powermock-api-mockito
+ com.mashape.unirest
+ unirest-java
+ 1.4.9
test
- org.powermock
- powermock-module-junit4
+ org.junit.jupiter
+ junit-jupiter-params
+ 5.7.1
test
- net.jodah
- concurrentunit
- 0.4.4
+ org.scalatest
+ scalatest_${spark.scala.binary.version}
+ ${scalatest.version}
test
- com.mashape.unirest
- unirest-java
- 1.4.9
+ org.scalacheck
+ scalacheck_${spark.scala.binary.version}
+ ${scalacheck.version}
test
+
+ com.google.guava
+ guava
+ 23.0
+
-
- maven-enforcer-plugin
-
-
- enforce
- none
-
-
-
-
-
-
- 1.7
-
-
-
-
com.googlecode.maven-download-plugin
@@ -303,21 +296,6 @@
${project.build.directory}
-
-
- download-sparkr-files
- validate
-
- wget
-
-
- 60000
- 5
- ${spark.bin.download.url}
- true
- ${project.build.directory}
-
-
@@ -345,23 +323,6 @@
maven-resources-plugin
-
- copy-sparkr-files
- generate-resources
-
- copy-resources
-
-
- ${project.build.directory}/../../../interpreter/spark/R/lib
-
-
-
- ${project.build.directory}/spark-${spark.version}-bin-without-hadoop/R/lib
-
-
-
-
-
copy-interpreter-setting
package
@@ -375,14 +336,10 @@
-
- org.scalatest
- scalatest-maven-plugin
-
-
org.apache.maven.plugins
maven-surefire-plugin
+ 3.5.1
1
false
@@ -456,6 +413,85 @@
+
+
+ maven-failsafe-plugin
+
+
+
+ integration-test
+ verify
+
+
+
+
+ -Xmx2048m
+
+
+
+
+ net.alchim31.maven
+ scala-maven-plugin
+
+
+ eclipse-add-source
+
+ add-source
+
+
+
+ scala-compile-first
+ process-resources
+
+ compile
+
+
+
+ scala-test-compile-first
+ process-test-resources
+
+ testCompile
+
+
+
+
+ ${spark.scala.version}
+
+ -unchecked
+ -deprecation
+ -feature
+ -nobootcp
+
+
+ -Xms1024m
+ -Xmx1024m
+ -XX:MaxMetaspaceSize=${MaxMetaspace}
+
+
+ -source
+ ${java.version}
+ -target
+ ${java.version}
+ -Xlint:all,-serial,-path,-options
+
+
+
+
+
+ org.scalatest
+ scalatest-maven-plugin
+
+
+ test
+
+ test
+
+
+
+
+
+ target/scala-${spark.scala.binary.version}/classes
+ target/scala-${spark.scala.binary.version}/test-classes
diff --git a/spark/interpreter/src/main/java/org/apache/zeppelin/spark/AbstractSparkScalaInterpreter.java b/spark/interpreter/src/main/java/org/apache/zeppelin/spark/AbstractSparkScalaInterpreter.java
index bf3abd8cd..78c07c9a9 100644
--- a/spark/interpreter/src/main/java/org/apache/zeppelin/spark/AbstractSparkScalaInterpreter.java
+++ b/spark/interpreter/src/main/java/org/apache/zeppelin/spark/AbstractSparkScalaInterpreter.java
@@ -17,16 +17,31 @@
package org.apache.zeppelin.spark;
+import com.google.common.collect.Lists;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.yarn.api.records.ApplicationReport;
+import org.apache.hadoop.yarn.client.api.YarnClient;
+import org.apache.hadoop.yarn.conf.YarnConfiguration;
+import org.apache.hadoop.yarn.exceptions.YarnException;
+import org.apache.hadoop.yarn.util.ConverterUtils;
+import org.apache.spark.SparkConf;
import org.apache.spark.SparkContext;
+import org.apache.spark.SparkJobInfo;
+import org.apache.spark.SparkStageInfo;
import org.apache.spark.sql.SQLContext;
-import org.apache.zeppelin.interpreter.ZeppelinContext;
-import org.apache.zeppelin.interpreter.Interpreter;
-import org.apache.zeppelin.interpreter.InterpreterContext;
-import org.apache.zeppelin.interpreter.InterpreterException;
-import org.apache.zeppelin.interpreter.InterpreterResult;
+import org.apache.spark.sql.SparkSession;
+import org.apache.zeppelin.interpreter.*;
import org.apache.zeppelin.interpreter.thrift.InterpreterCompletion;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
-import java.util.List;
+import java.io.IOException;
+import java.util.*;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.stream.Collectors;
/**
* This is bridge class which bridge the communication between java side and scala side.
@@ -34,40 +49,282 @@
*/
public abstract class AbstractSparkScalaInterpreter {
- public abstract SparkContext getSparkContext();
+ private static final Logger LOGGER = LoggerFactory.getLogger(AbstractSparkScalaInterpreter.class);
+ private static final AtomicInteger SESSION_NUM = new AtomicInteger(0);
- public abstract SQLContext getSqlContext();
+ protected SparkConf conf;
+ protected SparkContext sc;
+ protected SparkSession sparkSession;
+ protected SQLContext sqlContext;
+ protected String sparkUrl;
+ protected ZeppelinContext z;
- public abstract Object getSparkSession();
+ protected Properties properties;
+ protected List depFiles;
- public abstract String getSparkUrl();
+ public AbstractSparkScalaInterpreter(SparkConf conf,
+ Properties properties,
+ List depFiles) {
+ this.conf = conf;
+ this.properties = properties;
+ this.depFiles = depFiles;
+ }
- public abstract ZeppelinContext getZeppelinContext();
+ public SparkContext getSparkContext() {
+ return this.sc;
+ }
- public int getProgress(InterpreterContext context) throws InterpreterException {
- return getProgress(Utils.buildJobGroupId(context), context);
+ public SQLContext getSqlContext() {
+ return this.sqlContext;
}
- public abstract int getProgress(String jobGroup,
- InterpreterContext context) throws InterpreterException;
+ public SparkSession getSparkSession() {
+ return this.sparkSession;
+ }
- public void cancel(InterpreterContext context) throws InterpreterException {
- getSparkContext().cancelJobGroup(Utils.buildJobGroupId(context));
+ public String getSparkUrl() {
+ return this.sparkUrl;
}
- public Interpreter.FormType getFormType() throws InterpreterException {
- return Interpreter.FormType.SIMPLE;
+ public ZeppelinContext getZeppelinContext() {
+ return this.z;
}
- public abstract void open();
+ public AbstractSparkScalaInterpreter() {
+ }
+
+ public void open() throws InterpreterException {
+ /* Required for scoped mode.
+ * In scoped mode multiple scala compiler (repl) generates class in the same directory.
+ * Class names is not randomly generated and look like '$line12.$read$$iw$$iw'
+ * Therefore it's possible to generated class conflict(overwrite) with other repl generated
+ * class.
+ *
+ * To prevent generated class name conflict,
+ * change prefix of generated class name from each scala compiler (repl) instance.
+ *
+ * In Spark 2.x, REPL generated wrapper class name should compatible with the pattern
+ * ^(\$line(?:\d+)\.\$read)(?:\$\$iw)+$
+ *
+ * As hashCode() can return a negative integer value and the minus character '-' is invalid
+ * in a package name we change it to a numeric value '0' which still conforms to the regexp.
+ *
+ */
+ System.setProperty("scala.repl.name.line", ("$line" + this.hashCode()).replace('-', '0'));
+ SESSION_NUM.incrementAndGet();
- public abstract void close();
+ createSparkILoop();
+ createSparkContext();
+ createZeppelinContext();
+ }
- public abstract InterpreterResult interpret(String st, InterpreterContext context);
+ public void close() throws InterpreterException {
+ // delete stagingDir for yarn mode
+ if (getSparkMaster().startsWith("yarn")) {
+ YarnConfiguration hadoopConf = new YarnConfiguration();
+ Path appStagingBaseDir = null;
+ if (conf.contains("spark.yarn.stagingDir")) {
+ appStagingBaseDir = new Path(conf.get("spark.yarn.stagingDir"));
+ } else {
+ try {
+ appStagingBaseDir = FileSystem.get(hadoopConf).getHomeDirectory();
+ } catch (IOException e) {
+ LOGGER.error("Fail to get stagingBaseDir", e);
+ }
+ }
+ if (appStagingBaseDir != null) {
+ Path stagingDirPath = new Path(appStagingBaseDir, ".sparkStaging" + "/" + sc.applicationId());
+ cleanupStagingDirInternal(stagingDirPath, hadoopConf);
+ }
+ }
+
+ if (sc != null) {
+ sc.stop();
+ sc = null;
+ }
+ if (sparkSession != null) {
+ sparkSession.stop();
+ sparkSession = null;
+ }
+ sqlContext = null;
+ z = null;
+ }
+
+ public abstract void createSparkILoop() throws InterpreterException;
+
+ public abstract void createZeppelinContext() throws InterpreterException;
+
+ public void cancel(InterpreterContext context) throws InterpreterException {
+ getSparkContext().cancelJobGroup(Utils.buildJobGroupId(context));
+ }
+
+ public abstract InterpreterResult interpret(String st,
+ InterpreterContext context) throws InterpreterException;
public abstract List completion(String buf,
int cursor,
- InterpreterContext interpreterContext);
+ InterpreterContext interpreterContext) throws InterpreterException;
+
+ public abstract void bind(String name,
+ String tpe,
+ Object value,
+ List modifier);
+
+ // throw exception when fail to execute the code in scala shell, only used in initialization.
+ // not used t run user code.
+ public abstract void scalaInterpretQuietly(String code) throws InterpreterException;
public abstract ClassLoader getScalaShellClassLoader();
+
+ private List getUserFiles() {
+ return depFiles.stream()
+ .filter(f -> f.endsWith(".jar"))
+ .collect(Collectors.toList());
+ }
+
+ private void createSparkContext() throws InterpreterException {
+ SparkSession.Builder builder = SparkSession.builder().config(conf);
+ if (conf.get("spark.sql.catalogImplementation", "in-memory").equalsIgnoreCase("hive")
+ || conf.get("zeppelin.spark.useHiveContext", "false").equalsIgnoreCase("true")) {
+ boolean hiveSiteExisted =
+ Thread.currentThread().getContextClassLoader().getResource("hive-site.xml") != null;
+ if (hiveSiteExisted && hiveClassesArePresent()) {
+ sparkSession = builder.enableHiveSupport().getOrCreate();
+ LOGGER.info("Created Spark session (with Hive support)");
+ } else {
+ if (!hiveClassesArePresent()) {
+ LOGGER.warn("Hive support can not be enabled because spark is not built with hive");
+ }
+ if (!hiveSiteExisted) {
+ LOGGER.warn("Hive support can not be enabled because no hive-site.xml found");
+ }
+ sparkSession = builder.getOrCreate();
+ LOGGER.info("Created Spark session (without Hive support)");
+ }
+ } else {
+ sparkSession = builder.getOrCreate();
+ LOGGER.info("Created Spark session (without Hive support)");
+ }
+
+ sc = sparkSession.sparkContext();
+ getUserFiles().forEach(file -> sc.addFile(file));
+ if (sc.uiWebUrl().isDefined()) {
+ sparkUrl = sc.uiWebUrl().get();
+ }
+ sqlContext = sparkSession.sqlContext();
+
+ initAndSendSparkWebUrl();
+
+ bind("spark", sparkSession.getClass().getCanonicalName(), sparkSession, Lists.newArrayList("@transient"));
+ bind("sc", "org.apache.spark.SparkContext", sc, Lists.newArrayList("@transient"));
+ bind("sqlContext", "org.apache.spark.sql.SQLContext", sqlContext, Lists.newArrayList("@transient"));
+
+ scalaInterpretQuietly("import org.apache.spark.SparkContext._");
+ scalaInterpretQuietly("import spark.implicits._");
+ scalaInterpretQuietly("import sqlContext.implicits._");
+ scalaInterpretQuietly("import spark.sql");
+ scalaInterpretQuietly("import org.apache.spark.sql.functions._");
+ // print empty string otherwise the last statement's output of this method
+ // (aka. import org.apache.spark.sql.functions._) will mix with the output of user code
+ scalaInterpretQuietly("print(\"\")");
+ }
+
+ /**
+ * @return true if Hive classes can be loaded, otherwise false.
+ */
+ private boolean hiveClassesArePresent() {
+ try {
+ Class.forName("org.apache.spark.sql.hive.HiveSessionStateBuilder");
+ Class.forName("org.apache.hadoop.hive.conf.HiveConf");
+ return true;
+ } catch (ClassNotFoundException | NoClassDefFoundError e) {
+ return false;
+ }
+ }
+
+ private void initAndSendSparkWebUrl() {
+ String webUiUrl = properties.getProperty("zeppelin.spark.uiWebUrl");
+ if (!StringUtils.isBlank(webUiUrl)) {
+ this.sparkUrl = webUiUrl.replace("{{applicationId}}", sc.applicationId());
+ } else {
+ useYarnProxyURLIfNeeded();
+ }
+ InterpreterContext.get().getIntpEventClient().sendWebUrlInfo(this.sparkUrl);
+ }
+
+ private String getSparkMaster() {
+ if (conf == null) {
+ return "";
+ } else {
+ return conf.get(SparkStringConstants.MASTER_PROP_NAME,
+ SparkStringConstants.DEFAULT_MASTER_VALUE);
+ }
+ }
+
+ private void cleanupStagingDirInternal(Path stagingDirPath, Configuration hadoopConf) {
+ try {
+ FileSystem fs = stagingDirPath.getFileSystem(hadoopConf);
+ if (fs.delete(stagingDirPath, true)) {
+ LOGGER.info("Deleted staging directory " + stagingDirPath);
+ }
+ } catch (IOException e) {
+ LOGGER.warn("Failed to cleanup staging dir " + stagingDirPath, e);
+ }
+ }
+
+ private void useYarnProxyURLIfNeeded() {
+ if (Boolean.parseBoolean(properties.getProperty("spark.webui.yarn.useProxy", "false"))) {
+ if (getSparkMaster().startsWith("yarn")) {
+ String appId = sc.applicationId();
+ YarnClient yarnClient = YarnClient.createYarnClient();
+ YarnConfiguration yarnConf = new YarnConfiguration();
+ // disable timeline service as we only query yarn app here.
+ // Otherwise we may hit this kind of ERROR:
+ // java.lang.ClassNotFoundException: com.sun.jersey.api.client.config.ClientConfig
+ yarnConf.set("yarn.timeline-service.enabled", "false");
+ yarnClient.init(yarnConf);
+ yarnClient.start();
+ ApplicationReport appReport = null;
+ try {
+ appReport = yarnClient.getApplicationReport(ConverterUtils.toApplicationId(appId));
+ this.sparkUrl = appReport.getTrackingUrl();
+ } catch (YarnException | IOException e) {
+ LOGGER.error("Fail to get yarn app report", e);
+ }
+ }
+ }
+ }
+
+ public int getProgress(InterpreterContext context) throws InterpreterException {
+ String jobGroup = Utils.buildJobGroupId(context);
+ // Each paragraph has one unique jobGroup, and one paragraph may run multiple times.
+ // So only look for the first job which match the jobGroup
+ Optional jobInfoOptional = Arrays.stream(sc.statusTracker().getJobIdsForGroup(jobGroup))
+ .mapToObj(jobId -> sc.statusTracker().getJobInfo(jobId))
+ .filter(jobInfo -> jobInfo.isDefined())
+ .map(jobInfo -> jobInfo.get())
+ .findFirst();
+ if (jobInfoOptional.isPresent()) {
+ List stageInfoList = Arrays.stream(jobInfoOptional.get().stageIds())
+ .mapToObj(stageId -> sc.statusTracker().getStageInfo(stageId))
+ .filter(stageInfo -> stageInfo.isDefined())
+ .map(stageInfo -> stageInfo.get())
+ .collect(Collectors.toList());
+ int taskCount = stageInfoList.stream()
+ .map(stageInfo -> stageInfo.numTasks())
+ .collect(Collectors.summingInt(Integer::intValue));
+ int completedTaskCount = stageInfoList.stream()
+ .map(stageInfo -> stageInfo.numCompletedTasks())
+ .collect(Collectors.summingInt(Integer::intValue));
+ LOGGER.debug("Total TaskCount: " + taskCount);
+ LOGGER.debug("Completed TaskCount: " + completedTaskCount);
+ if (taskCount == 0) {
+ return 0;
+ } else {
+ return 100 * completedTaskCount / taskCount;
+ }
+ } else {
+ return 0;
+ }
+ }
}
diff --git a/spark/interpreter/src/main/java/org/apache/zeppelin/spark/KotlinSparkInterpreter.java b/spark/interpreter/src/main/java/org/apache/zeppelin/spark/KotlinSparkInterpreter.java
deleted file mode 100644
index 32de4b426..000000000
--- a/spark/interpreter/src/main/java/org/apache/zeppelin/spark/KotlinSparkInterpreter.java
+++ /dev/null
@@ -1,200 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.zeppelin.spark;
-
-import static org.apache.zeppelin.spark.Utils.buildJobDesc;
-import static org.apache.zeppelin.spark.Utils.buildJobGroupId;
-import org.apache.spark.SparkConf;
-import org.apache.spark.api.java.JavaSparkContext;
-import org.apache.spark.sql.SQLContext;
-import org.apache.spark.util.Utils;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import java.io.File;
-import java.io.IOException;
-import java.io.PrintStream;
-import java.util.Arrays;
-import java.util.Collections;
-import java.util.List;
-import java.util.Properties;
-import java.util.regex.Pattern;
-import java.util.stream.Collectors;
-import java.util.stream.Stream;
-import scala.Console;
-import org.apache.zeppelin.interpreter.ZeppelinContext;
-import org.apache.zeppelin.interpreter.Interpreter;
-import org.apache.zeppelin.interpreter.InterpreterContext;
-import org.apache.zeppelin.interpreter.InterpreterException;
-import org.apache.zeppelin.interpreter.InterpreterOutput;
-import org.apache.zeppelin.interpreter.InterpreterResult;
-import org.apache.zeppelin.interpreter.thrift.InterpreterCompletion;
-import org.apache.zeppelin.kotlin.KotlinInterpreter;
-import org.apache.zeppelin.spark.kotlin.KotlinZeppelinBindings;
-import org.apache.zeppelin.spark.kotlin.SparkKotlinReceiver;
-
-public class KotlinSparkInterpreter extends Interpreter {
- private static Logger logger = LoggerFactory.getLogger(KotlinSparkInterpreter.class);
- private static final SparkVersion KOTLIN_SPARK_SUPPORTED_VERSION = SparkVersion.SPARK_2_4_0;
-
- private InterpreterResult unsupportedMessage;
- private KotlinInterpreter interpreter;
- private SparkInterpreter sparkInterpreter;
- private ZeppelinContext z;
- private JavaSparkContext jsc;
-
- public KotlinSparkInterpreter(Properties properties) {
- super(properties);
- logger.debug("Creating KotlinSparkInterpreter");
- interpreter = new KotlinInterpreter(properties);
- }
-
- @Override
- public void open() throws InterpreterException {
- sparkInterpreter =
- getInterpreterInTheSameSessionByClassName(SparkInterpreter.class);
- jsc = sparkInterpreter.getJavaSparkContext();
-
- SparkVersion sparkVersion = SparkVersion.fromVersionString(jsc.version());
- if (sparkVersion.olderThan(KOTLIN_SPARK_SUPPORTED_VERSION)) {
- unsupportedMessage = new InterpreterResult(
- InterpreterResult.Code.ERROR,
- "Spark version is " + sparkVersion + ", only " +
- KOTLIN_SPARK_SUPPORTED_VERSION + " and newer are supported");
- }
-
- z = sparkInterpreter.getZeppelinContext();
-
- // convert Object to SQLContext explicitly, that means Kotlin Spark may not work with Spark 1.x
- SparkKotlinReceiver ctx = new SparkKotlinReceiver(
- sparkInterpreter.getSparkSession(),
- jsc,
- (SQLContext) sparkInterpreter.getSQLContext(),
- z);
-
- List classpath = sparkClasspath();
-
- String outputDir = null;
- SparkConf conf = jsc.getConf();
- if (conf != null) {
- outputDir = conf.getOption("spark.repl.class.outputDir").getOrElse(null);
- }
-
- interpreter.getKotlinReplProperties()
- .receiver(ctx)
- .classPath(classpath)
- .outputDir(outputDir)
- .codeOnLoad(KotlinZeppelinBindings.Z_SELECT_KOTLIN_SYNTAX)
- .codeOnLoad(KotlinZeppelinBindings.SPARK_UDF_IMPORTS)
- .codeOnLoad(KotlinZeppelinBindings.CAST_SPARK_SESSION);
- interpreter.open();
- }
-
- @Override
- public void close() throws InterpreterException {
- interpreter.close();
- }
-
- @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;
- }
-
- @Override
- public void cancel(InterpreterContext context) throws InterpreterException {
- if (isSparkVersionUnsupported()) {
- return;
- }
- jsc.cancelJobGroup(buildJobGroupId(context));
- interpreter.cancel(context);
- }
-
- @Override
- public FormType getFormType() throws InterpreterException {
- return interpreter.getFormType();
- }
-
- @Override
- public int getProgress(InterpreterContext context) throws InterpreterException {
- if (isSparkVersionUnsupported()) {
- return 0;
- }
- return sparkInterpreter.getProgress(context);
- }
-
- @Override
- public List completion(String buf, int cursor,
- InterpreterContext interpreterContext) throws InterpreterException {
- if (isSparkVersionUnsupported()) {
- return Collections.emptyList();
- }
- return interpreter.completion(buf, cursor, interpreterContext);
- }
-
- boolean isSparkVersionUnsupported() {
- return unsupportedMessage != null;
- }
-
- private static List sparkClasspath() {
- String sparkJars = System.getProperty("spark.jars");
- Pattern isKotlinJar = Pattern.compile("/kotlin-[a-z]*(-.*)?\\.jar");
-
- Stream addedJars = Arrays.stream(Utils.resolveURIs(sparkJars).split(","))
- .filter(s -> !s.trim().equals(""))
- .filter(s -> !isKotlinJar.matcher(s).find())
- .map(s -> {
- int p = s.indexOf(':');
- return new File(s.substring(p + 1));
- });
-
- Stream systemJars = Arrays.stream(
- System.getProperty("java.class.path").split(File.pathSeparator))
- .map(File::new);
-
- return Stream.concat(addedJars, systemJars)
- .map(file -> {
- try {
- return file.getCanonicalPath();
- } catch (IOException e) {
- return "";
- }
- })
- .collect(Collectors.toList());
- }
-}
diff --git a/spark/interpreter/src/main/java/org/apache/zeppelin/spark/PySparkInterpreter.java b/spark/interpreter/src/main/java/org/apache/zeppelin/spark/PySparkInterpreter.java
index 56d1c6fc8..7b42e9309 100644
--- a/spark/interpreter/src/main/java/org/apache/zeppelin/spark/PySparkInterpreter.java
+++ b/spark/interpreter/src/main/java/org/apache/zeppelin/spark/PySparkInterpreter.java
@@ -86,15 +86,12 @@ public void open() throws InterpreterException {
// must create spark interpreter after ClassLoader is set, otherwise the additional jars
// can not be loaded by spark repl.
this.sparkInterpreter = getInterpreterInTheSameSessionByClassName(SparkInterpreter.class);
- setProperty("zeppelin.py4j.useAuth",
- sparkInterpreter.getSparkVersion().isSecretSocketSupported() + "");
// create Python Process and JVM gateway
super.open();
} finally {
Thread.currentThread().setContextClassLoader(oldCl);
}
- // Initialize Spark in Python Process
try {
bootstrapInterpreter("python/zeppelin_pyspark.py");
} catch (IOException e) {
@@ -109,6 +106,7 @@ public void close() throws InterpreterException {
super.close();
}
+
@Override
protected ZeppelinContext createZeppelinContext() {
return sparkInterpreter.getZeppelinContext();
@@ -183,6 +181,7 @@ String getPythonExec(SparkConf sparkConf) {
return "python";
}
+ @Override
public ZeppelinContext getZeppelinContext() {
if (sparkInterpreter != null) {
return sparkInterpreter.getZeppelinContext();
@@ -224,11 +223,13 @@ public Object getSQLContext() {
}
}
- public boolean isSpark1() {
- return sparkInterpreter.getSparkVersion().getMajorVersion() == 1;
- }
-
+ // Used by PySpark
public boolean isSpark3() {
return sparkInterpreter.getSparkVersion().getMajorVersion() == 3;
}
+
+ // Used by PySpark
+ public boolean isAfterSpark33() {
+ return sparkInterpreter.getSparkVersion().newerThanEquals(SparkVersion.SPARK_3_3_0);
+ }
}
diff --git a/spark/interpreter/src/main/java/org/apache/zeppelin/spark/SparkInterpreter.java b/spark/interpreter/src/main/java/org/apache/zeppelin/spark/SparkInterpreter.java
index 500fd2842..c57fca62a 100644
--- a/spark/interpreter/src/main/java/org/apache/zeppelin/spark/SparkInterpreter.java
+++ b/spark/interpreter/src/main/java/org/apache/zeppelin/spark/SparkInterpreter.java
@@ -22,6 +22,7 @@
import org.apache.spark.SparkContext;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.sql.SQLContext;
+import org.apache.spark.sql.SparkSession;
import org.apache.zeppelin.interpreter.AbstractInterpreter;
import org.apache.zeppelin.interpreter.ZeppelinContext;
import org.apache.zeppelin.interpreter.InterpreterContext;
@@ -66,12 +67,13 @@ public class SparkInterpreter extends AbstractInterpreter {
}
private static AtomicInteger SESSION_NUM = new AtomicInteger(0);
+ private static Class innerInterpreterClazz;
private AbstractSparkScalaInterpreter innerInterpreter;
private Map innerInterpreterClassMap = new HashMap<>();
private SparkContext sc;
private JavaSparkContext jsc;
private SQLContext sqlContext;
- private Object sparkSession;
+ private SparkSession sparkSession;
private SparkVersion sparkVersion;
private String scalaVersion;
@@ -83,11 +85,11 @@ public SparkInterpreter(Properties properties) {
if (Boolean.parseBoolean(properties.getProperty("zeppelin.spark.scala.color", "true"))) {
System.setProperty("scala.color", "true");
}
+
this.enableSupportedVersionCheck = java.lang.Boolean.parseBoolean(
properties.getProperty("zeppelin.spark.enableSupportedVersionCheck", "true"));
- innerInterpreterClassMap.put("2.10", "org.apache.zeppelin.spark.SparkScala210Interpreter");
- innerInterpreterClassMap.put("2.11", "org.apache.zeppelin.spark.SparkScala211Interpreter");
innerInterpreterClassMap.put("2.12", "org.apache.zeppelin.spark.SparkScala212Interpreter");
+ innerInterpreterClassMap.put("2.13", "org.apache.zeppelin.spark.SparkScala213Interpreter");
}
@Override
@@ -143,9 +145,9 @@ public void open() throws InterpreterException {
* Load AbstractSparkScalaInterpreter based on the runtime scala version.
* Load AbstractSparkScalaInterpreter from the following location:
*
- * SparkScala210Interpreter ZEPPELIN_HOME/interpreter/spark/scala-2.10
* SparkScala211Interpreter ZEPPELIN_HOME/interpreter/spark/scala-2.11
* SparkScala212Interpreter ZEPPELIN_HOME/interpreter/spark/scala-2.12
+ * SparkScala213Interpreter ZEPPELIN_HOME/interpreter/spark/scala-2.13
*
* @param conf
* @return AbstractSparkScalaInterpreter
@@ -153,29 +155,35 @@ public void open() throws InterpreterException {
*/
private AbstractSparkScalaInterpreter loadSparkScalaInterpreter(SparkConf conf) throws Exception {
scalaVersion = extractScalaVersion(conf);
- ClassLoader scalaInterpreterClassLoader = Thread.currentThread().getContextClassLoader();
-
- String zeppelinHome = System.getenv("ZEPPELIN_HOME");
- if (zeppelinHome != null) {
- // ZEPPELIN_HOME is null in yarn-cluster mode, load it directly via current ClassLoader.
- // otherwise, load from the specific folder ZEPPELIN_HOME/interpreter/spark/scala-
-
- File scalaJarFolder = new File(zeppelinHome + "/interpreter/spark/scala-" + scalaVersion);
- List urls = new ArrayList<>();
- for (File file : scalaJarFolder.listFiles()) {
- LOGGER.debug("Add file " + file.getAbsolutePath() + " to classpath of spark scala interpreter: "
- + scalaJarFolder);
- urls.add(file.toURI().toURL());
+ // Make sure the innerInterpreter Class is loaded only once into JVM
+ // Use double lock to ensure thread safety
+ if (innerInterpreterClazz == null) {
+ synchronized (SparkInterpreter.class) {
+ if (innerInterpreterClazz == null) {
+ LOGGER.debug("innerInterpreterClazz is null, thread:{}", Thread.currentThread().getName());
+ ClassLoader scalaInterpreterClassLoader = Thread.currentThread().getContextClassLoader();
+ String zeppelinHome = System.getenv("ZEPPELIN_HOME");
+ if (zeppelinHome != null) {
+ // ZEPPELIN_HOME is null in yarn-cluster mode, load it directly via current ClassLoader.
+ // otherwise, load from the specific folder ZEPPELIN_HOME/interpreter/spark/scala-
+ File scalaJarFolder = new File(zeppelinHome + "/interpreter/spark/scala-" + scalaVersion);
+ List urls = new ArrayList<>();
+ for (File file : scalaJarFolder.listFiles()) {
+ LOGGER.debug("Add file {} to classpath of spark scala interpreter: {}", file.getAbsolutePath(),
+ scalaJarFolder);
+ urls.add(file.toURI().toURL());
+ }
+ scalaInterpreterClassLoader = new URLClassLoader(urls.toArray(new URL[0]),
+ Thread.currentThread().getContextClassLoader());
+ }
+ String innerIntpClassName = innerInterpreterClassMap.get(scalaVersion);
+ innerInterpreterClazz = scalaInterpreterClassLoader.loadClass(innerIntpClassName);
+ }
}
- scalaInterpreterClassLoader = new URLClassLoader(urls.toArray(new URL[0]),
- Thread.currentThread().getContextClassLoader());
}
-
- String innerIntpClassName = innerInterpreterClassMap.get(scalaVersion);
- Class clazz = scalaInterpreterClassLoader.loadClass(innerIntpClassName);
return (AbstractSparkScalaInterpreter)
- clazz.getConstructor(SparkConf.class, List.class, Properties.class, InterpreterGroup.class, URLClassLoader.class, File.class)
- .newInstance(conf, getDependencyFiles(), getProperties(), getInterpreterGroup(), scalaInterpreterClassLoader, scalaShellOutputDir);
+ innerInterpreterClazz.getConstructor(SparkConf.class, List.class, Properties.class, InterpreterGroup.class, URLClassLoader.class, File.class)
+ .newInstance(conf, getDependencyFiles(), getProperties(), getInterpreterGroup(), innerInterpreterClazz.getClassLoader(), scalaShellOutputDir);
}
@Override
@@ -183,8 +191,9 @@ public void close() throws InterpreterException {
LOGGER.info("Close SparkInterpreter");
if (SESSION_NUM.decrementAndGet() == 0 && innerInterpreter != null) {
innerInterpreter.close();
- innerInterpreter = null;
+ innerInterpreterClazz = null;
}
+ innerInterpreter = null;
}
@Override
@@ -218,9 +227,10 @@ public FormType getFormType() {
@Override
public int getProgress(InterpreterContext context) throws InterpreterException {
- return innerInterpreter.getProgress(Utils.buildJobGroupId(context), context);
+ return innerInterpreter.getProgress(context);
}
+ @Override
public ZeppelinContext getZeppelinContext() {
if (this.innerInterpreter == null) {
throw new RuntimeException("innerInterpreterContext is null");
@@ -232,14 +242,7 @@ public SparkContext getSparkContext() {
return this.sc;
}
- /**
- * Must use Object, because the its api signature in Spark 1.x is different from
- * that of Spark 2.x.
- * e.g. SqlContext.sql(sql) return different type.
- *
- * @return
- */
- public Object getSQLContext() {
+ public SQLContext getSQLContext() {
return sqlContext;
}
@@ -247,7 +250,7 @@ public JavaSparkContext getJavaSparkContext() {
return this.jsc;
}
- public Object getSparkSession() {
+ public SparkSession getSparkSession() {
return sparkSession;
}
@@ -264,29 +267,28 @@ private String extractScalaVersion(SparkConf conf) throws InterpreterException {
if (conf.contains("zeppelin.spark.scala.version")) {
scalaVersionString = conf.get("zeppelin.spark.scala.version");
} else {
- scalaVersionString = scala.util.Properties.versionNumberString();
+ scalaVersionString = scala.util.Properties.versionString();
}
- LOGGER.info("Using Scala: " + scalaVersionString);
+ LOGGER.info("Using Scala: {}", scalaVersionString);
if (StringUtils.isEmpty(scalaVersionString)) {
throw new InterpreterException("Scala Version is empty");
- } else if (scalaVersionString.startsWith("2.10")) {
- return "2.10";
- } else if (scalaVersionString.startsWith("2.11")) {
- return "2.11";
- } else if (scalaVersionString.startsWith("2.12")) {
+ } else if (scalaVersionString.contains("2.12")) {
return "2.12";
+ } else if (scalaVersionString.contains("2.13")) {
+ return "2.13";
} else {
throw new InterpreterException("Unsupported scala version: " + scalaVersionString);
}
}
- public boolean isScala212() throws InterpreterException {
+
+ public boolean isScala212() {
return scalaVersion.equals("2.12");
}
- public boolean isScala210() throws InterpreterException {
- return scalaVersion.equals("2.10");
+ public boolean isScala213() {
+ return scalaVersion.equals("2.13");
}
private List getDependencyFiles() throws InterpreterException {
@@ -314,4 +316,9 @@ public ClassLoader getScalaShellClassLoader() {
public boolean isUnsupportedSparkVersion() {
return enableSupportedVersionCheck && sparkVersion.isUnsupportedVersion();
}
+
+ public AbstractSparkScalaInterpreter getInnerInterpreter() {
+ return innerInterpreter;
+ }
+
}
diff --git a/spark/interpreter/src/main/java/org/apache/zeppelin/spark/SparkSqlInterpreter.java b/spark/interpreter/src/main/java/org/apache/zeppelin/spark/SparkSqlInterpreter.java
index 6c06399fb..b2b9a69d3 100644
--- a/spark/interpreter/src/main/java/org/apache/zeppelin/spark/SparkSqlInterpreter.java
+++ b/spark/interpreter/src/main/java/org/apache/zeppelin/spark/SparkSqlInterpreter.java
@@ -20,6 +20,7 @@
import org.apache.commons.lang3.exception.ExceptionUtils;
import org.apache.spark.SparkContext;
import org.apache.spark.sql.AnalysisException;
+import org.apache.spark.sql.SQLContext;
import org.apache.zeppelin.interpreter.AbstractInterpreter;
import org.apache.zeppelin.interpreter.ZeppelinContext;
import org.apache.zeppelin.interpreter.InterpreterContext;
@@ -57,7 +58,7 @@ public void open() throws InterpreterException {
this.sqlSplitter = new SqlSplitter();
}
- public boolean concurrentSQL() {
+ private boolean concurrentSQL() {
return Boolean.parseBoolean(getProperty("zeppelin.spark.concurrentSQL"));
}
@@ -83,7 +84,7 @@ public InterpreterResult internalInterpret(String st, InterpreterContext context
}
Utils.printDeprecateMessage(sparkInterpreter.getSparkVersion(), context, properties);
sparkInterpreter.getZeppelinContext().setInterpreterContext(context);
- Object sqlContext = sparkInterpreter.getSQLContext();
+ SQLContext sqlContext = sparkInterpreter.getSQLContext();
SparkContext sc = sparkInterpreter.getSparkContext();
List sqls = sqlSplitter.splitSql(st);
@@ -95,15 +96,11 @@ public InterpreterResult internalInterpret(String st, InterpreterContext context
String curSql = null;
ClassLoader originalClassLoader = Thread.currentThread().getContextClassLoader();
try {
- if (!sparkInterpreter.isScala212()) {
- // TODO(zjffdu) scala 2.12 still doesn't work for codegen (ZEPPELIN-4627)
Thread.currentThread().setContextClassLoader(sparkInterpreter.getScalaShellClassLoader());
- }
- Method method = sqlContext.getClass().getMethod("sql", String.class);
for (String sql : sqls) {
curSql = sql;
String result = sparkInterpreter.getZeppelinContext()
- .showData(method.invoke(sqlContext, sql), maxResult);
+ .showData(sqlContext.sql(sql), maxResult);
context.out.write(result);
}
context.out.flush();
@@ -142,9 +139,7 @@ public InterpreterResult internalInterpret(String st, InterpreterContext context
}
} finally {
sc.clearJobGroup();
- if (!sparkInterpreter.isScala212()) {
- Thread.currentThread().setContextClassLoader(originalClassLoader);
- }
+ Thread.currentThread().setContextClassLoader(originalClassLoader);
}
return new InterpreterResult(Code.SUCCESS);
@@ -161,7 +156,6 @@ public FormType getFormType() {
return FormType.SIMPLE;
}
-
@Override
public int getProgress(InterpreterContext context) throws InterpreterException {
return sparkInterpreter.getProgress(context);
diff --git a/spark/interpreter/src/main/java/org/apache/zeppelin/spark/Utils.java b/spark/interpreter/src/main/java/org/apache/zeppelin/spark/Utils.java
index ea8fb8b4d..b325f40ee 100644
--- a/spark/interpreter/src/main/java/org/apache/zeppelin/spark/Utils.java
+++ b/spark/interpreter/src/main/java/org/apache/zeppelin/spark/Utils.java
@@ -27,83 +27,12 @@
import java.lang.reflect.Constructor;
import java.lang.reflect.InvocationTargetException;
import java.util.Properties;
-import java.util.regex.Matcher;
-import java.util.regex.Pattern;
/**
* Utility and helper functions for the Spark Interpreter
*/
class Utils {
- public static Logger logger = LoggerFactory.getLogger(Utils.class);
- public static String DEPRRECATED_MESSAGE =
- "%html Spark lower than 2.2 is deprecated, " +
- "if you don't want to see this message, please set " +
- "zeppelin.spark.deprecateMsg.show to false.";
-
- static Object invokeMethod(Object o, String name) {
- return invokeMethod(o, name, new Class[]{}, new Object[]{});
- }
-
- static Object invokeMethod(Object o, String name, Class[] argTypes, Object[] params) {
- try {
- return o.getClass().getMethod(name, argTypes).invoke(o, params);
- } catch (NoSuchMethodException | IllegalAccessException | InvocationTargetException e) {
- logger.error(e.getMessage(), e);
- }
- return null;
- }
-
- static Object invokeStaticMethod(Class c, String name, Class[] argTypes, Object[] params) {
- try {
- return c.getMethod(name, argTypes).invoke(null, params);
- } catch (NoSuchMethodException | InvocationTargetException | IllegalAccessException e) {
- logger.error(e.getMessage(), e);
- }
- return null;
- }
-
- static Object invokeStaticMethod(Class c, String name) {
- return invokeStaticMethod(c, name, new Class[]{}, new Object[]{});
- }
-
- static Class findClass(String name) {
- return findClass(name, false);
- }
-
- static Class findClass(String name, boolean silence) {
- try {
- return Class.forName(name);
- } catch (ClassNotFoundException e) {
- if (!silence) {
- logger.error(e.getMessage(), e);
- }
- return null;
- }
- }
-
- static Object instantiateClass(String name, Class[] argTypes, Object[] params) {
- try {
- Constructor constructor = Utils.class.getClassLoader()
- .loadClass(name).getConstructor(argTypes);
- return constructor.newInstance(params);
- } catch (NoSuchMethodException | ClassNotFoundException | IllegalAccessException |
- InstantiationException | InvocationTargetException e) {
- logger.error(e.getMessage(), e);
- }
- return null;
- }
-
- // function works after intp is initialized
- static boolean isScala2_10() {
- try {
- Class.forName("org.apache.spark.repl.SparkIMain");
- return true;
- } catch (ClassNotFoundException e) {
- return false;
- } catch (IncompatibleClassChangeError e) {
- return false;
- }
- }
+ private static final Logger LOGGER = LoggerFactory.getLogger(Utils.class);
public static String buildJobGroupId(InterpreterContext context) {
String uName = "anonymous";
@@ -129,18 +58,10 @@ public static String getUserName(AuthenticationInfo info) {
}
public static void printDeprecateMessage(SparkVersion sparkVersion,
- InterpreterContext context,
- Properties properties) throws InterpreterException {
+ InterpreterContext context,
+ Properties properties) throws InterpreterException {
context.out.clear();
- if (sparkVersion.olderThan(SparkVersion.SPARK_2_2_0)
- && Boolean.parseBoolean(
- properties.getProperty("zeppelin.spark.deprecatedMsg.show", "true"))) {
- try {
- context.out.write(DEPRRECATED_MESSAGE);
- context.out.write("%text ");
- } catch (IOException e) {
- throw new InterpreterException(e);
- }
- }
+ // print deprecated message only when zeppelin.spark.deprecatedMsg.show is true and
+ // sparkVersion meets the certain requirements
}
}
diff --git a/spark/interpreter/src/main/java/org/apache/zeppelin/spark/kotlin/KotlinZeppelinBindings.java b/spark/interpreter/src/main/java/org/apache/zeppelin/spark/kotlin/KotlinZeppelinBindings.java
deleted file mode 100644
index f315838d8..000000000
--- a/spark/interpreter/src/main/java/org/apache/zeppelin/spark/kotlin/KotlinZeppelinBindings.java
+++ /dev/null
@@ -1,52 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.zeppelin.spark.kotlin;
-
-/**
- * Pre-executed code on KotlinSparkInterpreter opening.
- */
-public class KotlinZeppelinBindings {
-
- //Simpler Kotlin syntax for z.select
- public static final String Z_SELECT_KOTLIN_SYNTAX =
- "import org.apache.zeppelin.display.ui.OptionInput.ParamOption\n" +
- "import org.apache.zeppelin.interpreter.ZeppelinContext\n" +
- "\n" +
- "fun ZeppelinContext.select(name: String, defaultValue: Any?, " +
- "options: List>): Any? {\n" +
- " return select(name, defaultValue, " +
- "options.map{ ParamOption(it.first, it.second) }.toTypedArray())\n" +
- "}\n" +
- "\n" +
- "fun ZeppelinContext.select(name: String, options: List>): Any? {\n" +
- " return select(name, \"\", options)\n" +
- "}";
-
- /**
- * Automatic imports for Spark SQL UDFs.
- */
- public static final String SPARK_UDF_IMPORTS =
- "import org.apache.spark.sql.types.DataTypes\n" +
- "import org.apache.spark.sql.functions.*\n" +
- "import org.apache.spark.sql.expressions.UserDefinedFunction\n" +
- "import org.apache.spark.sql.api.java.*";
-
- public static final String CAST_SPARK_SESSION = "" +
- "import org.apache.spark.sql.SparkSession\n" +
- "val spark = _sparkObject as SparkSession";
-}
diff --git a/spark/interpreter/src/main/java/org/apache/zeppelin/spark/kotlin/SparkKotlinReceiver.java b/spark/interpreter/src/main/java/org/apache/zeppelin/spark/kotlin/SparkKotlinReceiver.java
deleted file mode 100644
index 157989118..000000000
--- a/spark/interpreter/src/main/java/org/apache/zeppelin/spark/kotlin/SparkKotlinReceiver.java
+++ /dev/null
@@ -1,43 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.zeppelin.spark.kotlin;
-
-import org.apache.spark.api.java.JavaSparkContext;
-import org.apache.spark.sql.SQLContext;
-import org.apache.zeppelin.interpreter.ZeppelinContext;
-import org.apache.zeppelin.kotlin.context.KotlinReceiver;
-
-/**
- * Implicit receiver for Kotlin REPL with Spark's context (see KotlinReceiver for more details)
- */
-public class SparkKotlinReceiver extends KotlinReceiver {
- public final Object _sparkObject;
- public final JavaSparkContext sc;
- public final SQLContext sqlContext;
- public final ZeppelinContext z;
-
- public SparkKotlinReceiver(Object spark,
- JavaSparkContext sc,
- SQLContext sqlContext,
- ZeppelinContext z) {
- this._sparkObject = spark;
- this.sc = sc;
- this.sqlContext = sqlContext;
- this.z = z;
- }
-}
diff --git a/spark/interpreter/src/main/resources/interpreter-setting.json b/spark/interpreter/src/main/resources/interpreter-setting.json
index 262bc235e..a32650f3c 100644
--- a/spark/interpreter/src/main/resources/interpreter-setting.json
+++ b/spark/interpreter/src/main/resources/interpreter-setting.json
@@ -244,39 +244,5 @@
"completionKey": "TAB",
"completionSupport": true
}
- },
- {
- "group": "spark",
- "name": "kotlin",
- "className": "org.apache.zeppelin.spark.KotlinSparkInterpreter",
- "properties": {
- "zeppelin.spark.printREPLOutput": {
- "envName": null,
- "propertyName": "zeppelin.spark.printREPLOutput",
- "defaultValue": true,
- "description": "Print REPL output",
- "type": "checkbox"
- },
- "zeppelin.spark.maxResult": {
- "envName": null,
- "propertyName": "zeppelin.kotlin.maxResult",
- "defaultValue": "1000",
- "description": "Max number of result to display.",
- "type": "number"
- },
- "zeppelin.kotlin.shortenTypes": {
- "envName": null,
- "propertyName": "zeppelin.kotlin.shortenTypes",
- "defaultValue": true,
- "description": "Show short types instead of full, e.g. List or kotlin.collections.List",
- "type": "checkbox"
- }
- },
- "editor": {
- "language": "kotlin",
- "editOnDblClick": false,
- "completionKey": "TAB",
- "completionSupport": false
- }
}
]
diff --git a/spark/interpreter/src/main/resources/python/zeppelin_pyspark.py b/spark/interpreter/src/main/resources/python/zeppelin_pyspark.py
index 2038c14ac..a77c38388 100644
--- a/spark/interpreter/src/main/resources/python/zeppelin_pyspark.py
+++ b/spark/interpreter/src/main/resources/python/zeppelin_pyspark.py
@@ -48,14 +48,13 @@
conf = SparkConf(_jvm = gateway.jvm, _jconf = jconf)
sc = _zsc_ = SparkContext(jsc=jsc, gateway=gateway, conf=conf)
-if not intp.isSpark1():
- from pyspark.sql import SparkSession
- spark = __zSpark__ = SparkSession(sc, intp.getSparkSession())
- sqlc = __zSqlc__ = __zSpark__._wrapped
+from pyspark.sql import SparkSession
+from pyspark.sql import SQLContext
+spark = __zSpark__ = SparkSession(sc, intp.getSparkSession())
+if intp.isAfterSpark33():
+ sqlContext = sqlc = __zSqlc__ = SQLContext._get_or_create(sc)
else:
- sqlc = __zSqlc__ = SQLContext(sparkContext=sc, sqlContext=intp.getSQLContext())
-
-sqlContext = __zSqlc__
+ sqlContext = sqlc = __zSqlc__ = __zSpark__._wrapped
from zeppelin_context import PyZeppelinContext
diff --git a/spark/interpreter/src/test/java/org/apache/zeppelin/spark/KotlinSparkInterpreterTest.java b/spark/interpreter/src/test/java/org/apache/zeppelin/spark/KotlinSparkInterpreterTest.java
deleted file mode 100644
index 1c14c0978..000000000
--- a/spark/interpreter/src/test/java/org/apache/zeppelin/spark/KotlinSparkInterpreterTest.java
+++ /dev/null
@@ -1,246 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.zeppelin.spark;
-
-import static org.apache.zeppelin.interpreter.InterpreterResult.Code.ERROR;
-import static org.apache.zeppelin.interpreter.InterpreterResult.Code.SUCCESS;
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertTrue;
-import static org.mockito.Mockito.mock;
-import org.junit.AfterClass;
-import org.junit.Assert;
-import org.junit.Assume;
-import org.junit.Before;
-import org.junit.BeforeClass;
-import org.junit.ClassRule;
-import org.junit.Rule;
-import org.junit.Test;
-import org.junit.rules.ExpectedException;
-import org.junit.rules.TemporaryFolder;
-import java.io.IOException;
-import java.nio.file.Files;
-import java.nio.file.Path;
-import java.nio.file.Paths;
-import java.util.LinkedList;
-import java.util.Properties;
-import org.apache.zeppelin.display.AngularObjectRegistry;
-import org.apache.zeppelin.display.ui.TextBox;
-import org.apache.zeppelin.interpreter.Interpreter;
-import org.apache.zeppelin.interpreter.InterpreterContext;
-import org.apache.zeppelin.interpreter.InterpreterException;
-import org.apache.zeppelin.interpreter.InterpreterGroup;
-import org.apache.zeppelin.interpreter.InterpreterOutput;
-import org.apache.zeppelin.interpreter.InterpreterOutputListener;
-import org.apache.zeppelin.interpreter.InterpreterResult;
-import org.apache.zeppelin.interpreter.InterpreterResultMessageOutput;
-import org.apache.zeppelin.interpreter.remote.RemoteInterpreterEventClient;
-import org.apache.zeppelin.resource.LocalResourcePool;
-
-public class KotlinSparkInterpreterTest {
-
- @ClassRule
- public static TemporaryFolder tmpDir = new TemporaryFolder();
-
- @Rule
- public ExpectedException exceptionRule = ExpectedException.none();
-
- private static SparkInterpreter repl;
- private static InterpreterGroup intpGroup;
- private static InterpreterContext context;
- private static KotlinSparkInterpreter interpreter;
- private static String output;
- private static boolean sparkSupported;
-
- public static Properties getSparkTestProperties(TemporaryFolder tmpDir) throws IOException {
- Properties p = new Properties();
- p.setProperty(SparkStringConstants.MASTER_PROP_NAME, "local[*]");
- p.setProperty(SparkStringConstants.APP_NAME_PROP_NAME, "Zeppelin Test");
- p.setProperty("zeppelin.spark.useHiveContext", "true");
- p.setProperty("zeppelin.spark.maxResult", "1000");
- p.setProperty("zeppelin.spark.importImplicit", "true");
- p.setProperty("zeppelin.dep.localrepo", tmpDir.newFolder().getAbsolutePath());
- p.setProperty("zeppelin.spark.property_1", "value_1");
- return p;
- }
-
- private static void testCodeForResult(String code, String expected) throws Exception {
- InterpreterResult result = interpreter.interpret(code, context);
-
- String value;
- if (result.message().isEmpty()) {
- value = "";
- } else {
- String message = result.message().get(0).getData().trim();
- // "res0 : kotlin.Int = 1" -> "kotlin.Int = 1"
- value = message.substring(message.indexOf(':') + 2);
- }
-
- assertEquals(SUCCESS, result.code());
- assertEquals(expected, value);
- }
-
- @BeforeClass
- public static void setUp() throws Exception {
- intpGroup = new InterpreterGroup();
- context = InterpreterContext.builder()
- .setNoteId("noteId")
- .setParagraphId("paragraphId")
- .setParagraphTitle("title")
- .setAngularObjectRegistry(new AngularObjectRegistry(intpGroup.getId(), null))
- .setResourcePool(new LocalResourcePool("id"))
- .setInterpreterOut(new InterpreterOutput())
- .setIntpEventClient(mock(RemoteInterpreterEventClient.class))
- .build();
- context.out = new InterpreterOutput(
- new InterpreterOutputListener() {
- @Override
- public void onUpdateAll(InterpreterOutput out) {
-
- }
-
- @Override
- public void onAppend(int index, InterpreterResultMessageOutput out, byte[] line) {
- try {
- output = out.toInterpreterResultMessage().getData();
- } catch (IOException e) {
- e.printStackTrace();
- }
- }
-
- @Override
- public void onUpdate(int index, InterpreterResultMessageOutput out) {
-
- }
- });
-
- InterpreterContext.set(context);
-
- intpGroup.put("note", new LinkedList());
-
- Properties properties = getSparkTestProperties(tmpDir);
- repl = new SparkInterpreter(properties);
- repl.setInterpreterGroup(intpGroup);
- intpGroup.get("note").add(repl);
- repl.open();
- repl.interpret("sc", context);
-
- interpreter = new KotlinSparkInterpreter(properties);
- interpreter.setInterpreterGroup(intpGroup);
- intpGroup.get("note").add(interpreter);
- try {
- interpreter.open();
- sparkSupported = true;
- } catch (UnsupportedClassVersionError e) {
- sparkSupported = false;
- }
- }
-
- @AfterClass
- public static void tearDown() throws InterpreterException {
- repl.close();
- }
-
- @Before
- public void expectUnsupportedError() {
- if (!sparkSupported) {
- exceptionRule.expect(UnsupportedClassVersionError.class);
- }
- Assume.assumeFalse("Spark version should be >= 2.4.", interpreter.isSparkVersionUnsupported());
- }
-
- @Test
- public void simpleKotlinTest() throws Exception {
- testCodeForResult("1 + 1", "Int = 2");
- }
-
- @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 testCancel() throws Exception {
- Thread t = new Thread(() -> {
- try {
- InterpreterResult result = interpreter.interpret(
- "spark.range(10).foreach { Thread.sleep(1000) }", context);
- assertEquals(ERROR, result.code());
- assertTrue(result.message().get(0).getData().trim().contains("cancelled"));
- } catch (UnsupportedClassVersionError e) {
- if (sparkSupported) {
- Assert.fail(e.getMessage());
- }
- } catch (InterpreterException e) {
- Assert.fail(e.getMessage());
- }
- });
- t.start();
- Thread.sleep(1000);
- interpreter.cancel(context);
- }
-
- @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());
- }
-}
diff --git a/spark/interpreter/src/test/java/org/apache/zeppelin/spark/PySparkInterpreterMatplotlibTest.java b/spark/interpreter/src/test/java/org/apache/zeppelin/spark/PySparkInterpreterMatplotlibTest.java
index 24e8afe45..d842f23ba 100644
--- a/spark/interpreter/src/test/java/org/apache/zeppelin/spark/PySparkInterpreterMatplotlibTest.java
+++ b/spark/interpreter/src/test/java/org/apache/zeppelin/spark/PySparkInterpreterMatplotlibTest.java
@@ -27,31 +27,31 @@
import org.apache.zeppelin.interpreter.InterpreterResult.Type;
import org.apache.zeppelin.interpreter.InterpreterResultMessage;
import org.apache.zeppelin.interpreter.remote.RemoteInterpreterEventClient;
-import org.junit.AfterClass;
-import org.junit.BeforeClass;
-import org.junit.ClassRule;
-import org.junit.FixMethodOrder;
-import org.junit.Test;
-import org.junit.rules.TemporaryFolder;
-import org.junit.runners.MethodSorters;
+import org.junit.jupiter.api.TestMethodOrder;
+import org.junit.jupiter.api.io.TempDir;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.io.File;
import java.io.IOException;
import java.util.LinkedList;
import java.util.List;
import java.util.Properties;
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertNotSame;
-import static org.junit.Assert.assertTrue;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.MethodOrderer;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.Mockito.mock;
-@FixMethodOrder(MethodSorters.NAME_ASCENDING)
+@TestMethodOrder(MethodOrderer.MethodName.class)
public class PySparkInterpreterMatplotlibTest {
- @ClassRule
- public static TemporaryFolder tmpDir = new TemporaryFolder();
+ @TempDir
+ static File tmpDir;
static SparkInterpreter sparkInterpreter;
static PySparkInterpreter pyspark;
@@ -99,7 +99,7 @@ private static Properties getPySparkTestProperties() throws IOException {
p.setProperty("zeppelin.spark.maxResult", "1000");
p.setProperty("zeppelin.spark.importImplicit", "true");
p.setProperty("zeppelin.pyspark.python", "python");
- p.setProperty("zeppelin.dep.localrepo", tmpDir.newFolder().getAbsolutePath());
+ p.setProperty("zeppelin.dep.localrepo", tmpDir.getAbsolutePath());
p.setProperty("zeppelin.pyspark.useIPython", "false");
p.setProperty("zeppelin.python.gatewayserver_address", "127.0.0.1");
p.setProperty("zeppelin.spark.deprecatedMsg.show", "false");
@@ -120,7 +120,7 @@ public static int getSparkVersionNumber() {
return version;
}
- @BeforeClass
+ @BeforeAll
public static void setUp() throws Exception {
intpGroup = new InterpreterGroup();
intpGroup.put("note", new LinkedList());
@@ -143,25 +143,25 @@ public static void setUp() throws Exception {
pyspark.open();
}
- @AfterClass
+ @AfterAll
public static void tearDown() throws InterpreterException {
pyspark.close();
sparkInterpreter.close();
}
@Test
- public void dependenciesAreInstalled() throws InterpreterException {
+ void dependenciesAreInstalled() throws InterpreterException {
// matplotlib
InterpreterResult ret = pyspark.interpret("import matplotlib", context);
- assertEquals(ret.message().toString(), InterpreterResult.Code.SUCCESS, ret.code());
+ assertEquals(InterpreterResult.Code.SUCCESS, ret.code(), ret.message().toString());
// inline backend
ret = pyspark.interpret("import backend_zinline", context);
- assertEquals(ret.message().toString(), InterpreterResult.Code.SUCCESS, ret.code());
+ assertEquals(InterpreterResult.Code.SUCCESS, ret.code(), ret.message().toString());
}
@Test
- public void showPlot() throws InterpreterException {
+ void showPlot() throws InterpreterException {
// Simple plot test
InterpreterResult ret;
ret = pyspark.interpret("import matplotlib.pyplot as plt", context);
@@ -170,15 +170,15 @@ public void showPlot() throws InterpreterException {
ret = pyspark.interpret("plt.plot([1, 2, 3])", context);
ret = pyspark.interpret("plt.show()", context);
- assertEquals(ret.message().toString(), InterpreterResult.Code.SUCCESS, ret.code());
- assertEquals(ret.message().toString(), Type.HTML, ret.message().get(0).getType());
+ assertEquals(InterpreterResult.Code.SUCCESS, ret.code(), ret.message().toString());
+ assertEquals(Type.HTML, ret.message().get(0).getType(), ret.message().toString());
assertTrue(ret.message().get(0).getData().contains("data:image/png;base64"));
assertTrue(ret.message().get(0).getData().contains(""));
}
@Test
// Test for when configuration is set to auto-close figures after show().
- public void testClose() throws InterpreterException {
+ void testClose() throws InterpreterException {
InterpreterResult ret;
InterpreterResult ret1;
InterpreterResult ret2;
@@ -205,7 +205,7 @@ public void testClose() throws InterpreterException {
@Test
// Test for when configuration is set to not auto-close figures after show().
- public void testNoClose() throws InterpreterException {
+ void testNoClose() throws InterpreterException {
InterpreterResult ret;
InterpreterResult ret1;
InterpreterResult ret2;
@@ -220,7 +220,7 @@ public void testNoClose() throws InterpreterException {
// of FigureManager, causing show() to set the output
// type to HTML even though the figure is inactive.
ret = pyspark.interpret("plt.show()", context);
- assertEquals(ret.message().toString(), InterpreterResult.Code.SUCCESS, ret.code());
+ assertEquals(InterpreterResult.Code.SUCCESS, ret.code(), ret.message().toString());
// Now test that plot can be reshown if it is updated. It should be
// different from the previous one because it will plot the same line
@@ -232,15 +232,15 @@ public void testNoClose() throws InterpreterException {
@Test
// Test angular mode
- public void testAngular() throws InterpreterException {
+ void testAngular() throws InterpreterException {
InterpreterResult ret;
ret = pyspark.interpret("import matplotlib.pyplot as plt", context);
ret = pyspark.interpret("plt.close()", context);
ret = pyspark.interpret("z.configure_mpl(interactive=False, close=False, angular=True)", context);
ret = pyspark.interpret("plt.plot([1, 2, 3])", context);
ret = pyspark.interpret("plt.show()", context);
- assertEquals(ret.message().toString(), InterpreterResult.Code.SUCCESS, ret.code());
- assertEquals(ret.message().toString(), Type.ANGULAR, ret.message().get(0).getType());
+ assertEquals(InterpreterResult.Code.SUCCESS, ret.code(), ret.message().toString());
+ assertEquals(Type.ANGULAR, ret.message().get(0).getType(), ret.message().toString());
// Check if the figure data is in the Angular Object Registry
AngularObjectRegistry registry = context.getAngularObjectRegistry();
diff --git a/spark/interpreter/src/test/java/org/apache/zeppelin/spark/PySparkInterpreterTest.java b/spark/interpreter/src/test/java/org/apache/zeppelin/spark/PySparkInterpreterTest.java
index fa7ace2e9..628812164 100644
--- a/spark/interpreter/src/test/java/org/apache/zeppelin/spark/PySparkInterpreterTest.java
+++ b/spark/interpreter/src/test/java/org/apache/zeppelin/spark/PySparkInterpreterTest.java
@@ -27,23 +27,28 @@
import org.apache.zeppelin.interpreter.LazyOpenInterpreter;
import org.apache.zeppelin.interpreter.remote.RemoteInterpreterEventClient;
import org.apache.zeppelin.python.PythonInterpreterTest;
-import org.junit.Test;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Disabled;
+import org.junit.jupiter.api.Test;
import java.io.IOException;
import java.nio.file.Files;
import java.util.LinkedList;
import java.util.Properties;
-import static org.junit.Assert.assertTrue;
-import static org.junit.Assert.fail;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.junit.jupiter.api.Assertions.fail;
import static org.mockito.Mockito.mock;
-public class PySparkInterpreterTest extends PythonInterpreterTest {
+@Disabled(value="Won't build because it depends on Spark212 being available, setup fails")
+class PySparkInterpreterTest extends PythonInterpreterTest {
private RemoteInterpreterEventClient mockRemoteEventClient = mock(RemoteInterpreterEventClient.class);
@Override
+ @BeforeEach
public void setUp() throws InterpreterException {
Properties properties = new Properties();
properties.setProperty(SparkStringConstants.MASTER_PROP_NAME, "local");
@@ -76,7 +81,6 @@ public void setUp() throws InterpreterException {
intpGroup.get("note").add(sparkInterpreter);
sparkInterpreter.setInterpreterGroup(intpGroup);
-
interpreter = new LazyOpenInterpreter(new PySparkInterpreter(properties));
intpGroup.get("note").add(interpreter);
interpreter.setInterpreterGroup(intpGroup);
@@ -85,15 +89,13 @@ public void setUp() throws InterpreterException {
}
@Override
+ @AfterEach
public void tearDown() throws InterpreterException {
intpGroup.close();
intpGroup = null;
interpreter = null;
}
- @Test
- public void testPySpark() throws InterruptedException, InterpreterException, IOException {
- }
@Override
@Test
@@ -126,7 +128,7 @@ public void testFailtoLaunchPythonProcess() throws InterpreterException {
fail("Should fail to open PySparkInterpreter");
} catch (InterpreterException e) {
String stacktrace = ExceptionUtils.getStackTrace(e);
- assertTrue(stacktrace, stacktrace.contains("No such file or directory"));
+ assertTrue(stacktrace.contains("No such file or directory"), stacktrace);
}
}
diff --git a/spark/interpreter/src/test/java/org/apache/zeppelin/spark/SparkInterpreterTest.java b/spark/interpreter/src/test/java/org/apache/zeppelin/spark/SparkInterpreterTest.java
index c750ea9b8..0ea519eb1 100644
--- a/spark/interpreter/src/test/java/org/apache/zeppelin/spark/SparkInterpreterTest.java
+++ b/spark/interpreter/src/test/java/org/apache/zeppelin/spark/SparkInterpreterTest.java
@@ -17,6 +17,7 @@
package org.apache.zeppelin.spark;
+import org.apache.commons.lang3.StringUtils;
import org.apache.zeppelin.display.AngularObjectRegistry;
import org.apache.zeppelin.display.ui.CheckBox;
import org.apache.zeppelin.display.ui.Password;
@@ -31,9 +32,10 @@
import org.apache.zeppelin.interpreter.InterpreterResultMessageOutput;
import org.apache.zeppelin.interpreter.remote.RemoteInterpreterEventClient;
import org.apache.zeppelin.interpreter.thrift.InterpreterCompletion;
-import org.junit.After;
-import org.junit.Before;
-import org.junit.Test;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Disabled;
+import org.junit.jupiter.api.Test;
import org.mockito.ArgumentCaptor;
import java.io.IOException;
@@ -41,8 +43,8 @@
import java.util.Map;
import java.util.Properties;
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertTrue;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.Matchers.any;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
@@ -50,7 +52,7 @@
import static org.mockito.Mockito.verify;
-public class SparkInterpreterTest {
+class SparkInterpreterTest {
private SparkInterpreter interpreter;
@@ -61,13 +63,14 @@ public class SparkInterpreterTest {
private RemoteInterpreterEventClient mockRemoteEventClient;
- @Before
+ @BeforeEach
public void setUp() {
mockRemoteEventClient = mock(RemoteInterpreterEventClient.class);
}
@Test
- public void testSparkInterpreter() throws IOException, InterruptedException, InterpreterException {
+ @Disabled(value="Won't build because it depends on Spark212 being available")
+ void testSparkInterpreter() throws IOException, InterruptedException, InterpreterException {
Properties properties = new Properties();
properties.setProperty(SparkStringConstants.MASTER_PROP_NAME, "local");
properties.setProperty(SparkStringConstants.APP_NAME_PROP_NAME, "test");
@@ -90,7 +93,8 @@ public void testSparkInterpreter() throws IOException, InterruptedException, Int
InterpreterResult result = interpreter.interpret("val a=\"hello world\"", getInterpreterContext());
assertEquals(InterpreterResult.Code.SUCCESS, result.code());
- assertEquals("a: String = hello world\n", output);
+ // Use contains instead of equals, because there's behavior difference between different scala versions
+ assertTrue(output.contains("a: String = hello world\n"), output);
result = interpreter.interpret("print(a)", getInterpreterContext());
assertEquals(InterpreterResult.Code.SUCCESS, result.code());
@@ -122,9 +126,12 @@ public void testSparkInterpreter() throws IOException, InterruptedException, Int
result = interpreter.interpret("/*comment here*/\nprint(\"hello world\")", getInterpreterContext());
assertEquals(InterpreterResult.Code.SUCCESS, result.code());
- // multiple line comment
- result = interpreter.interpret("/*line 1 \n line 2*/", getInterpreterContext());
- assertEquals(InterpreterResult.Code.SUCCESS, result.code());
+ if (!interpreter.isScala213()) {
+ // multiple line comment, not supported by scala-2.13
+ context = getInterpreterContext();
+ result = interpreter.interpret("/*line 1 \n line 2*/", context);
+ assertEquals(InterpreterResult.Code.SUCCESS, result.code(), context.out.toString());
+ }
// test function
result = interpreter.interpret("def add(x:Int, y:Int)\n{ return x+y }", getInterpreterContext());
@@ -136,12 +143,6 @@ public void testSparkInterpreter() throws IOException, InterruptedException, Int
result = interpreter.interpret("/*line 1 \n line 2*/print(\"hello world\")", getInterpreterContext());
assertEquals(InterpreterResult.Code.SUCCESS, result.code());
- // test $intp, only works for scala after 2.11
- if (!interpreter.isScala210()) {
- result = interpreter.interpret("$intp", getInterpreterContext());
- assertEquals(InterpreterResult.Code.SUCCESS, result.code());
- }
-
// Companion object with case class
result = interpreter.interpret("import scala.math._\n" +
"object Circle {\n" +
@@ -155,6 +156,11 @@ public void testSparkInterpreter() throws IOException, InterruptedException, Int
"val circle1 = new Circle(5.0)", getInterpreterContext());
assertEquals(InterpreterResult.Code.SUCCESS, result.code());
+ // use case class in spark
+ // context = getInterpreterContext();
+ // result = interpreter.interpret("sc\n.range(1, 10)\n.map(e=>Circle(e))\n.collect()", context);
+ // assertEquals(context.out.toString(), InterpreterResult.Code.SUCCESS, result.code());
+
// class extend
result = interpreter.interpret("import java.util.ArrayList", getInterpreterContext());
assertEquals(InterpreterResult.Code.SUCCESS, result.code());
@@ -165,7 +171,7 @@ public void testSparkInterpreter() throws IOException, InterruptedException, Int
// spark rdd operation
context = getInterpreterContext();
context.setParagraphId("pid_1");
- result = interpreter.interpret("sc\n.range(1, 10)\n.sum", context);
+ result = interpreter.interpret("sc\n.range(1, 10)\n.map(e=>e)\n.sum", context);
assertEquals(InterpreterResult.Code.SUCCESS, result.code());
assertTrue(output.contains("45"));
ArgumentCaptor