Skip to content

Commit 6139631

Browse files
committed
fix --packages for mesos
1 parent a9339db commit 6139631

1 file changed

Lines changed: 50 additions & 39 deletions

File tree

core/src/main/scala/org/apache/spark/deploy/SparkSubmit.scala

Lines changed: 50 additions & 39 deletions
Original file line numberDiff line numberDiff line change
@@ -269,6 +269,25 @@ object SparkSubmit extends CommandLineUtils {
269269
}
270270
}
271271

272+
// Fail fast, the following modes are not supported or applicable
273+
(clusterManager, deployMode) match {
274+
case (STANDALONE, CLUSTER) if args.isPython =>
275+
printErrorAndExit("Cluster deploy mode is currently not supported for python " +
276+
"applications on standalone clusters.")
277+
case (STANDALONE, CLUSTER) if args.isR =>
278+
printErrorAndExit("Cluster deploy mode is currently not supported for R " +
279+
"applications on standalone clusters.")
280+
case (LOCAL, CLUSTER) =>
281+
printErrorAndExit("Cluster deploy mode is not compatible with master \"local\"")
282+
case (_, CLUSTER) if isShell(args.primaryResource) =>
283+
printErrorAndExit("Cluster deploy mode is not applicable to Spark shells.")
284+
case (_, CLUSTER) if isSqlShell(args.mainClass) =>
285+
printErrorAndExit("Cluster deploy mode is not applicable to Spark SQL shell.")
286+
case (_, CLUSTER) if isThriftServer(args.mainClass) =>
287+
printErrorAndExit("Cluster deploy mode is not applicable to Spark Thrift server.")
288+
case _ =>
289+
}
290+
272291
// Update args.deployMode if it is null. It will be passed down as a Spark property later.
273292
(args.deployMode, deployMode) match {
274293
case (null, CLIENT) => args.deployMode = "client"
@@ -278,36 +297,40 @@ object SparkSubmit extends CommandLineUtils {
278297
val isYarnCluster = clusterManager == YARN && deployMode == CLUSTER
279298
val isMesosCluster = clusterManager == MESOS && deployMode == CLUSTER
280299

281-
// Resolve maven dependencies if there are any and add classpath to jars. Add them to py-files
282-
// too for packages that include Python code
283-
val exclusions: Seq[String] =
300+
if (!isMesosCluster) {
301+
// Resolve maven dependencies if there are any and add classpath to jars. Add them to py-files
302+
// too for packages that include Python code
303+
val exclusions: Seq[String] =
284304
if (!StringUtils.isBlank(args.packagesExclusions)) {
285305
args.packagesExclusions.split(",")
286306
} else {
287307
Nil
288308
}
289309

290-
// Create the IvySettings, either load from file or build defaults
291-
val ivySettings = args.sparkProperties.get("spark.jars.ivySettings").map { ivySettingsFile =>
292-
SparkSubmitUtils.loadIvySettings(ivySettingsFile, Option(args.repositories),
293-
Option(args.ivyRepoPath))
294-
}.getOrElse {
295-
SparkSubmitUtils.buildIvySettings(Option(args.repositories), Option(args.ivyRepoPath))
296-
}
310+
// Create the IvySettings, either load from file or build defaults
311+
val ivySettings = args.sparkProperties.get("spark.jars.ivySettings").map { ivySettingsFile =>
312+
SparkSubmitUtils.loadIvySettings(ivySettingsFile, Option(args.repositories),
313+
Option(args.ivyRepoPath))
314+
}.getOrElse {
315+
SparkSubmitUtils.buildIvySettings(Option(args.repositories), Option(args.ivyRepoPath))
316+
}
297317

298-
val resolvedMavenCoordinates = SparkSubmitUtils.resolveMavenCoordinates(args.packages,
299-
ivySettings, exclusions = exclusions)
300-
if (!StringUtils.isBlank(resolvedMavenCoordinates)) {
301-
args.jars = mergeFileLists(args.jars, resolvedMavenCoordinates)
302-
if (args.isPython) {
303-
args.pyFiles = mergeFileLists(args.pyFiles, resolvedMavenCoordinates)
318+
val resolvedMavenCoordinates = SparkSubmitUtils.resolveMavenCoordinates(args.packages,
319+
ivySettings, exclusions = exclusions)
320+
321+
322+
if (!StringUtils.isBlank(resolvedMavenCoordinates)) {
323+
args.jars = mergeFileLists(args.jars, resolvedMavenCoordinates)
324+
if (args.isPython) {
325+
args.pyFiles = mergeFileLists(args.pyFiles, resolvedMavenCoordinates)
326+
}
304327
}
305-
}
306328

307-
// install any R packages that may have been passed through --jars or --packages.
308-
// Spark Packages may contain R source code inside the jar.
309-
if (args.isR && !StringUtils.isBlank(args.jars)) {
310-
RPackageUtils.checkAndBuildRPackage(args.jars, printStream, args.verbose)
329+
// install any R packages that may have been passed through --jars or --packages.
330+
// Spark Packages may contain R source code inside the jar.
331+
if (args.isR && !StringUtils.isBlank(args.jars)) {
332+
RPackageUtils.checkAndBuildRPackage(args.jars, printStream, args.verbose)
333+
}
311334
}
312335

313336
// In client mode, download remote files.
@@ -339,24 +362,6 @@ object SparkSubmit extends CommandLineUtils {
339362
}
340363
}
341364

342-
// The following modes are not supported or applicable
343-
(clusterManager, deployMode) match {
344-
case (STANDALONE, CLUSTER) if args.isPython =>
345-
printErrorAndExit("Cluster deploy mode is currently not supported for python " +
346-
"applications on standalone clusters.")
347-
case (STANDALONE, CLUSTER) if args.isR =>
348-
printErrorAndExit("Cluster deploy mode is currently not supported for R " +
349-
"applications on standalone clusters.")
350-
case (LOCAL, CLUSTER) =>
351-
printErrorAndExit("Cluster deploy mode is not compatible with master \"local\"")
352-
case (_, CLUSTER) if isShell(args.primaryResource) =>
353-
printErrorAndExit("Cluster deploy mode is not applicable to Spark shells.")
354-
case (_, CLUSTER) if isSqlShell(args.mainClass) =>
355-
printErrorAndExit("Cluster deploy mode is not applicable to Spark SQL shell.")
356-
case (_, CLUSTER) if isThriftServer(args.mainClass) =>
357-
printErrorAndExit("Cluster deploy mode is not applicable to Spark Thrift server.")
358-
case _ =>
359-
}
360365

361366
// If we're running a python app, set the main class to our specific python runner
362367
if (args.isPython && deployMode == CLIENT) {
@@ -464,6 +469,12 @@ object SparkSubmit extends CommandLineUtils {
464469
OptionAssigner(args.driverExtraLibraryPath, ALL_CLUSTER_MGRS, ALL_DEPLOY_MODES,
465470
sysProp = "spark.driver.extraLibraryPath"),
466471

472+
// Mesos only - propagate attributes for dependency resolution at the driver side
473+
OptionAssigner(args.packages, MESOS, CLUSTER, sysProp = "spark.jars.packages"),
474+
OptionAssigner(args.repositories, MESOS, CLUSTER, sysProp = "spark.jars.repositories"),
475+
OptionAssigner(args.ivyRepoPath, MESOS, CLUSTER, sysProp = "spark.jars.ivy"),
476+
OptionAssigner(args.packagesExclusions, MESOS, CLUSTER, sysProp = "spark.jars.excludes"),
477+
467478
// Yarn only
468479
OptionAssigner(args.queue, YARN, ALL_DEPLOY_MODES, sysProp = "spark.yarn.queue"),
469480
OptionAssigner(args.numExecutors, YARN, ALL_DEPLOY_MODES,

0 commit comments

Comments
 (0)