Skip to content

[SPARK-22404][YARN] Provide an option to use unmanaged AM in yarn-client mode #19616

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
devaraj-kavali wants to merge 16 commits into from
Closed

[SPARK-22404][YARN] Provide an option to use unmanaged AM in yarn-client mode #19616

Show file tree
Hide file tree
Changes from 9 commits
Commits
Show all changes
16 commits
Select commit Hold shift + click to select a range
e51f99e
[SPARK-22404][YARN] Provide an option to use unmanaged AM in yarn-client
Oct 31, 2017
764c302
Merge branch 'master' into SPARK-22404
Dec 13, 2017
640013b
Adding Client changes missed while merging master
Dec 13, 2017
cba0c6d
Resolving ApplicationMaster conflicts with the merge
Dec 14, 2017
19b6c3a
Fixed the review comments
Feb 15, 2018
ce94235
Merge branch 'master' into SPARK-22404
Feb 15, 2018
0921f7a
Merge branch 'master' into SPARK-22404
Jul 16, 2018
837d25f
Merge branch 'master' into SPARK-22404
Dec 12, 2018
65aeba9
Fixing the review comments
Dec 14, 2018
93b016f
Addressing review comments
Dec 20, 2018
dc31940
Review comments addressing
Jan 10, 2019
23ad9de
Merge branch 'master' into SPARK-22404
Jan 10, 2019
1c02b7d
Merge branch 'master' into SPARK-22404
Jan 10, 2019
2429e19
Changes to fix issues with master merge
Jan 15, 2019
6854fc4
Updated with review comments fix and added a test
Jan 24, 2019
3b377af
Updated to use YARN_UNMANAGED_AM.key
Jan 25, 2019
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
129 changes: 84 additions & 45 deletions resource-managers/yarn/src/main/scala/org/apache/spark/deploy/yarn/ApplicationMaster.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -35,6 +35,7 @@ import org.apache.hadoop.yarn.api._
import org.apache.hadoop.yarn.api.records._
import org.apache.hadoop.yarn.conf.YarnConfiguration
import org.apache.hadoop.yarn.exceptions.ApplicationAttemptNotFoundException
import org.apache.hadoop.yarn.server.webproxy.ProxyUriUtils
import org.apache.hadoop.yarn.util.{ConverterUtils, Records}

