Elgato_dark/zeppelin
1
0
Fork 0
mirror of https://github.com/apache/zeppelin synced 2026-05-24 09:38:26 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 1

ZEPPELIN-1263. Should specify zeppelin's spark configuration through --conf arguments of spark-submit

Browse source
This commit is contained in:
Jeff Zhang 2016-08-03 12:50:04 +08:00
parent 1c23f21388
commit 151e991427
3 changed files with 29 additions and 27 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

4
bin/interpreter.sh
View file

@ -202,9 +202,9 @@ fi
if [[ -n "${SPARK_SUBMIT}" ]]; then
if [[ -n "$ZEPPELIN_IMPERSONATE_USER" ]] && [[ "$ZEPPELIN_IMPERSONATE_SPARK_PROXY_USER" != "false" ]]; then
INTERPRETER_RUN_COMMAND+=' '` echo ${SPARK_SUBMIT} --class ${ZEPPELIN_SERVER} --driver-class-path \"${ZEPPELIN_INTP_CLASSPATH_OVERRIDES}:${ZEPPELIN_INTP_CLASSPATH}\" --driver-java-options \"${JAVA_INTP_OPTS}\" ${SPARK_SUBMIT_OPTIONS} --proxy-user ${ZEPPELIN_IMPERSONATE_USER} ${SPARK_APP_JAR} ${PORT}`
INTERPRETER_RUN_COMMAND+=' '` echo ${SPARK_SUBMIT} --class ${ZEPPELIN_SERVER} --driver-class-path \"${ZEPPELIN_INTP_CLASSPATH_OVERRIDES}:${ZEPPELIN_INTP_CLASSPATH}\" --driver-java-options \"${JAVA_INTP_OPTS}\" ${SPARK_SUBMIT_OPTIONS} ${ZEPPELIN_SPARK_CONF} --proxy-user ${ZEPPELIN_IMPERSONATE_USER} ${SPARK_APP_JAR} ${PORT}`
else
INTERPRETER_RUN_COMMAND+=' '` echo ${SPARK_SUBMIT} --class ${ZEPPELIN_SERVER} --driver-class-path \"${ZEPPELIN_INTP_CLASSPATH_OVERRIDES}:${ZEPPELIN_INTP_CLASSPATH}\" --driver-java-options \"${JAVA_INTP_OPTS}\" ${SPARK_SUBMIT_OPTIONS} ${SPARK_APP_JAR} ${PORT}`
INTERPRETER_RUN_COMMAND+=' '` echo ${SPARK_SUBMIT} --class ${ZEPPELIN_SERVER} --driver-class-path \"${ZEPPELIN_INTP_CLASSPATH_OVERRIDES}:${ZEPPELIN_INTP_CLASSPATH}\" --driver-java-options \"${JAVA_INTP_OPTS}\" ${SPARK_SUBMIT_OPTIONS} ${ZEPPELIN_SPARK_CONF} ${SPARK_APP_JAR} ${PORT}`
fi
else
INTERPRETER_RUN_COMMAND+=' '` echo ${ZEPPELIN_RUNNER} ${JAVA_INTP_OPTS} ${ZEPPELIN_INTP_MEM} -cp ${ZEPPELIN_INTP_CLASSPATH_OVERRIDES}:${ZEPPELIN_INTP_CLASSPATH} ${ZEPPELIN_SERVER} ${PORT} `

27
spark/src/main/java/org/apache/zeppelin/spark/SparkInterpreter.java
View file

