You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@zeppelin.apache.org by pd...@apache.org on 2023/10/11 08:12:24 UTC
[zeppelin] branch master updated: [ZEPPELIN-5946][FOLLOWUP] Clean up code for legacy Spark versions (#4668)
This is an automated email from the ASF dual-hosted git repository.
pdallig pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/zeppelin.git
The following commit(s) were added to refs/heads/master by this push:
new b9cb004515 [ZEPPELIN-5946][FOLLOWUP] Clean up code for legacy Spark versions (#4668)
b9cb004515 is described below
commit b9cb004515fdf48332d8b9da10640ad86afef840
Author: Cheng Pan <ch...@apache.org>
AuthorDate: Wed Oct 11 16:12:15 2023 +0800
[ZEPPELIN-5946][FOLLOWUP] Clean up code for legacy Spark versions (#4668)
* [ZEPPELIN-5946][FOLLOWUP] Clean up code for legacy Spark versions
* eliminate the reflection
* address
---
.../apache/zeppelin/python/IPythonInterpreter.java | 8 +---
.../apache/zeppelin/python/PythonInterpreter.java | 9 +---
.../org/apache/zeppelin/python/PythonUtils.java | 41 +++++------------
.../java/org/apache/zeppelin/r/IRInterpreter.java | 11 +----
.../java/org/apache/zeppelin/r/RInterpreter.java | 11 +----
.../java/org/apache/zeppelin/r/SparkRBackend.java | 17 +++----
.../main/java/org/apache/zeppelin/r/ZeppelinR.java | 6 +--
.../apache/zeppelin/spark/IPySparkInterpreter.java | 4 --
.../apache/zeppelin/spark/PySparkInterpreter.java | 2 -
.../apache/zeppelin/spark/SparkIRInterpreter.java | 35 ++++----------
.../apache/zeppelin/spark/SparkInterpreter.java | 6 ---
.../apache/zeppelin/spark/SparkRInterpreter.java | 19 +++-----
.../main/java/org/apache/zeppelin/spark/Utils.java | 21 ++-------
.../org/apache/zeppelin/spark/SparkShimsTest.java | 8 +---
.../apache/zeppelin/spark/SparkVersionTest.java | 30 ++++++------
.../java/org/apache/zeppelin/spark/SparkShims.java | 3 --
.../org/apache/zeppelin/spark/SparkVersion.java | 17 +------
.../launcher/SparkInterpreterLauncher.java | 53 ++++++++--------------
18 files changed, 83 insertions(+), 218 deletions(-)
diff --git a/python/src/main/java/org/apache/zeppelin/python/IPythonInterpreter.java b/python/src/main/java/org/apache/zeppelin/python/IPythonInterpreter.java
index 2f357270c3..d40cef49b8 100644
--- a/python/src/main/java/org/apache/zeppelin/python/IPythonInterpreter.java
+++ b/python/src/main/java/org/apache/zeppelin/python/IPythonInterpreter.java
@@ -55,7 +55,6 @@ public class IPythonInterpreter extends JupyterKernelInterpreter {
private String additionalPythonPath;
private String additionalPythonInitFile;
private boolean useBuiltinPy4j = true;
- private boolean usePy4JAuth = true;
private String py4jGatewaySecret;
public IPythonInterpreter(Properties properties) {
@@ -128,7 +127,7 @@ public class IPythonInterpreter extends JupyterKernelInterpreter {
private void setupJVMGateway(String gatewayHost, int gatewayPort) throws IOException {
this.gatewayServer = PythonUtils.createGatewayServer(this, gatewayHost,
- gatewayPort, py4jGatewaySecret, usePy4JAuth);
+ gatewayPort, py4jGatewaySecret);
gatewayServer.start();
}
@@ -199,11 +198,8 @@ public class IPythonInterpreter extends JupyterKernelInterpreter {
envs.put("PYTHONPATH", additionalPythonPath);
}
- this.usePy4JAuth = Boolean.parseBoolean(getProperty("zeppelin.py4j.useAuth", "true"));
this.py4jGatewaySecret = PythonUtils.createSecret(256);
- if (usePy4JAuth) {
- envs.put("PY4J_GATEWAY_SECRET", this.py4jGatewaySecret);
- }
+ envs.put("PY4J_GATEWAY_SECRET", this.py4jGatewaySecret);
LOGGER.info("PYTHONPATH: {}", envs.get("PYTHONPATH"));
return envs;
}
diff --git a/python/src/main/java/org/apache/zeppelin/python/PythonInterpreter.java b/python/src/main/java/org/apache/zeppelin/python/PythonInterpreter.java
index 6dfea9941f..46dbf8badd 100644
--- a/python/src/main/java/org/apache/zeppelin/python/PythonInterpreter.java
+++ b/python/src/main/java/org/apache/zeppelin/python/PythonInterpreter.java
@@ -70,7 +70,6 @@ public class PythonInterpreter extends Interpreter {
private ZeppelinContext zeppelinContext;
// set by PythonCondaInterpreter
private String condaPythonExec;
- private boolean usePy4jAuth = false;
public PythonInterpreter(Properties property) {
super(property);
@@ -116,7 +115,6 @@ public class PythonInterpreter extends Interpreter {
}
try {
- this.usePy4jAuth = Boolean.parseBoolean(getProperty("zeppelin.py4j.useAuth", "true"));
createGatewayServerAndStartScript();
} catch (IOException e) {
LOGGER.error("Fail to open PythonInterpreter", e);
@@ -132,8 +130,7 @@ public class PythonInterpreter extends Interpreter {
// container can also connect to this gateway server.
String serverAddress = PythonUtils.getLocalIP(properties);
String secret = PythonUtils.createSecret(256);
- this.gatewayServer = PythonUtils.createGatewayServer(this, serverAddress, port, secret,
- usePy4jAuth);
+ this.gatewayServer = PythonUtils.createGatewayServer(this, serverAddress, port, secret);
gatewayServer.start();
// launch python process to connect to the gateway server in JVM side
@@ -149,9 +146,7 @@ public class PythonInterpreter extends Interpreter {
outputStream = new InterpreterOutputStream(LOGGER);
Map<String, String> env = setupPythonEnv();
- if (usePy4jAuth) {
- env.put("PY4J_GATEWAY_SECRET", secret);
- }
+ env.put("PY4J_GATEWAY_SECRET", secret);
if (LOGGER.isInfoEnabled()) {
LOGGER.info("Launching Python Process Command: {} {}",
cmd.getExecutable(), StringUtils.join(cmd.getArguments(), " "));
diff --git a/python/src/main/java/org/apache/zeppelin/python/PythonUtils.java b/python/src/main/java/org/apache/zeppelin/python/PythonUtils.java
index 15b71f0c8e..7e63b7d236 100644
--- a/python/src/main/java/org/apache/zeppelin/python/PythonUtils.java
+++ b/python/src/main/java/org/apache/zeppelin/python/PythonUtils.java
@@ -27,7 +27,6 @@ import java.net.Inet4Address;
import java.net.InetAddress;
import java.net.UnknownHostException;
import java.security.SecureRandom;
-import java.util.List;
import java.util.Properties;
public class PythonUtils {
@@ -37,35 +36,17 @@ public class PythonUtils {
public static GatewayServer createGatewayServer(Object entryPoint,
String serverAddress,
int port,
- String secretKey,
- boolean useAuth) throws IOException {
- LOGGER.info("Launching GatewayServer at " + serverAddress + ":" + port +
- ", useAuth: " + useAuth);
- if (useAuth) {
- try {
- Class clz = Class.forName("py4j.GatewayServer$GatewayServerBuilder", true,
- Thread.currentThread().getContextClassLoader());
- Object builder = clz.getConstructor(Object.class).newInstance(entryPoint);
- builder.getClass().getMethod("authToken", String.class).invoke(builder, secretKey);
- builder.getClass().getMethod("javaPort", int.class).invoke(builder, port);
- builder.getClass().getMethod("javaAddress", InetAddress.class).invoke(builder,
- InetAddress.getByName(serverAddress));
- builder.getClass()
- .getMethod("callbackClient", int.class, InetAddress.class, String.class)
- .invoke(builder, port, InetAddress.getByName(serverAddress), secretKey);
- return (GatewayServer) builder.getClass().getMethod("build").invoke(builder);
- } catch (Exception e) {
- throw new IOException(e);
- }
- } else {
- return new GatewayServer(entryPoint,
- port,
- GatewayServer.DEFAULT_PYTHON_PORT,
- InetAddress.getByName(serverAddress),
- InetAddress.getByName(serverAddress),
- GatewayServer.DEFAULT_CONNECT_TIMEOUT,
- GatewayServer.DEFAULT_READ_TIMEOUT,
- (List) null);
+ String secretKey) throws IOException {
+ LOGGER.info("Launching GatewayServer at {}:{}", serverAddress, port);
+ try {
+ return new GatewayServer.GatewayServerBuilder(entryPoint)
+ .authToken(secretKey)
+ .javaPort(port)
+ .javaAddress(InetAddress.getByName(serverAddress))
+ .callbackClient(port, InetAddress.getByName(serverAddress), secretKey)
+ .build();
+ } catch (Exception e) {
+ throw new IOException(e);
}
}
diff --git a/rlang/src/main/java/org/apache/zeppelin/r/IRInterpreter.java b/rlang/src/main/java/org/apache/zeppelin/r/IRInterpreter.java
index 56d0673105..6407459354 100644
--- a/rlang/src/main/java/org/apache/zeppelin/r/IRInterpreter.java
+++ b/rlang/src/main/java/org/apache/zeppelin/r/IRInterpreter.java
@@ -80,15 +80,6 @@ public class IRInterpreter extends JupyterKernelInterpreter {
return 20404;
}
- /**
- * Spark 2.4.3 need secret for socket communication between R process and jvm process.
- * Sub class can override this, e.g. SparkRInterpreter
- * @return
- */
- protected boolean isSecretSupported() {
- return true;
- }
-
@Override
public void open() throws InterpreterException {
super.open();
@@ -98,7 +89,7 @@ public class IRInterpreter extends JupyterKernelInterpreter {
synchronized (sparkRBackend) {
if (!sparkRBackend.isStarted()) {
try {
- sparkRBackend.init(isSecretSupported());
+ sparkRBackend.init();
} catch (Exception e) {
throw new InterpreterException("Fail to init SparkRBackend", e);
}
diff --git a/rlang/src/main/java/org/apache/zeppelin/r/RInterpreter.java b/rlang/src/main/java/org/apache/zeppelin/r/RInterpreter.java
index 76aa8ea147..b4837db1c3 100644
--- a/rlang/src/main/java/org/apache/zeppelin/r/RInterpreter.java
+++ b/rlang/src/main/java/org/apache/zeppelin/r/RInterpreter.java
@@ -71,15 +71,6 @@ public class RInterpreter extends AbstractInterpreter {
return 20403;
}
- /**
- * Spark 2.4.3 need secret for socket communication between R process and jvm process.
- * Sub class can override this, e.g. SparkRInterpreter
- * @return
- */
- protected boolean isSecretSupported() {
- return true;
- }
-
@Override
public void open() throws InterpreterException {
this.sparkRBackend = SparkRBackend.get();
@@ -87,7 +78,7 @@ public class RInterpreter extends AbstractInterpreter {
synchronized (sparkRBackend) {
if (!sparkRBackend.isStarted()) {
try {
- sparkRBackend.init(isSecretSupported());
+ sparkRBackend.init();
} catch (Exception e) {
throw new InterpreterException("Fail to init SparkRBackend", e);
}
diff --git a/rlang/src/main/java/org/apache/zeppelin/r/SparkRBackend.java b/rlang/src/main/java/org/apache/zeppelin/r/SparkRBackend.java
index 1b07668fa5..7c482e0432 100644
--- a/rlang/src/main/java/org/apache/zeppelin/r/SparkRBackend.java
+++ b/rlang/src/main/java/org/apache/zeppelin/r/SparkRBackend.java
@@ -49,17 +49,12 @@ public class SparkRBackend {
};
}
- public void init(boolean isSecretSocketSupported) throws Exception {
- Class rBackendClass = RBackend.class;
- if (isSecretSocketSupported) {
- Tuple2<Integer, Object> result =
- (Tuple2<Integer, Object>) rBackendClass.getMethod("init").invoke(backend);
- portNumber = result._1;
- Object rAuthHelper = result._2;
- secret = (String) rAuthHelper.getClass().getMethod("secret").invoke(rAuthHelper);
- } else {
- portNumber = (Integer) rBackendClass.getMethod("init").invoke(backend);
- }
+ public void init() throws Exception {
+ Tuple2<Integer, Object> result =
+ (Tuple2<Integer, Object>) RBackend.class.getMethod("init").invoke(backend);
+ portNumber = result._1;
+ Object rAuthHelper = result._2;
+ secret = (String) rAuthHelper.getClass().getMethod("secret").invoke(rAuthHelper);
}
public void start() {
diff --git a/rlang/src/main/java/org/apache/zeppelin/r/ZeppelinR.java b/rlang/src/main/java/org/apache/zeppelin/r/ZeppelinR.java
index 1ca389b0e1..617521c3b1 100644
--- a/rlang/src/main/java/org/apache/zeppelin/r/ZeppelinR.java
+++ b/rlang/src/main/java/org/apache/zeppelin/r/ZeppelinR.java
@@ -124,11 +124,9 @@ public class ZeppelinR {
cmd.addArgument(rInterpreter.sparkVersion() + "");
cmd.addArgument(timeout);
cmd.addArgument(rInterpreter.isSparkSupported() + "");
- if (rInterpreter.isSecretSupported()) {
- cmd.addArgument(SparkRBackend.get().socketSecret());
- }
+ cmd.addArgument(SparkRBackend.get().socketSecret());
// dump out the R command to facilitate manually running it, e.g. for fault diagnosis purposes
- LOGGER.info("R Command: " + cmd.toString());
+ LOGGER.info("R Command: {}", cmd);
processOutputStream = new RProcessLogOutputStream(rInterpreter);
Map env = EnvironmentUtils.getProcEnvironment();
rProcessLauncher = new RProcessLauncher(cmd, env, processOutputStream);
diff --git a/spark/interpreter/src/main/java/org/apache/zeppelin/spark/IPySparkInterpreter.java b/spark/interpreter/src/main/java/org/apache/zeppelin/spark/IPySparkInterpreter.java
index 2e945ed2bb..85b8d4cf0f 100644
--- a/spark/interpreter/src/main/java/org/apache/zeppelin/spark/IPySparkInterpreter.java
+++ b/spark/interpreter/src/main/java/org/apache/zeppelin/spark/IPySparkInterpreter.java
@@ -59,8 +59,6 @@ public class IPySparkInterpreter extends IPythonInterpreter {
getInterpreterInTheSameSessionByClassName(PySparkInterpreter.class, false);
setProperty("zeppelin.python", pySparkInterpreter.getPythonExec(sparkInterpreter.getSparkContext().conf()));
- setProperty("zeppelin.py4j.useAuth",
- sparkInterpreter.getSparkVersion().isSecretSocketSupported() + "");
SparkConf conf = sparkInterpreter.getSparkContext().getConf();
// only set PYTHONPATH in embedded, local or yarn-client mode.
// yarn-cluster will setup PYTHONPATH automatically.
@@ -70,8 +68,6 @@ public class IPySparkInterpreter extends IPythonInterpreter {
}
setUseBuiltinPy4j(false);
setAdditionalPythonInitFile("python/zeppelin_ipyspark.py");
- setProperty("zeppelin.py4j.useAuth",
- sparkInterpreter.getSparkVersion().isSecretSocketSupported() + "");
super.open();
opened = true;
}
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 ee891a62d4..2530d0c959 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
@@ -88,8 +88,6 @@ public class PySparkInterpreter extends PythonInterpreter {
// 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 {
diff --git a/spark/interpreter/src/main/java/org/apache/zeppelin/spark/SparkIRInterpreter.java b/spark/interpreter/src/main/java/org/apache/zeppelin/spark/SparkIRInterpreter.java
index ee16c7266e..3477eb0748 100644
--- a/spark/interpreter/src/main/java/org/apache/zeppelin/spark/SparkIRInterpreter.java
+++ b/spark/interpreter/src/main/java/org/apache/zeppelin/spark/SparkIRInterpreter.java
@@ -38,7 +38,6 @@ public class SparkIRInterpreter extends IRInterpreter {
private SparkInterpreter sparkInterpreter;
private SparkVersion sparkVersion;
- private boolean isSpark2;
private SparkContext sc;
private JavaSparkContext jsc;
@@ -54,10 +53,6 @@ public class SparkIRInterpreter extends IRInterpreter {
return this.sparkVersion.toNumber();
}
- protected boolean isSecretSupported() {
- return this.sparkVersion.isSecretSocketSupported();
- }
-
/**
* We can inject SparkInterpreter in the case that SparkIRInterpreter is used by
* SparkShinyInterpreter in which case it is not in the same InterpreterGroup of
@@ -75,13 +70,10 @@ public class SparkIRInterpreter extends IRInterpreter {
this.sc = sparkInterpreter.getSparkContext();
this.jsc = sparkInterpreter.getJavaSparkContext();
this.sparkVersion = new SparkVersion(sc.version());
- this.isSpark2 = sparkVersion.newerThanEquals(SparkVersion.SPARK_2_0_0);
ZeppelinRContext.setSparkContext(sc);
ZeppelinRContext.setJavaSparkContext(jsc);
- if (isSpark2) {
- ZeppelinRContext.setSparkSession(sparkInterpreter.getSparkSession());
- }
+ ZeppelinRContext.setSparkSession(sparkInterpreter.getSparkSession());
ZeppelinRContext.setSqlContext(sparkInterpreter.getSQLContext());
ZeppelinRContext.setZeppelinContext(sparkInterpreter.getZeppelinContext());
super.open();
@@ -94,25 +86,16 @@ public class SparkIRInterpreter extends IRInterpreter {
String jobGroup = Utils.buildJobGroupId(context);
String jobDesc = Utils.buildJobDesc(context);
sparkInterpreter.getSparkContext().setJobGroup(jobGroup, jobDesc, false);
- String setJobGroup = "";
- // assign setJobGroup to dummy__, otherwise it would print NULL for this statement
- if (isSpark2) {
- setJobGroup = "dummy__ <- setJobGroup(\"" + jobGroup +
- "\", \" +" + jobDesc + "\", TRUE)";
- } else {
- setJobGroup = "dummy__ <- setJobGroup(sc, \"" + jobGroup +
- "\", \"" + jobDesc + "\", TRUE)";
- }
+ // assign setJobGroup to dummy__
+ String setJobGroup = "dummy__ <- setJobGroup(\"" + jobGroup + "\", \" +" + jobDesc + "\", TRUE)";
lines = setJobGroup + "\n" + lines;
- if (sparkInterpreter.getSparkVersion().newerThanEquals(SparkVersion.SPARK_2_3_0)) {
- // setLocalProperty is only available from spark 2.3.0
- String setPoolStmt = "setLocalProperty('spark.scheduler.pool', NULL)";
- if (context.getLocalProperties().containsKey("pool")) {
- setPoolStmt = "setLocalProperty('spark.scheduler.pool', '" +
- context.getLocalProperties().get("pool") + "')";
- }
- lines = setPoolStmt + "\n" + lines;
+
+ String setPoolStmt = "setLocalProperty('spark.scheduler.pool', NULL)";
+ if (context.getLocalProperties().containsKey("pool")) {
+ setPoolStmt = "setLocalProperty('spark.scheduler.pool', '" +
+ context.getLocalProperties().get("pool") + "')";
}
+ lines = setPoolStmt + "\n" + lines;
return super.internalInterpret(lines, context);
}
}
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 65ab6f5966..c57fca62a9 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
@@ -88,7 +88,6 @@ public class SparkInterpreter extends AbstractInterpreter {
this.enableSupportedVersionCheck = java.lang.Boolean.parseBoolean(
properties.getProperty("zeppelin.spark.enableSupportedVersionCheck", "true"));
- 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");
}
@@ -274,8 +273,6 @@ public class SparkInterpreter extends AbstractInterpreter {
if (StringUtils.isEmpty(scalaVersionString)) {
throw new InterpreterException("Scala Version is empty");
- } else if (scalaVersionString.contains("2.11")) {
- return "2.11";
} else if (scalaVersionString.contains("2.12")) {
return "2.12";
} else if (scalaVersionString.contains("2.13")) {
@@ -285,9 +282,6 @@ public class SparkInterpreter extends AbstractInterpreter {
}
}
- public boolean isScala211() {
- return scalaVersion.equals("2.11");
- }
public boolean isScala212() {
return scalaVersion.equals("2.12");
diff --git a/spark/interpreter/src/main/java/org/apache/zeppelin/spark/SparkRInterpreter.java b/spark/interpreter/src/main/java/org/apache/zeppelin/spark/SparkRInterpreter.java
index 4335b6a107..0a66ed5fd8 100644
--- a/spark/interpreter/src/main/java/org/apache/zeppelin/spark/SparkRInterpreter.java
+++ b/spark/interpreter/src/main/java/org/apache/zeppelin/spark/SparkRInterpreter.java
@@ -55,11 +55,6 @@ public class SparkRInterpreter extends RInterpreter {
return true;
}
- @Override
- protected boolean isSecretSupported() {
- return sparkVersion.isSecretSocketSupported();
- }
-
@Override
protected int sparkVersion() {
return new SparkVersion(sc.version()).toNumber();
@@ -99,15 +94,13 @@ public class SparkRInterpreter extends RInterpreter {
setJobGroup = "dummy__ <- setJobGroup(\"" + jobGroup +
"\", \" +" + jobDesc + "\", TRUE)";
lines = setJobGroup + "\n" + lines;
- if (sparkInterpreter.getSparkVersion().newerThanEquals(SparkVersion.SPARK_2_3_0)) {
- // setLocalProperty is only available from spark 2.3.0
- String setPoolStmt = "setLocalProperty('spark.scheduler.pool', NULL)";
- if (interpreterContext.getLocalProperties().containsKey("pool")) {
- setPoolStmt = "setLocalProperty('spark.scheduler.pool', '" +
- interpreterContext.getLocalProperties().get("pool") + "')";
- }
- lines = setPoolStmt + "\n" + lines;
+
+ String setPoolStmt = "setLocalProperty('spark.scheduler.pool', NULL)";
+ if (interpreterContext.getLocalProperties().containsKey("pool")) {
+ setPoolStmt = "setLocalProperty('spark.scheduler.pool', '" +
+ interpreterContext.getLocalProperties().get("pool") + "')";
}
+ lines = setPoolStmt + "\n" + lines;
return super.internalInterpret(lines, interpreterContext);
}
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 578af8ed47..b325f40eec 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
@@ -33,11 +33,6 @@ import java.util.Properties;
*/
class Utils {
private static final Logger LOGGER = LoggerFactory.getLogger(Utils.class);
- private static String DEPRECATED_MESSAGE =
- "%html <font color=\"red\">Spark lower than 2.2 is deprecated, " +
- "if you don't want to see this message, please set " +
- "zeppelin.spark.deprecateMsg.show to false.</font>";
-
public static String buildJobGroupId(InterpreterContext context) {
String uName = "anonymous";
@@ -63,18 +58,10 @@ class Utils {
}
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(DEPRECATED_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/test/java/org/apache/zeppelin/spark/SparkShimsTest.java b/spark/interpreter/src/test/java/org/apache/zeppelin/spark/SparkShimsTest.java
index e76d0ca2e1..b8720f86e5 100644
--- a/spark/interpreter/src/test/java/org/apache/zeppelin/spark/SparkShimsTest.java
+++ b/spark/interpreter/src/test/java/org/apache/zeppelin/spark/SparkShimsTest.java
@@ -97,13 +97,9 @@ class SparkShimsTest {
when(mockContext.getIntpEventClient()).thenReturn(mockIntpEventClient);
try {
- sparkShims = SparkShims.getInstance(SparkVersion.SPARK_3_1_0.toString(), new Properties(), null);
+ sparkShims = SparkShims.getInstance(SparkVersion.SPARK_3_2_0.toString(), new Properties(), null);
} catch (Throwable e1) {
- try {
- sparkShims = SparkShims.getInstance(SparkVersion.SPARK_2_0_0.toString(), new Properties(), null);
- } catch (Throwable e2) {
- throw new RuntimeException("All SparkShims are tried, but no one can be created.");
- }
+ throw new RuntimeException("All SparkShims are tried, but no one can be created.");
}
}
diff --git a/spark/interpreter/src/test/java/org/apache/zeppelin/spark/SparkVersionTest.java b/spark/interpreter/src/test/java/org/apache/zeppelin/spark/SparkVersionTest.java
index 56146b517d..a454854a7f 100644
--- a/spark/interpreter/src/test/java/org/apache/zeppelin/spark/SparkVersionTest.java
+++ b/spark/interpreter/src/test/java/org/apache/zeppelin/spark/SparkVersionTest.java
@@ -32,30 +32,30 @@ class SparkVersionTest {
@Test
void testUnsupportedVersion() {
assertTrue(SparkVersion.fromVersionString("1.4.2").isUnsupportedVersion());
- assertFalse(SparkVersion.fromVersionString("2.3.0").isUnsupportedVersion());
+ assertTrue(SparkVersion.fromVersionString("2.3.0").isUnsupportedVersion());
assertTrue(SparkVersion.fromVersionString("0.9.0").isUnsupportedVersion());
assertTrue(SparkVersion.UNSUPPORTED_FUTURE_VERSION.isUnsupportedVersion());
- // should support spark2 version of HDP 2.5
- assertFalse(SparkVersion.fromVersionString("2.0.0.2.5.0.0-1245").isUnsupportedVersion());
+ // should not support spark2 version of HDP 2.5
+ assertTrue(SparkVersion.fromVersionString("2.0.0.2.5.0.0-1245").isUnsupportedVersion());
}
@Test
void testSparkVersion() {
// test equals
- assertEquals(SparkVersion.SPARK_2_0_0, SparkVersion.fromVersionString("2.0.0"));
- assertEquals(SparkVersion.SPARK_2_0_0, SparkVersion.fromVersionString("2.0.0-SNAPSHOT"));
- // test spark2 version of HDP 2.5
- assertEquals(SparkVersion.SPARK_2_0_0, SparkVersion.fromVersionString("2.0.0.2.5.0.0-1245"));
+ assertEquals(SparkVersion.SPARK_3_5_0, SparkVersion.fromVersionString("3.5.0"));
+ assertEquals(SparkVersion.SPARK_3_5_0, SparkVersion.fromVersionString("3.5.0-SNAPSHOT"));
+ // test vendor spark version
+ assertEquals(SparkVersion.SPARK_3_5_0, SparkVersion.fromVersionString("3.5.0.2.5.0.0-1245"));
// test newer than
- assertTrue(SparkVersion.SPARK_2_3_0.newerThan(SparkVersion.SPARK_2_0_0));
- assertTrue(SparkVersion.SPARK_2_3_0.newerThanEquals(SparkVersion.SPARK_2_3_0));
- assertFalse(SparkVersion.SPARK_2_0_0.newerThan(SparkVersion.SPARK_2_3_0));
+ assertTrue(SparkVersion.SPARK_3_5_0.newerThan(SparkVersion.SPARK_3_2_0));
+ assertTrue(SparkVersion.SPARK_3_5_0.newerThanEquals(SparkVersion.SPARK_3_5_0));
+ assertFalse(SparkVersion.SPARK_3_2_0.newerThan(SparkVersion.SPARK_3_5_0));
// test older than
- assertTrue(SparkVersion.SPARK_2_0_0.olderThan(SparkVersion.SPARK_2_3_0));
- assertTrue(SparkVersion.SPARK_2_0_0.olderThanEquals(SparkVersion.SPARK_2_0_0));
- assertFalse(SparkVersion.SPARK_2_3_0.olderThan(SparkVersion.SPARK_2_0_0));
+ assertTrue(SparkVersion.SPARK_3_2_0.olderThan(SparkVersion.SPARK_3_5_0));
+ assertTrue(SparkVersion.SPARK_3_2_0.olderThanEquals(SparkVersion.SPARK_3_2_0));
+ assertFalse(SparkVersion.SPARK_3_5_0.olderThan(SparkVersion.SPARK_3_2_0));
// test newerThanEqualsPatchVersion
assertTrue(SparkVersion.fromVersionString("2.3.1")
@@ -66,7 +66,7 @@ class SparkVersionTest {
.newerThanEqualsPatchVersion(SparkVersion.fromVersionString("2.2.0")));
// conversion
- assertEquals(20300, SparkVersion.SPARK_2_3_0.toNumber());
- assertEquals("2.3.0", SparkVersion.SPARK_2_3_0.toString());
+ assertEquals(30500, SparkVersion.SPARK_3_5_0.toNumber());
+ assertEquals("3.5.0", SparkVersion.SPARK_3_5_0.toString());
}
}
diff --git a/spark/spark-shims/src/main/java/org/apache/zeppelin/spark/SparkShims.java b/spark/spark-shims/src/main/java/org/apache/zeppelin/spark/SparkShims.java
index 3ae534179b..542c3de8bb 100644
--- a/spark/spark-shims/src/main/java/org/apache/zeppelin/spark/SparkShims.java
+++ b/spark/spark-shims/src/main/java/org/apache/zeppelin/spark/SparkShims.java
@@ -61,9 +61,6 @@ public abstract class SparkShims {
if (sparkMajorVersion == 3) {
LOGGER.info("Initializing shims for Spark 3.x");
sparkShimsClass = Class.forName("org.apache.zeppelin.spark.Spark3Shims");
- } else if (sparkMajorVersion == 2) {
- LOGGER.info("Initializing shims for Spark 2.x");
- sparkShimsClass = Class.forName("org.apache.zeppelin.spark.Spark2Shims");
} else {
throw new Exception("Spark major version: '" + sparkMajorVersion + "' is not supported yet");
}
diff --git a/spark/spark-shims/src/main/java/org/apache/zeppelin/spark/SparkVersion.java b/spark/spark-shims/src/main/java/org/apache/zeppelin/spark/SparkVersion.java
index 7d76f40b74..219bc656d6 100644
--- a/spark/spark-shims/src/main/java/org/apache/zeppelin/spark/SparkVersion.java
+++ b/spark/spark-shims/src/main/java/org/apache/zeppelin/spark/SparkVersion.java
@@ -25,19 +25,13 @@ import org.slf4j.LoggerFactory;
public class SparkVersion {
private static final Logger logger = LoggerFactory.getLogger(SparkVersion.class);
- public static final SparkVersion SPARK_2_0_0 = SparkVersion.fromVersionString("2.0.0");
- public static final SparkVersion SPARK_2_2_0 = SparkVersion.fromVersionString("2.2.0");
- public static final SparkVersion SPARK_2_3_0 = SparkVersion.fromVersionString("2.3.0");
- public static final SparkVersion SPARK_2_3_1 = SparkVersion.fromVersionString("2.3.1");
- public static final SparkVersion SPARK_2_4_0 = SparkVersion.fromVersionString("2.4.0");
-
- public static final SparkVersion SPARK_3_1_0 = SparkVersion.fromVersionString("3.1.0");
+ public static final SparkVersion SPARK_3_2_0 = SparkVersion.fromVersionString("3.2.0");
public static final SparkVersion SPARK_3_3_0 = SparkVersion.fromVersionString("3.3.0");
public static final SparkVersion SPARK_3_5_0 = SparkVersion.fromVersionString("3.5.0");
- public static final SparkVersion MIN_SUPPORTED_VERSION = SPARK_2_0_0;
+ public static final SparkVersion MIN_SUPPORTED_VERSION = SPARK_3_2_0;
public static final SparkVersion UNSUPPORTED_FUTURE_VERSION = SPARK_3_5_0;
private int version;
@@ -92,13 +86,6 @@ public class SparkVersion {
return new SparkVersion(versionString);
}
- public boolean isSecretSocketSupported() {
- return this.newerThanEquals(SparkVersion.SPARK_2_4_0) ||
- this.newerThanEqualsPatchVersion(SPARK_2_3_1) ||
- this.newerThanEqualsPatchVersion(SparkVersion.fromVersionString("2.2.2")) ||
- this.newerThanEqualsPatchVersion(SparkVersion.fromVersionString("2.1.3"));
- }
-
public boolean equals(Object versionToCompare) {
return version == ((SparkVersion) versionToCompare).version;
}
diff --git a/zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/launcher/SparkInterpreterLauncher.java b/zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/launcher/SparkInterpreterLauncher.java
index cc68512bf2..33b3e4ba62 100644
--- a/zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/launcher/SparkInterpreterLauncher.java
+++ b/zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/launcher/SparkInterpreterLauncher.java
@@ -279,9 +279,7 @@ public class SparkInterpreterLauncher extends StandardInterpreterLauncher {
Matcher matcher = pattern.matcher(processOutput);
if (matcher.find()) {
String scalaVersion = matcher.group(1);
- if (scalaVersion.startsWith("2.11")) {
- return "2.11";
- } else if (scalaVersion.startsWith("2.12")) {
+ if (scalaVersion.startsWith("2.12")) {
return "2.12";
} else if (scalaVersion.startsWith("2.13")) {
return "2.13";
@@ -294,37 +292,26 @@ public class SparkInterpreterLauncher extends StandardInterpreterLauncher {
}
private String detectSparkScalaVersionByReplClass(String sparkHome) throws Exception {
- File sparkLibFolder = new File(sparkHome + "/lib");
- if (sparkLibFolder.exists()) {
- // spark 1.6 if spark/lib exists
- File[] sparkAssemblyJars = new File(sparkHome + "/lib").listFiles(new FilenameFilter() {
- @Override
- public boolean accept(File dir, String name) {
- return name.contains("spark-assembly");
- }
- });
- if (sparkAssemblyJars.length == 0) {
- throw new Exception("No spark assembly file found in SPARK_HOME: " + sparkHome);
- }
- if (sparkAssemblyJars.length > 1) {
- throw new Exception("Multiple spark assembly file found in SPARK_HOME: " + sparkHome);
- }
- try (URLClassLoader urlClassLoader = new URLClassLoader(new URL[]{sparkAssemblyJars[0].toURI().toURL()});){
- urlClassLoader.loadClass("org.apache.spark.repl.SparkCommandLine");
- return "2.10";
- } catch (ClassNotFoundException e) {
- return "2.11";
- }
+ File sparkJarsFolder = new File(sparkHome + "/jars");
+ File[] sparkJarFiles = sparkJarsFolder.listFiles();
+ long sparkReplFileNum =
+ Stream.of(sparkJarFiles).filter(file -> file.getName().contains("spark-repl_")).count();
+ if (sparkReplFileNum == 0) {
+ throw new Exception("No spark-repl jar found in SPARK_HOME: " + sparkHome);
+ }
+ if (sparkReplFileNum > 1) {
+ throw new Exception("Multiple spark-repl jar found in SPARK_HOME: " + sparkHome);
+ }
+ boolean sparkRepl212Exists =
+ Stream.of(sparkJarFiles).anyMatch(file -> file.getName().contains("spark-repl_2.12"));
+ boolean sparkRepl213Exists =
+ Stream.of(sparkJarFiles).anyMatch(file -> file.getName().contains("spark-repl_2.13"));
+ if (sparkRepl212Exists) {
+ return "2.12";
+ } else if (sparkRepl213Exists) {
+ return "2.13";
} else {
- // spark 2.x if spark/lib doesn't exists
- File sparkJarsFolder = new File(sparkHome + "/jars");
- boolean sparkRepl211Exists =
- Stream.of(sparkJarsFolder.listFiles()).anyMatch(file -> file.getName().contains("spark-repl_2.11"));
- if (sparkRepl211Exists) {
- return "2.11";
- } else {
- return "2.10";
- }
+ throw new Exception("Can not detect the scala version by spark-repl");
}
}