@Override
	public RepeatStatus execute(StepContribution contribution,
			ChunkContext chunkContext) throws Exception {

		StepExecution stepExecution = chunkContext.getStepContext().getStepExecution();
		ExitStatus exitStatus = stepExecution.getExitStatus();

		String moduleHome = environment.getProperty(MODULE_HOME);
		Assert.notNull(moduleHome, "Module home must not be null.");
		Resource[] resources = resolver.getResources(moduleHome + LIB_PATTERN);
		ArrayList<String> dependencies = new ArrayList<String>();
		for (int i = 0; i < resources.length; i++) {
			dependencies.add(resources[i].getURL().getFile());
		}
		ArrayList<String> args = new ArrayList<String>();
		if (StringUtils.hasText(name)) {
			args.add("--name");
			args.add(name);
		}
		args.add("--class");
		args.add(mainClass);
		args.add("--master");
		args.add(master);
		args.add("--deploy-mode");
		args.add("client");
		if (StringUtils.hasText(conf)) {
			Collection<String> configs = StringUtils.commaDelimitedListToSet(conf);
			for (String config : configs) {
				args.add("--conf");
				args.add(config.trim());
			}
		}
		if (StringUtils.hasText(files)) {
			args.add("--files");
			args.add(files);
		}
		args.add("--jars");
		args.add(StringUtils.collectionToCommaDelimitedString(dependencies));
		if (StringUtils.hasText(appJar)) {
			args.add(appJar);
		}
		if (StringUtils.hasText(programArgs)) {
			args.addAll(StringUtils.commaDelimitedListToSet(programArgs));
		}
		List<String> sparkCommand = new ArrayList<String>();
		sparkCommand.add("java");
		sparkCommand.add(SPARK_SUBMIT_CLASS);
		sparkCommand.addAll(args);

		URLClassLoader serverClassLoader;
		URLClassLoader taskletClassLoader;
		try {
			serverClassLoader = (URLClassLoader) Class.forName("org.springframework.xd.dirt.core.Job").getClassLoader();
			taskletClassLoader = (URLClassLoader) this.getClass().getClassLoader();
		}
		catch (Exception e) {
			throw new IllegalStateException("Unable to determine classpath from ClassLoader.", e);
		}

		List<String> classPath = new ArrayList<String>();
		for (URL url : serverClassLoader.getURLs()) {
			String file = url.getFile().split("\\!/", 2)[0];
			if (file.endsWith(".jar")) {
				classPath.add(file);
			}
		}
		for (URL url : taskletClassLoader.getURLs()) {
			String file = url.getFile().split("\\!/", 2)[0];
			if (file.endsWith(".jar") && !classPath.contains(file)) {
				classPath.add(file);
			}
		}
		StringBuilder classPathBuilder = new StringBuilder();
		String separator = System.getProperty("path.separator");
		for (String url : classPath) {
			if (!url.contains("logback")) {
				if (classPathBuilder.length() > 0) {
					classPathBuilder.append(separator);
				}
				classPathBuilder.append(url);
			}
		}

		ProcessBuilder pb = new ProcessBuilder(sparkCommand).redirectErrorStream(true);
		Map<String, String> env = pb.environment();
		env.put("CLASSPATH", classPathBuilder.toString());
		String msg = "Spark application '" + mainClass + "' is being launched";
		StringBuilder sparkCommandString = new StringBuilder();
		for (String cmd : sparkCommand) {
			sparkCommandString.append(cmd).append(" ");
		}
		stepExecution.getExecutionContext().putString("spark.command", sparkCommandString.toString());
		List<String> sparkLog = new ArrayList<String>();
		try {
			Process p = pb.start();
			p.waitFor();
			exitCode = p.exitValue();
			msg = "Spark application '" + mainClass + "' finished with exit code: " + exitCode;
			if (exitCode == 0) {
				logger.info(msg);
			}
			else {
				logger.error(msg);
			}
			sparkLog = getProcessOutput(p);
			p.destroy();
		}
		catch (IOException e) {
			msg = "Starting Spark application '" + mainClass + "' failed with: " + e;
			logger.error(msg);
		}
		catch (InterruptedException e) {
			msg = "Executing Spark application '" + mainClass + "' failed with: " + e;
			logger.error(msg);
		}
		finally {
			printLog(sparkLog, exitCode);
			StringBuilder firstException = new StringBuilder();
			if (exitCode != 0) {
				for (String line : sparkLog) {
					if (firstException.length() == 0) {
						if (line.contains("Exception")) {
							firstException.append(line).append("\n");
						}
					}
					else {
						if (line.startsWith("\t")) {
							firstException.append(line).append("\n");
						}
						else {
							break;
						}
					}
				}
				if (firstException.length() > 0) {
					msg = msg + "\n" + firstException.toString();
				}
			}
			StringBuilder sparkLogMessages = new StringBuilder();
			for (String line : sparkLog) {
				sparkLogMessages.append(line).append("</br>");
			}
			stepExecution.getExecutionContext().putString("spark.log", sparkLogMessages.toString());
			stepExecution.setExitStatus(exitStatus.addExitDescription(msg));
		}

		return RepeatStatus.FINISHED;
	}
Program arguments for the application main class.
extensions/spring-xd-extension-spark/src/main/java/org/springframework/xd/spark/tasklet/SparkTasklet.java