@ -340,10 +340,12 @@ public class SparkInterpreter extends Interpreter {
* Create SparkSession
*/
public Object createSparkSession() {
logger.info("------ Create new SparkContext {} -------", getProperty("master"));
// use local mode for embedded spark mode when spark.master is not found
if (!conf.contains("spark.master")) {
conf.setMaster("local");
}
logger.info("------ Create new SparkSession {} -------", conf.get("spark.master"));
String execUri = System.getenv("SPARK_EXECUTOR_URI");
conf.setAppName(getProperty("spark.app.name"));
if (outputDir != null) {
conf.set("spark.repl.class.outputDir", outputDir.getAbsolutePath());
}
@ -351,11 +353,6 @@ public class SparkInterpreter extends Interpreter {
if (execUri != null) {
conf.set("spark.executor.uri", execUri);
}
if (System.getenv("SPARK_HOME") != null) {
conf.setSparkHome(System.getenv("SPARK_HOME"));
}
conf.set("spark.scheduler.mode", "FAIR");
conf.setMaster(getProperty("master"));
if (isYarnMode()) {
@ -364,7 +361,6 @@ public class SparkInterpreter extends Interpreter {
}
Properties intpProperty = getProperty();
for (Object k : intpProperty.keySet()) {
String key = (String) k;
String val = toString(intpProperty.get(key));
@ -417,7 +413,11 @@ public class SparkInterpreter extends Interpreter {
}
public SparkContext createSparkContext_1() {
logger.info("------ Create new SparkContext {} -------", getProperty("master"));
// use local mode for embedded spark mode when spark.master is not found
if (!conf.contains("spark.master")) {
conf.setMaster("local");
}
logger.info("------ Create new SparkContext {} -------", conf.get("spark.master"));
String execUri = System.getenv("SPARK_EXECUTOR_URI");
String[] jars = null;
@ -471,9 +471,6 @@ public class SparkInterpreter extends Interpreter {
classServerUri = (String) Utils.invokeMethod(classServer, "uri");
}
conf.setMaster(getProperty("master"))
.setAppName(getProperty("spark.app.name"));
if (classServerUri != null) {
conf.set("spark.repl.class.uri", classServerUri);
}
@ -489,13 +486,9 @@ public class SparkInterpreter extends Interpreter {
if (execUri != null) {
conf.set("spark.executor.uri", execUri);
}
if (System.getenv("SPARK_HOME") != null) {
conf.setSparkHome(System.getenv("SPARK_HOME"));
}
conf.set("spark.scheduler.mode", "FAIR");
Properties intpProperty = getProperty();
for (Object k : intpProperty.keySet()) {
String key = (String) k;
String val = toString(intpProperty.get(key));

25
zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreter.java
View file

@ -19,6 +19,7 @@ package org.apache.zeppelin.interpreter.remote;
import java.util.*;
import org.apache.commons.lang3.StringUtils;
import org.apache.thrift.TException;
import org.apache.zeppelin.display.AngularObject;
import org.apache.zeppelin.display.AngularObjectRegistry;
@ -140,15 +141,27 @@ public class RemoteInterpreter extends Interpreter {
}
private Map<String, String> getEnvFromInterpreterProperty(Properties property) {
Map<String, String> env = new HashMap<>();
for (Object key : property.keySet()) {
if (RemoteInterpreterUtils.isEnvString((String) key)) {
env.put((String) key, property.getProperty((String) key));
Map<String, String> env = new HashMap<String, String>();
StringBuilder sparkConfBuilder = new StringBuilder();
for (String key : property.stringPropertyNames()) {
if (RemoteInterpreterUtils.isEnvString(key)) {
env.put(key, property.getProperty(key));
}
if (key.equals("master")) {
sparkConfBuilder.append(" --master " + property.getProperty("master"));
}
if (isSparkConf(key, property.getProperty(key))) {
sparkConfBuilder.append(" --conf " + key + "=" + property.getProperty(key));
}
}
env.put("ZEPPELIN_SPARK_CONF", sparkConfBuilder.toString());
return env;
}
static boolean isSparkConf(String key, String value) {
return !StringUtils.isEmpty(key) && key.startsWith("spark.") && !StringUtils.isEmpty(value);
}
@Override
public String getClassName() {
return className;
@ -559,10 +572,6 @@ public class RemoteInterpreter extends Interpreter {
return env;
}
public void setEnv(Map<String, String> env) {
this.env = env;
}
public void addEnv(Map<String, String> env) {
if (this.env == null) {
this.env = new HashMap<>();