import org.apache.spark._
Expand All @@ -53,35 +54,20 @@ import org.apache.spark.util._
/**
* Common application master functionality for Spark on Yarn.
*/
private[spark] class ApplicationMaster(args: ApplicationMasterArguments) extends Logging {
private[spark] class ApplicationMaster(
args: ApplicationMasterArguments,
sparkConf: SparkConf,
yarnConf: YarnConfiguration) extends Logging {

// TODO: Currently, task to container is computed once (TaskSetManager) - which need not be
// optimal as more containers are available. Might need to handle this better.

private val isClusterMode = args.userClass != null

private val sparkConf = new SparkConf()
if (args.propertiesFile != null) {
Utils.getPropertiesFromFile(args.propertiesFile).foreach { case (k, v) =>
sparkConf.set(k, v)
}
}

private val securityMgr = new SecurityManager(sparkConf)

private var metricsSystem: Option[MetricsSystem] = None

// Set system properties for each config entry. This covers two use cases:
// - The default configuration stored by the SparkHadoopUtil class
// - The user application creating a new SparkConf in cluster mode
//
// Both cases create a new SparkConf object which reads these configs from system properties.
sparkConf.getAll.foreach { case (k, v) =>
sys.props(k) = v
}

private val yarnConf = new YarnConfiguration(SparkHadoopUtil.newConfiguration(sparkConf))

private val userClassLoader = {
val classpath = Client.getUserClasspath(sparkConf)
val urls = classpath.map { entry =>
Expand Down Expand Up @@ -169,8 +155,6 @@ private[spark] class ApplicationMaster(args: ApplicationMasterArguments) extends
// Next wait interval before allocator poll.
private var nextAllocationInterval = initialAllocationInterval

private var rpcEnv: RpcEnv = null

// In cluster mode, used to tell the AM when the user's SparkContext has been initialized.
private val sparkContextPromise = Promise[SparkContext]()

Expand Down Expand Up @@ -234,20 +218,26 @@ private[spark] class ApplicationMaster(args: ApplicationMasterArguments) extends
resources.toMap
}

def getAttemptId(): ApplicationAttemptId = {
client.getAttemptId()
def getAttemptId(sparkConf: SparkConf): ApplicationAttemptId = {
client.getAttemptId(sparkConf)
}

final def run(): Int = {
doAsUser {
runImpl()
runImpl(
if (isClusterMode) {
runDriver()
} else {
runExecutorLauncher()
}
)
}
exitCode
}

private def runImpl(): Unit = {
private def runImpl(opBlock: => Unit): Unit = {

@vanzin vanzin Dec 18, 2018

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

There are things in this method that don't look right when you think about an unmanaged AM.

e.g., overriding spark.master, spark.ui.port, etc, look wrong.

The handling of app attempts also seems wrong, since with an unmanaged AM you don't have multiple attempts. Even the shutdown hooks seems a bit out of place.

Seems to me it would be easier not to try to use this method for the unmanaged AM.

@devaraj-kavali devaraj-kavali Dec 20, 2018

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

refactored this code

try {
val appAttemptId = client.getAttemptId()
val appAttemptId = client.getAttemptId(sparkConf)

var attemptID: Option[String] = None

Expand Down Expand Up @@ -277,7 +267,7 @@ private[spark] class ApplicationMaster(args: ApplicationMasterArguments) extends
val priority = ShutdownHookManager.SPARK_CONTEXT_SHUTDOWN_PRIORITY - 1
ShutdownHookManager.addShutdownHook(priority) { () =>
val maxAppAttempts = client.getMaxRegAttempts(sparkConf, yarnConf)
val isLastAttempt = client.getAttemptId().getAttemptId() >= maxAppAttempts
val isLastAttempt = client.getAttemptId(sparkConf).getAttemptId() >= maxAppAttempts

if (!finished) {
// The default state of ApplicationMaster is failed if it is invoked by shut down hook.
Expand All @@ -299,11 +289,8 @@ private[spark] class ApplicationMaster(args: ApplicationMasterArguments) extends
}
}

if (isClusterMode) {
runDriver()
} else {
runExecutorLauncher()
}
opBlock

} catch {
case e: Exception =>
// catch everything else if not specifically handled
Expand All @@ -324,6 +311,23 @@ private[spark] class ApplicationMaster(args: ApplicationMasterArguments) extends
}
}

def runUnmanaged(clientRpcEnv: RpcEnv): Unit = {
runImpl {
val driverRef = clientRpcEnv.setupEndpointRef(
RpcAddress(sparkConf.get("spark.driver.host"),
sparkConf.get("spark.driver.port").toInt),
YarnSchedulerBackend.ENDPOINT_NAME)
// The client-mode AM doesn't listen for incoming connections, so report an invalid port.
registerAM(
Utils.localHostName, -1, sparkConf, sparkConf.getOption("spark.driver.appUIAddress"))
addAmIpFilter(Some(driverRef))
createAllocator(driverRef, sparkConf, clientRpcEnv)

// In client mode the actor will stop the reporter thread.

@vanzin vanzin Dec 18, 2018

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

actor?

@devaraj-kavali devaraj-kavali Dec 20, 2018

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

removed this as part of the above comment refactor

reporterThread.join()
}
}

/**
* Set the default final application status for client mode to UNDEFINED to handle
* if YARN HA restarts the application so that it properly retries. Set the final
Expand Down Expand Up @@ -405,17 +409,21 @@ private[spark] class ApplicationMaster(args: ApplicationMasterArguments) extends
port: Int,
_sparkConf: SparkConf,
uiAddress: Option[String]): Unit = {
val appId = client.getAttemptId().getApplicationId().toString()
val attemptId = client.getAttemptId().getAttemptId().toString()
val appAttempt = client.getAttemptId(_sparkConf)
val appId = appAttempt.getApplicationId().toString()
val attemptId = appAttempt.getAttemptId().toString()
val historyAddress = ApplicationMaster
.getHistoryServerAddress(_sparkConf, yarnConf, appId, attemptId)

client.register(host, port, yarnConf, _sparkConf, uiAddress, historyAddress)
registered = true
}

private def createAllocator(driverRef: RpcEndpointRef, _sparkConf: SparkConf): Unit = {
val appId = client.getAttemptId().getApplicationId().toString()
private def createAllocator(
driverRef: RpcEndpointRef,
_sparkConf: SparkConf,
rpcEnv: RpcEnv): Unit = {
val appId = client.getAttemptId(_sparkConf).getApplicationId().toString()
val driverUrl = RpcEndpointAddress(driverRef.address.host, driverRef.address.port,
CoarseGrainedSchedulerBackend.ENDPOINT_NAME).toString

Expand Down Expand Up @@ -466,7 +474,7 @@ private[spark] class ApplicationMaster(args: ApplicationMasterArguments) extends
val sc = ThreadUtils.awaitResult(sparkContextPromise.future,
Duration(totalWaitTime, TimeUnit.MILLISECONDS))
if (sc != null) {
rpcEnv = sc.env.rpcEnv
val rpcEnv = sc.env.rpcEnv

val userConf = sc.getConf
val host = userConf.get("spark.driver.host")
Expand All @@ -476,7 +484,7 @@ private[spark] class ApplicationMaster(args: ApplicationMasterArguments) extends
val driverRef = rpcEnv.setupEndpointRef(
RpcAddress(host, port),
YarnSchedulerBackend.ENDPOINT_NAME)
createAllocator(driverRef, userConf)
createAllocator(driverRef, userConf, rpcEnv)
} else {
// Sanity check; should never happen in normal operation, since sc should only be null
// if the user app did not create a SparkContext.
Expand All @@ -500,7 +508,7 @@ private[spark] class ApplicationMaster(args: ApplicationMasterArguments) extends
private def runExecutorLauncher(): Unit = {
val hostname = Utils.localHostName
val amCores = sparkConf.get(AM_CORES)
rpcEnv = RpcEnv.create("sparkYarnAM", hostname, hostname, -1, sparkConf, securityMgr,
val rpcEnv = RpcEnv.create("sparkYarnAM", hostname, hostname, -1, sparkConf, securityMgr,
amCores, true)

// The client-mode AM doesn't listen for incoming connections, so report an invalid port.
Expand All @@ -513,7 +521,7 @@ private[spark] class ApplicationMaster(args: ApplicationMasterArguments) extends
RpcAddress(driverHost, driverPort),
YarnSchedulerBackend.ENDPOINT_NAME)
addAmIpFilter(Some(driverRef))
createAllocator(driverRef, sparkConf)
createAllocator(driverRef, sparkConf, rpcEnv)

// In client mode the actor will stop the reporter thread.
reporterThread.join()
Expand Down Expand Up @@ -617,7 +625,14 @@ private[spark] class ApplicationMaster(args: ApplicationMasterArguments) extends
try {
val preserveFiles = sparkConf.get(PRESERVE_STAGING_FILES)
if (!preserveFiles) {
stagingDirPath = new Path(System.getenv("SPARK_YARN_STAGING_DIR"))
var stagingDir = System.getenv("SPARK_YARN_STAGING_DIR")

@vanzin vanzin Dec 18, 2018

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

val stagingDir = sys.props.get("...").getOrElse { ... }

@devaraj-kavali devaraj-kavali Dec 20, 2018

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Made the change to pass the stagingDir from Client.scala

if (stagingDir == null) {
val appStagingBaseDir = sparkConf.get(STAGING_DIR).map { new Path(_) }

@vanzin vanzin Dec 18, 2018

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This looks similar to the logic in Client.scala. Maybe the value calculated there should be plumbed through, instead of adding this code.

@devaraj-kavali devaraj-kavali Dec 20, 2018

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Made the change to pass the stagingDir from Client.scala

.getOrElse(FileSystem.get(yarnConf).getHomeDirectory())
stagingDir = appStagingBaseDir.toString + Path.SEPARATOR +
getAttemptId(sparkConf).getApplicationId.toString
}
stagingDirPath = new Path(stagingDir)
logInfo("Deleting staging directory " + stagingDirPath)
val fs = stagingDirPath.getFileSystem(yarnConf)
fs.delete(stagingDirPath, true)
Expand All @@ -630,7 +645,7 @@ private[spark] class ApplicationMaster(args: ApplicationMasterArguments) extends

/** Add the Yarn IP filter that is required for properly securing the UI. */
private def addAmIpFilter(driver: Option[RpcEndpointRef]) = {
val proxyBase = System.getenv(ApplicationConstants.APPLICATION_WEB_PROXY_BASE_ENV)
val proxyBase = getProxyBase
val amFilter = "org.apache.hadoop.yarn.server.webproxy.amfilter.AmIpFilter"
val params = client.getAmIpFilterParams(yarnConf, proxyBase)
driver match {
Expand All @@ -643,6 +658,14 @@ private[spark] class ApplicationMaster(args: ApplicationMasterArguments) extends
}
}

private def getProxyBase: String = {
var proxyBase = System.getenv(ApplicationConstants.APPLICATION_WEB_PROXY_BASE_ENV)
if (proxyBase == null) {
proxyBase = ProxyUriUtils.getPath(getAttemptId(sparkConf).getApplicationId)
}
proxyBase
}

/**
* Start the user class, which contains the spark driver, in a separate Thread.
* If the main routine exits cleanly or exits with System.exit(N) for any N
Expand Down Expand Up @@ -791,16 +814,32 @@ object ApplicationMaster extends Logging {
def main(args: Array[String]): Unit = {
SignalUtils.registerLogger(log)
val amArgs = new ApplicationMasterArguments(args)
master = new ApplicationMaster(amArgs)
val sparkConf = new SparkConf()
if (amArgs.propertiesFile != null) {
Utils.getPropertiesFromFile(amArgs.propertiesFile).foreach { case (k, v) =>
sparkConf.set(k, v)
}
}
// Set system properties for each config entry. This covers two use cases:
// - The default configuration stored by the SparkHadoopUtil class
// - The user application creating a new SparkConf in cluster mode
//
// Both cases create a new SparkConf object which reads these configs from system properties.
sparkConf.getAll.foreach { case (k, v) =>
sys.props(k) = v
}

val yarnConf = new YarnConfiguration(SparkHadoopUtil.newConfiguration(sparkConf))
master = new ApplicationMaster(amArgs, sparkConf, yarnConf)
System.exit(master.run())
}

private[spark] def sparkContextInitialized(sc: SparkContext): Unit = {
master.sparkContextInitialized(sc)
}

private[spark] def getAttemptId(): ApplicationAttemptId = {
master.getAttemptId
private[spark] def getAttemptId(sparkConf: SparkConf): ApplicationAttemptId = {
master.getAttemptId(sparkConf)
}

private[spark] def getHistoryServerAddress(
Expand Down
48 changes: 42 additions & 6 deletions resource-managers/yarn/src/main/scala/org/apache/spark/deploy/yarn/Client.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -34,7 +34,7 @@ import com.google.common.io.Files
import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.fs._
import org.apache.hadoop.fs.permission.FsPermission
import org.apache.hadoop.io.DataOutputBuffer
import org.apache.hadoop.io.{DataOutputBuffer, Text}
import org.apache.hadoop.mapreduce.MRJobConfig
import org.apache.hadoop.security.{Credentials, UserGroupInformation}
import org.apache.hadoop.util.StringUtils
Expand All @@ -45,6 +45,7 @@ import org.apache.hadoop.yarn.api.records._
import org.apache.hadoop.yarn.client.api.{YarnClient, YarnClientApplication}
import org.apache.hadoop.yarn.conf.YarnConfiguration
import org.apache.hadoop.yarn.exceptions.ApplicationNotFoundException
import org.apache.hadoop.yarn.security.AMRMTokenIdentifier
import org.apache.hadoop.yarn.util.Records

import org.apache.spark.{SecurityManager, SparkConf, SparkException}
Expand All @@ -54,11 +55,13 @@ import org.apache.spark.deploy.yarn.security.YARNHadoopDelegationTokenManager
import org.apache.spark.internal.Logging
import org.apache.spark.internal.config._
import org.apache.spark.launcher.{LauncherBackend, SparkAppHandle, YarnCommandBuilderUtils}
import org.apache.spark.rpc.RpcEnv
import org.apache.spark.util.{CallerContext, Utils}

private[spark] class Client(
val args: ClientArguments,
val sparkConf: SparkConf)
val sparkConf: SparkConf,
val rpcEnv: RpcEnv)
extends Logging {

import Client._
Expand All @@ -69,6 +72,9 @@ private[spark] class Client(

private val isClusterMode = sparkConf.get("spark.submit.deployMode", "client") == "cluster"

private val isClientUnmanagedAMEnabled = sparkConf.get(YARN_UNMANAGED_AM) && !isClusterMode
private var amServiceStarted = false

@vanzin vanzin Jan 17, 2019

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Do you need this extra flag? Could you just check if appMaster != null?


// AM related configurations
private val amMemory = if (isClusterMode) {
sparkConf.get(DRIVER_MEMORY).toInt
Expand Down Expand Up @@ -297,7 +303,10 @@ private[spark] class Client(
"does not support it", e)
}
}

if (isClientUnmanagedAMEnabled) {
// Set Unmanaged AM to true in Application Submission Context
appContext.setUnmanagedAM(true)

@vanzin vanzin Dec 18, 2018

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

appContext.setUnmanagedAM(isClientUnmanagedAMEnabled)

Which also makes the comment unnecessary.

@devaraj-kavali devaraj-kavali Dec 20, 2018

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

updated

}
appContext
}

Expand Down Expand Up @@ -660,7 +669,9 @@ private[spark] class Client(
// Clear the cache-related entries from the configuration to avoid them polluting the
// UI's environment page. This works for client mode; for cluster mode, this is handled
// by the AM.
CACHE_CONFIGS.foreach(sparkConf.remove)
if (!isClientUnmanagedAMEnabled) {

@vanzin vanzin Feb 1, 2018

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Why is this needed in the new mode?

@devaraj-kavali devaraj-kavali Feb 15, 2018

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It is clearing the classpath entries and leading to this error in Executors.

Error: Could not find or load main class org.apache.spark.executor.CoarseGrainedExecutorBackend

@vanzin vanzin Dec 10, 2018

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think this is happening because you're starting the AM after these are removed from the conf. Should probably juggle things around or change how these are provided to the AM, since these configs are super noisy and shouldn't really show up in the UI.

CACHE_CONFIGS.foreach(sparkConf.remove)
}

localResources
}
Expand Down Expand Up @@ -1098,14 +1109,39 @@ private[spark] class Client(
if (returnOnRunning && state == YarnApplicationState.RUNNING) {
return createAppReport(report)
}

if (state == YarnApplicationState.ACCEPTED && isClientUnmanagedAMEnabled
&& !amServiceStarted && report.getAMRMToken != null) {

@vanzin vanzin Dec 18, 2018

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

indent one more level

@devaraj-kavali devaraj-kavali Dec 20, 2018

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

updated

amServiceStarted = true
startApplicationMasterService(report)
}
lastState = state
}

// Never reached, but keeps compiler happy
throw new SparkException("While loop is depleted! This should never happen...")
}

private def startApplicationMasterService(report: ApplicationReport) = {

@vanzin vanzin Dec 20, 2018

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

: Unit =

But given you should be explicitly stopping the AM, this should probably return the AM itself.

// Add AMRMToken to establish connection between RM and AM
val token = report.getAMRMToken
val amRMToken: org.apache.hadoop.security.token.Token[AMRMTokenIdentifier] =

@vanzin vanzin Feb 1, 2018

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Why do you need to make this copy? Isn't the Token above enough?

@devaraj-kavali devaraj-kavali Feb 15, 2018

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

report.getAMRMToken gives org.apache.hadoop.yarn.api.records.Token type instance, but currentUGI.addToken expects org.apache.hadoop.security.token.Token type instance.

new org.apache.hadoop.security.token.Token[AMRMTokenIdentifier](
token.getIdentifier().array(), token.getPassword().array,
new Text(token.getKind()), new Text(token.getService()))
val currentUGI = UserGroupInformation.getCurrentUser
currentUGI.addToken(amRMToken)

sparkConf.set("spark.yarn.containerId",
ContainerId.newContainerId(report.getCurrentApplicationAttemptId, 1).toString)

@vanzin vanzin Dec 18, 2018

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Won't this name be the same as the first executor created by the app?

I'd rather special-case getContainerId to return some baked-in string when the env variable is not set.

@devaraj-kavali devaraj-kavali Dec 20, 2018

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

made the change to pass the appAttemptId from Client.scala

// Start Application Service in a separate thread and continue with application monitoring
val amService = new Thread("Unmanaged Application Master Service") {
override def run(): Unit = new ApplicationMaster(new ApplicationMasterArguments(Array.empty),

@vanzin vanzin Dec 18, 2018

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This is a pretty long line. Break it down.

@devaraj-kavali devaraj-kavali Dec 20, 2018

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

updated it.

sparkConf, hadoopConf).runUnmanaged(rpcEnv)
}
amService.setDaemon(true)
amService.start()
}

private def formatReportDetails(report: ApplicationReport): String = {
val details = Seq[(String, String)](
("client token", getClientToken(report)),
Expand Down Expand Up @@ -1527,7 +1563,7 @@ private[spark] class YarnClusterApplication extends SparkApplication {
conf.remove("spark.jars")
conf.remove("spark.files")

new Client(new ClientArguments(args), conf).run()
new Client(new ClientArguments(args), conf, null).run()
}

}
Expand Down
Loading