Adding options to SparkYarnTasklet [SHDP-582, SHDP-559]

- adding extraClassPath option

- adding executorCores, driverCores and driverMemory options

Author: trisberg

spring-hadoop-spark/src/main/java/org/springframework/data/hadoop/batch/spark/SparkYarnTasklet.java

@@ -44,6 +44,8 @@ public class SparkYarnTasklet implements InitializingBean, Tasklet, StepExecutio
 
 	private String sparkAssemblyJar;
 
+	private String extraClassPath;
+
 	private Configuration hadoopConfiguration;
 
 	private String appName;
@@ -58,13 +60,23 @@ public class SparkYarnTasklet implements InitializingBean, Tasklet, StepExecutio
 
 	private String executorMemory;
 
+	private String executorCores;
+
+	private String driverMemory;
+
+	private String driverCores;
+
 	private int numExecutors;
 
 	private String[] arguments;
 
 	public RepeatStatus execute(StepContribution contribution, ChunkContext chunkContext) throws Exception {
 		SparkConf sparkConf = new SparkConf();
 		sparkConf.set("spark.yarn.jar", sparkAssemblyJar);
+		if (StringUtils.hasText(extraClassPath)) {
+			sparkConf.set("spark.driver.extraClassPath", extraClassPath);
+			sparkConf.set("spark.executor.extraClassPath", extraClassPath);
+		}
 		List<String> submitArgs = new ArrayList<String>();
 		if (StringUtils.hasText(appName)) {
 			submitArgs.add("--name");
@@ -82,10 +94,22 @@ public RepeatStatus execute(StepContribution contribution, ChunkContext chunkCon
 			submitArgs.add("--archives");
 			submitArgs.add(resourceArchives);
 		}
-		submitArgs.add("--executor-memory");
-		submitArgs.add(executorMemory);
 		submitArgs.add("--num-executors");
 		submitArgs.add("" + numExecutors);
+		submitArgs.add("--executor-memory");
+		submitArgs.add(executorMemory);
+		if (!StringUtils.hasText(executorCores)) {
+			submitArgs.add("--executor-cores");
+			submitArgs.add(executorCores);
+		}
+		if (!StringUtils.hasText(driverMemory)) {
+			submitArgs.add("--driver-memory");
+			submitArgs.add(driverMemory);
+		}
+		if (!StringUtils.hasText(driverCores)) {
+			submitArgs.add("--driver-cores");
+			submitArgs.add(driverCores);
+		}
 		for (String arg : arguments) {
 			submitArgs.add("--arg");
 			submitArgs.add(arg);
@@ -137,6 +161,10 @@ public void setSparkAssemblyJar(String sparkAssemblyJar) {
 		this.sparkAssemblyJar = sparkAssemblyJar;
 	}
 
+	public void setExtraClassPath(String extraClassPath) {
+		this.extraClassPath = extraClassPath;
+	}
+
 	public void setHadoopConfiguration(Configuration configuration) {
 		this.hadoopConfiguration = configuration;
 	}
@@ -165,6 +193,18 @@ public void setExecutorMemory(String executorMemory) {
 		this.executorMemory = executorMemory;
 	}
 
+	public void setExecutorCores(String executorCores) {
+		this.executorCores = executorCores;
+	}
+
+	public void setDriverMemory(String driverMemory) {
+		this.driverMemory = driverMemory;
+	}
+
+	public void setDriverCores(String driverCores) {
+		this.driverCores = driverCores;
+	}
+
 	public void setNumExecutors(int numExecutors) {
 		this.numExecutors = numExecutors;
 	}