Skip to content

[ZEPPELIN-4710]. Allow to inject application id into custom spark url #3711

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
zjffdu wants to merge 2 commits into from
Closed

[ZEPPELIN-4710]. Allow to inject application id into custom spark url #3711

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
68fceec
[ZEPPELIN-4710]. Allow to inject application id into custom spark url
zjffdu Mar 31, 2020
a887199
remove code duplicate
zjffdu Apr 7, 2020
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 3 additions & 2 deletions spark/interpreter/src/test/java/org/apache/zeppelin/spark/SparkInterpreterTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -72,7 +72,7 @@ public void testSparkInterpreter() throws IOException, InterruptedException, Int
properties.setProperty("spark.master", "local");
properties.setProperty("spark.app.name", "test");
properties.setProperty("zeppelin.spark.maxResult", "100");
properties.setProperty("zeppelin.spark.uiWebUrl", "fake_spark_weburl");
properties.setProperty("zeppelin.spark.uiWebUrl", "fake_spark_weburl/{{applicationId}}");
// disable color output for easy testing
properties.setProperty("zeppelin.spark.scala.color", "false");
properties.setProperty("zeppelin.spark.deprecatedMsg.show", "false");
Expand Down Expand Up @@ -180,7 +180,8 @@ public void testSparkInterpreter() throws IOException, InterruptedException, Int
// spark job url is sent
ArgumentCaptor<Map> onParaInfosReceivedArg = ArgumentCaptor.forClass(Map.class);
verify(mockRemoteEventClient).onParaInfosReceived(onParaInfosReceivedArg.capture());
assertTrue(((String) onParaInfosReceivedArg.getValue().get("jobUrl")).startsWith("fake_spark_weburl"));
assertTrue(((String) onParaInfosReceivedArg.getValue().get("jobUrl")).startsWith("fake_spark_weburl/"
+ interpreter.getJavaSparkContext().sc().applicationId()));

// case class
result = interpreter.interpret("val bankText = sc.textFile(\"bank.csv\")", getInterpreterContext());
Expand Down
21 changes: 14 additions & 7 deletions ...ark-scala-parent/src/main/scala/org/apache/zeppelin/spark/BaseSparkScalaInterpreter.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -237,6 +237,8 @@ abstract class BaseSparkScalaInterpreter(val conf: SparkConf,
case None =>
}

initSparkWebUrl()

val hiveSiteExisted: Boolean =
Thread.currentThread().getContextClassLoader.getResource("hive-site.xml") != null
val hiveEnabled = conf.getBoolean("zeppelin.spark.useHiveContext", false)
Expand Down Expand Up @@ -306,7 +308,8 @@ abstract class BaseSparkScalaInterpreter(val conf: SparkConf,
case Some(url) => sparkUrl = url
case None =>
}
useYarnProxyURLIfNeeded()

initSparkWebUrl()

bind("spark", sparkSession.getClass.getCanonicalName, sparkSession, List("""@transient"""))
bind("sc", "org.apache.spark.SparkContext", sc, List("""@transient"""))
Expand All @@ -321,6 +324,15 @@ abstract class BaseSparkScalaInterpreter(val conf: SparkConf,
scalaInterpret("print(\"\")")
}

private def initSparkWebUrl(): Unit = {
val webUiUrl = properties.getProperty("zeppelin.spark.uiWebUrl");
if (!StringUtils.isBlank(webUiUrl)) {
this.sparkUrl = webUiUrl.replace("{{applicationId}}", sc.applicationId);
} else {
useYarnProxyURLIfNeeded()
}
}

protected def createZeppelinContext(): Unit = {

var sparkShims: SparkShims = null
Expand All @@ -329,13 +341,8 @@ abstract class BaseSparkScalaInterpreter(val conf: SparkConf,
} else {
sparkShims = SparkShims.getInstance(sc.version, properties, sc)
}
var webUiUrl = properties.getProperty("zeppelin.spark.uiWebUrl");
if (StringUtils.isBlank(webUiUrl)) {
webUiUrl = sparkUrl;
}
useYarnProxyURLIfNeeded()

sparkShims.setupSparkListener(sc.master, webUiUrl, InterpreterContext.get)
sparkShims.setupSparkListener(sc.master, sparkUrl, InterpreterContext.get)

z = new SparkZeppelinContext(sc, sparkShims,
interpreterGroup.getInterpreterHookRegistry,
Expand Down