zjffdu commented on a change in pull request #3375: ZEPPELIN-4176. Remove old
spark interpreter
URL: https://github.com/apache/zeppelin/pull/3375#discussion_r290221120
##########
File path:
spark/interpreter/src/main/java/org/apache/zeppelin/spark/SparkInterpreter.java
##########
@@ -50,46 +71,103 @@ public SparkInterpreter(Properties properties) {
if
(Boolean.parseBoolean(properties.getProperty("zeppelin.spark.scala.color",
"true"))) {
System.setProperty("scala.color", "true");
}
- if (Boolean.parseBoolean(properties.getProperty("zeppelin.spark.useNew",
"false"))) {
- delegation = new NewSparkInterpreter(properties);
- } else {
- delegation = new OldSparkInterpreter(properties);
- }
- delegation.setParentSparkInterpreter(this);
+ 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");
}
@Override
public void open() throws InterpreterException {
- delegation.setInterpreterGroup(getInterpreterGroup());
- delegation.setUserName(getUserName());
- delegation.setClassloaderUrls(getClassloaderUrls());
-
- delegation.open();
+ try {
+ String scalaVersion = extractScalaVersion();
+ LOGGER.info("Using Scala Version: " + scalaVersion);
+ SparkConf conf = new SparkConf();
+ for (Map.Entry<Object, Object> entry : getProperties().entrySet()) {
+ if (!StringUtils.isBlank(entry.getValue().toString())) {
+ conf.set(entry.getKey().toString(), entry.getValue().toString());
+ }
+ // zeppelin.spark.useHiveContext & zeppelin.spark.concurrentSQL are
legacy zeppelin
+ // properties, convert them to spark properties here.
+ if (entry.getKey().toString().equals("zeppelin.spark.useHiveContext"))
{
+ conf.set("spark.useHiveContext", entry.getValue().toString());
+ }
+ if (entry.getKey().toString().equals("zeppelin.spark.concurrentSQL")
+ && entry.getValue().toString().equals("true")) {
+ conf.set("spark.scheduler.mode", "FAIR");
+ }
+ }
+ // use local mode for embedded spark mode when spark.master is not found
+ conf.setIfMissing("spark.master", "local");
+
+ String innerIntpClassName = innerInterpreterClassMap.get(scalaVersion);
+ Class clazz = Class.forName(innerIntpClassName);
+ this.innerInterpreter = (BaseSparkScalaInterpreter)
+ clazz.getConstructor(SparkConf.class, List.class, Boolean.class)
+ .newInstance(conf, getDependencyFiles(),
+
Boolean.parseBoolean(getProperty("zeppelin.spark.printREPLOutput", "true")));
+ this.innerInterpreter.open();
+
+ sc = this.innerInterpreter.sc();
+ jsc = JavaSparkContext.fromSparkContext(sc);
+ sparkVersion = SparkVersion.fromVersionString(sc.version());
+ if (enableSupportedVersionCheck && sparkVersion.isUnsupportedVersion()) {
+ throw new Exception("This is not officially supported spark version: "
+ sparkVersion
+ + "\nYou can set zeppelin.spark.enableSupportedVersionCheck to
false if you really" +
+ " want to try this version of spark.");
+ }
+ sqlContext = this.innerInterpreter.sqlContext();
+ sparkSession = this.innerInterpreter.sparkSession();
+ hooks = getInterpreterGroup().getInterpreterHookRegistry();
+ sparkUrl = this.innerInterpreter.sparkUrl();
+ String sparkUrlProp = getProperty("zeppelin.spark.uiWebUrl", "");
+ if (!StringUtils.isBlank(sparkUrlProp)) {
+ sparkUrl = sparkUrlProp;
+ }
+ sparkShims = SparkShims.getInstance(sc.version(), getProperties());
+ sparkShims.setupSparkListener(sc.master(), sparkUrl,
InterpreterContext.get());
+
+ z = new SparkZeppelinContext(sc, sparkShims, hooks,
+ Integer.parseInt(getProperty("zeppelin.spark.maxResult")));
+ this.innerInterpreter.bind("z", z.getClass().getCanonicalName(), z,
+ Lists.newArrayList("@transient"));
+ } catch (Exception e) {
+ LOGGER.error("Fail to open SparkInterpreter",
ExceptionUtils.getStackTrace(e));
Review comment:
Fixed
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
