Skip to content

[SPARK-17062][MESOS] add conf option to mesos dispatcher #14650

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
skonto wants to merge 5 commits into from
Closed

[SPARK-17062][MESOS] add conf option to mesos dispatcher #14650

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
db42e31
add conf option to mesos dispatcher
Aug 15, 2016
3532f5d
Merge branch 'master' into dipatcher_conf
skonto Sep 13, 2016
497e517
refactor
skonto Sep 13, 2016
d40b928
fix test
skonto Sep 14, 2016
946202d
fixes
skonto Oct 17, 2016
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
17 changes: 16 additions & 1 deletion core/src/main/scala/org/apache/spark/deploy/mesos/MesosClusterDispatcherArguments.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@
package org.apache.spark.deploy.mesos

import scala.annotation.tailrec
import scala.collection.mutable

import org.apache.spark.SparkConf
import org.apache.spark.util.{IntParam, Utils}
Expand All @@ -31,11 +32,19 @@ private[mesos] class MesosClusterDispatcherArguments(args: Array[String], conf:
var masterUrl: String = _
var zookeeperUrl: Option[String] = None
var propertiesFile: String = _
val cliSparkConfig = new mutable.ListBuffer[String]()

parse(args.toList)

propertiesFile = Utils.loadDefaultSparkProperties(conf, propertiesFile)

updateSparkConfigFromcliConfig()

private def updateSparkConfigFromcliConfig() : Unit = {
val properties = Utils.loadPropertiesFromString(cliSparkConfig.mkString("\n"))
Utils.updateSparkConfigFromProperties(conf, properties)
}

@tailrec
private def parse(args: List[String]): Unit = args match {
case ("--host" | "-h") :: value :: tail =>
Expand Down Expand Up @@ -73,6 +82,10 @@ private[mesos] class MesosClusterDispatcherArguments(args: Array[String], conf:
propertiesFile = value
parse(tail)

case ("--conf") :: value :: tail =>
cliSparkConfig += value

@mgummelt mgummelt Aug 19, 2016

Copy link
Copy Markdown

Choose a reason for hiding this comment

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

Let's make this consistent with SparkSubmitArguments, and parse these into a HashMap here.

parse(tail)

case ("--help") :: tail =>
printUsageAndExit(0)

Expand Down Expand Up @@ -102,7 +115,9 @@ private[mesos] class MesosClusterDispatcherArguments(args: Array[String], conf:
" -z --zk ZOOKEEPER Comma delimited URLs for connecting to \n" +
" Zookeeper for persistence\n" +
" --properties-file FILE Path to a custom Spark properties file.\n" +
" Default is conf/spark-defaults.conf.")
" Default is conf/spark-defaults.conf \n" +
" --conf PROP=VALUE Arbitrary Spark configuration property.\n" +

@skonto skonto Aug 19, 2016

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

I just copied the help output of spark-submit. For consistency reasons lets keep it that way.

" Takes precedence over defined properties in properties-file.")
// scalastyle:on println
System.exit(exitCode)
}
Expand Down
20 changes: 20 additions & 0 deletions core/src/main/scala/org/apache/spark/util/Utils.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1962,6 +1962,26 @@ private[spark] object Utils extends Logging {
path
}

/**
* Updates Spark config with properties from a set of Properties.
* Provided properties have the highest priority.
*/
def updateSparkConfigFromProperties(conf: SparkConf, properties: Properties): Unit = {

@mgummelt mgummelt Aug 19, 2016

Copy link
Copy Markdown

Choose a reason for hiding this comment

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

I think you can fold this into loadFromSystemProperties in SparkConf

@skonto skonto Aug 19, 2016 •
edited
Loading

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

Yes with the exception of this line: sys.props.getOrElseUpdate(k, v) for which again as I said I am not sure and which is missing from loadFromSystemProperties.

@mgummelt mgummelt Aug 19, 2016

Copy link
Copy Markdown

Choose a reason for hiding this comment

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

I'm pretty sure we don't need to set system properties. They're only used to set SparkConf anyway, I think.

@skonto skonto Aug 21, 2016

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

Ok I will remove it and use that function in SparkConf.

properties.asScala.filter { case (k, v) =>
k.startsWith("spark.")

@vanzin vanzin Sep 21, 2016

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.

You can move this to the foreach body.

On a side note, this whole method feels a little funny, but I have to see how it's used.

@skonto skonto Sep 22, 2016 •
edited
Loading

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

I just re-used the code found here.

}.foreach { case (k, v) =>
conf.set(k, v)
sys.props.getOrElseUpdate(k, v)

@mgummelt mgummelt Aug 19, 2016

Copy link
Copy Markdown

Choose a reason for hiding this comment

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

Why set them as System properties as well?

@skonto skonto Aug 19, 2016

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

For consistency if you check a few lines above with other methods which do the same.

}
}

/** Loads properties from a String */
def loadPropertiesFromString(input: String) : Properties = {

@mgummelt mgummelt Aug 19, 2016

Copy link
Copy Markdown

Choose a reason for hiding this comment

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

I don't think we need this. We can parse the input into either a Properties object or a HashMap in the above file.

@skonto skonto Aug 19, 2016 •
edited
Loading

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

The reason I did it that way is to avoid splitting the "property=value" thing myself in the first place... I let the load method to do that so it can throw the needed exception if something is wrong. Its more elegant that way to me although more code added.
Check the logic I need to replicate here.

Ideally since the --conf is everywhere such util functions from SparkSubmitArguments should be re-factored out as common functionality to be re-used elsewhere so error messages are consistent as well.

@skonto skonto Aug 19, 2016 •
edited
Loading

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

@mgummelt, @srowen If this is a blocker let me know. Let's discuss also if this refactoring needs to be addressed here (if we agree there is a need for it, which to me is needed but not for now). We could agree on a consistent way of setting properties across the code base and implement it in a different jira issue.

@mgummelt mgummelt Aug 19, 2016

Copy link
Copy Markdown

Choose a reason for hiding this comment

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

My only issue is that this is inconsistent with how SparkSubmitArguments parses {{--conf}}. So if you prefer this approach, let's refactor SparkSubmitArguments to use this function as well.

@skonto skonto Aug 21, 2016

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

Ok I will try to use the same function either the one in SparkSubmitArguments or load. The error message must remain the same one way or another.

val properties = new Properties()
properties.load(new StringReader(input))
properties
}

/** Load properties present in the given file. */
def getPropertiesFromFile(filename: String): Map[String, String] = {
val file = new File(filename)
Expand Down