[SPARK-20001] Conda Runner & full Python conda support by dansanduleac · Pull Request #115 · palantir/spark

dansanduleac · 2017-03-06T14:17:06Z

What changes were proposed in this pull request?

Add a CondaRunner abstract class that implements main(String[]) where it sets up a conda environment when certain options are set, then delegates its arguments to an abstract method, along with the optional CondaEnvironment.

Then, changed PythonRunner and RRunner to implement CondaRunner and run their respective subprocess from within the conda env, if one is set up.
NOTE: actually R needs more changes, so for now RRunner will just fail early if conda settings are found.

As for the executor-side logic:

Updated context.py with a addCondaPackage method, which will add the package to the locally set up conda environment (if one is defined), and for distribution on the executors
Updated PythonRDD to set up the desired conda env (that should match the one on the driver, which includes any additional packages supplied to addCondaPackage) before shelling out to python inside that env

How was this patch tested?

End-to-end python test that

bootstraps an env with only python installed (python=3.5)
calls addCondaPackage('numpy=1.11.1')
attempts to import numpy on the driver which should succeed
maps on the executor with a function that attempts to use the numpy import, and asserts that the version is the expected one

End-to-end yarn test that

runs in both client and cluster mode
essentially very similar to the above - tests that an added package is available on both driver & executor

* Add -DskipTests to dev docs * Remove extraneous skipTests

justinuang · 2017-03-10T11:20:56Z

Might be worth updating the PR message

…Also install the conda environment on python executors

…d being final for some reason

…N is set.

…SparkConf.get(x)

…r into /tmp to get shorter path

…rcle

justinuang · 2017-03-10T11:22:16Z

Lets remove always_yes, unless we want to force all boxes with conda to have it set as well. Otherwise, our test env might be different from production.

i agree, already removed in next commit

justinuang · 2017-03-10T11:22:54Z

any reason why we update conda? It makes it dynamic and we are sure what is running anymore

Yeah, we probably shouldn't.. but it's gonna be harder to avoid that now that, in the next commit, I've changed the whole pyenv setup to install a miniconda3 for me, and use that to create two envs (for python 2.7 and 3.4 with numpy installed on both), rather than pip-installing numpy which takes like 2:30min each.
I noticed when doing it that way it always calls update anyway, so even if I pyenv install miniconda3-4.1.11 it will still automatically upgrade it to 4.3.14 post install, and I couldn't figure out how to disable that.

justinuang · 2017-03-12T15:20:08Z

+
+  def create(baseDir: String, bootstrapDeps: Seq[String]): CondaEnvironment = {
+    require(bootstrapDeps.nonEmpty, "Expected at least one bootstrap dependency.")
+    // Attempt to create environment


this comment doesn't seem to describe the next line

yep, will move it

yup, will move it down

justinuang · 2017-03-12T15:27:50Z

+
+    val command = Process(
+      List(condaBinaryPath, "create", "-n", name, "-y", "--override-channels", "-vv")
+        ++: condaChannelUrls.flatMap(Iterator("--channel", _))


maybe let's add a "--" so someone can't have a bootstrap dep that is "--verbose" or something?

justinuang · 2017-03-12T15:28:28Z

+    Files.createSymbolicLink(linkedBaseDir, Paths get baseDir)
+
+    // Expose the symlinked path, through /tmp
+    val condaEnvDir = (linkedBaseDir resolve name).toString


in general, any reason why we don't do linkedBasedDir.resolve(name).toString instead. What is more idiomatic in scala?

I just prefer this style, and it passes scalacheck

justinuang · 2017-03-12T15:51:18Z

    var serverSocket: ServerSocket = null
    try {
-      serverSocket = new ServerSocket(0, 1, InetAddress.getByAddress(Array(127, 0, 0, 1)))
+      serverSocket = new ServerSocket(0, 1, InetAddress.getByAddress(Array[Byte](127, 0, 0, 1)))


Are you sure this is correct. http://stackoverflow.com/a/24010893

this is not changing any behaviour - you'd still have to cast'em if you supply >127 values. But it prevents the line from being red in intellij.

justinuang · 2017-03-12T15:51:49Z

    if (useDaemon) {
      synchronized {
-        if (idleWorkers.size > 0) {
+        if (idleWorkers.nonEmpty) {


Lets try not to do unnecessary refactors if it doesn't have big benefits, it will make it harder to merge upstream changes.

I don't think we plan to merge this upstream, at least not in its current form

More worried about upstream changes, which we need to merge into our fork.

justinuang · 2017-03-12T16:01:51Z

+    builder.redirectOutput(Redirect.INHERIT)
+    builder.redirectError(Redirect.INHERIT)
    sys.env.get("PYTHONHASHSEED").foreach(env.put("PYTHONHASHSEED", _))
-    builder.redirectErrorStream(true) // Ugly but needed for stdout and stderr to synchronize


We don't need this anymore?

I changed the behaviour so that stderr & stdout are inherited, rather than both go to parent's stdout, which makes more sense. I could probably change this back

justinuang · 2017-03-12T16:02:20Z

    // Time to wait for SparkR backend to initialize in seconds
    val backendTimeout = sys.env.getOrElse("SPARKR_BACKEND_TIMEOUT", "120").toInt
-    val rCommand = {
+    val rCommand = maybeConda.map(_.condaEnvDir + "/bin/r").getOrElse {


Unless we want to test and support this, maybe we should just throw an exception if maybeConda is not None.

justinuang · 2017-03-12T16:09:34Z

+            self._conda_env_manager = self._jvm.CondaEnvironmentManager.fromConf(self._conf._jconf)
+
+        self._conda_packages = list()
+        internal_config = getattr(getattr(self._jvm.org.apache.spark.internal.config, 'package$'),


this seems wrong. Is there not another way to get the spark conf, perhaps off the jvm SparkContext?

This is the typed way to do it.
We already have the SparkConf (both jvm and the python SparkConf that wraps it), but the problem is I want to use the SparkConf.get(ConfigEntry) accessor, and this is the only way to get the ConfigEntry is this way.
Alternatively I could get it as a string and split on , like they do just below, I just don't like duplicated logic like that.

…r than using pip

…, will do that later

ash211

My first time looking at this thoroughly -- pretty cool!

Have you guys opened an issue upstream for this? Josh Rosen might be interested in this activity

ash211 · 2017-03-15T21:32:28Z

    - "build_classes"
+    - "build/zinc-0.3.11"
+    - "build/apache-maven-3.3.9"
+    - "build/scala-2.11.8"


any way to make these globs so we don't get version mismatches in future upgrades?

no :( i might just cache the whole build directory

ash211 · 2017-03-15T21:35:35Z

+    // must link in /tmp to reduce path length in case baseDir is very long...
+    // If baseDir path is too long, this breaks conda's 220 character limit for binary replacement.
+    // Don't even try to use java.io.tmpdir - yarn sets this to a very long path
+    val linkedBaseDir = Utils.createTempDir("/tmp", "conda").toPath.resolve("real")


what happens if /tmp is full? We've seen it get full in many places. Is it reasonable to make this configurable via Spark property?

We could. The important thing is this must be really short.

However note that the risk is not high at all since all I'm doing is creating a directory with a symlink to the real stuff. I'm not actually creating any files directly under that /tmp directory directly.

ash211 · 2017-03-15T21:41:31Z

+    val exitCode = command.run(collectErrOutToBuffer).exitValue()
+    if (exitCode != 0) {
+      throw new SparkException(s"Attempt to $description exited with code: "
+      + f"$exitCode%nCommand was: $command%nOutput was:%n${buffer.toString}")


ash211 · 2017-03-15T21:45:55Z

+
+  def activatedEnvironment(startEnv: Map[String, String] = Map.empty): Map[String, String] = {
+    require(!startEnv.contains("PATH"),
+      s"It's not allowed to define PATH in the startEnv, but found: ${startEnv("PATH")}")


reword to Defining PATH in a CondaEnvironment's startEnv is prohibited; found PATH=${startEnv("PATH")}

ash211 · 2017-03-15T21:48:25Z

+
+  def fromConf(sparkConf: SparkConf): CondaEnvironmentManager = {
+    val condaBinaryPath = sparkConf.get(CONDA_BINARY_PATH).getOrElse(
+      sys.error(s"Expected $CONDA_BINARY_PATH to be set"))


do Config items provide the toString you expect? I'd expect this to evaluate to spark.conda.binaryPath

It evaluates to

s"ConfigEntry(key=$key, defaultValue=$defaultValueString, doc=$doc, public=$isPublic)"

would you rather I just print the key?

I suspect just the key is printed in other places so yes would prefer printing just the key. Feel free to check me on that assumption though if you'd like

Don't feel strongly but sure

ash211 · 2017-03-15T22:00:03Z

+
    sys.env.get("PYTHONHASHSEED").foreach(env.put("PYTHONHASHSEED", _))
-    builder.redirectErrorStream(true) // Ugly but needed for stdout and stderr to synchronize
+    builder.redirectErrorStream(true)


needless diff? just looks like merge conflict potential

ash211 · 2017-03-15T22:03:09Z

-      if (sys.props.getOrElse("spark.submit.deployMode", "client") == "client") {
-        cmd = sys.props.getOrElse("spark.r.driver.command", cmd)
+    val rCommand = maybeConda
+      .map(_ => sys.error("RRunner doesn't support running from a Conda environment yet."))


didn't know conda supported R -- pretty cool!
https://www.continuum.io/content/preliminary-support-r-conda

Do you plan on making this work anytime soon?

If there is a need, yes. The total changes should be pretty similar with these ones for python

ash211 · 2017-03-15T22:04:11Z

 */
-object RRunner {
-  def main(args: Array[String]): Unit = {
+object RRunner extends CondaRunner {


given this doesn't actually support Conda yet, part of me is inclined to not touch anything in RRunner at all to reduce merge conflict potential. Maybe we only touch this class when we're actually adding conda support for R? Rather than just the message now that SparkR doesn't support running in Conda?

makes sense

ash211 · 2017-03-15T22:06:15Z

+    ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.spark.ml.classification.OneVsRestModel.this"),
+
+    // CondaRunner is meant to own the main() method then delegate to another method
+    ProblemFilters.exclude[FinalMethodProblem]("org.apache.spark.deploy.CondaRunner.main")


adding to the bottom is prone to merge conflicts when upstream also adds to the bottom -- add to the middle of the list instead?

makes sense

ash211 · 2017-03-15T22:07:37Z

    java_import(gateway.jvm, "org.apache.spark.SparkConf")
    java_import(gateway.jvm, "org.apache.spark.api.java.*")
    java_import(gateway.jvm, "org.apache.spark.api.python.*")
+    java_import(gateway.jvm, "org.apache.spark.api.conda.*")


move up a couple lines to maintain alphabetization

justinuang · 2017-03-15T22:24:05Z

+   *
+   * This hack is necessary otherwise conda tries to use the homedir for pkgs cache.
+   */
+  private[this] def generateCondarc(baseRoot: Path): Path = {


Do we still need this if we are faking home?

yes - the motivation is, we need to ensure our fresh cache dir comes first in pkgs_dirs. Conda will try to write to the first cache (by default anaconda_dir/pkgs) if it's writable, and we don't want it writing to the Anaconda dir

justinuang · 2017-03-15T22:28:41Z

Other than whether we still need the CONDARC and removing unnecessary diffs to minimize conflicts in the future, looks good!

justinuang · 2017-03-15T22:29:24Z

    .stringConf
    .createOptional

+  private[spark] val CONDA_BOOTSTRAP_PACKAGES = ConfigBuilder("spark.conda.bootstrapPackages")


These will almost definitely conflict with upstream changes right? Perhaps we should revert to strings from SparkConf and then splitting on commas.

Why would this conflict?

@justinuang ?

justinuang

+1 after removing unnecessary diffs. no need for a re-review from me

…ven/scala dirs

…g comes from)

…f merge conflict

dansanduleac

alright, addressed changes

dansanduleac · 2017-03-16T11:25:27Z

+  private[this] val localdirs =
+    SparkEnv.get.blockManager.diskBlockManager.localDirs.map(f => f.getPath).mkString(",")
+
+  private[this] val firstFunc = funcs.head.funcs.head


dansanduleac · 2017-03-16T11:26:28Z

+        val localDirs = env.blockManager.diskBlockManager.localDirs
+        val hash = Utils.nonNegativeHash(condaPackages)
+        val dirId = hash % localDirs.length
+        Utils.createTempDir(localDirs(dirId).getAbsolutePath, "conda").getAbsolutePath


I don't understand. Do you want this file to maintain a separate list of dirs that it should clean up itself?

dansanduleac · 2017-03-16T11:26:36Z

+
+    if (CondaEnvironmentManager.isConfigured(sparkConf)) {
+      val condaBootstrapDeps = sparkConf.get(CONDA_BOOTSTRAP_PACKAGES)
+      val condaBaseDir = Utils.createTempDir(Utils.getLocalDir(sparkConf), "conda").getAbsolutePath


VM shutdown

dansanduleac · 2017-03-16T11:36:50Z

+    val pythonExec = maybeConda.map { conda =>
+      presetPythonExec.foreach { exec =>
+        sys.error(
+          s"It's forbidden to set the PYSPARK python path when using conda, but found: $exec")


sure - I've promoted presetPythonExec to Option[Provenance] which includes where the setting was retrieved from

dansanduleac · 2017-03-16T11:37:52Z

+      }
+      conda.condaEnvDir + "/bin/python"
+    }.orElse(presetPythonExec)
+     .getOrElse("python")


That breaks the logic - I want to check there is no preset python exec being set, if I fall back to setting it to python how will I know if the user has set it?

dansanduleac · 2017-03-16T11:41:45Z

+
    sys.env.get("PYTHONHASHSEED").foreach(env.put("PYTHONHASHSEED", _))
-    builder.redirectErrorStream(true) // Ugly but needed for stdout and stderr to synchronize
+    builder.redirectErrorStream(true)


dansanduleac · 2017-03-16T11:43:12Z

 */
-object RRunner {
-  def main(args: Array[String]): Unit = {
+object RRunner extends CondaRunner {


makes sense

dansanduleac · 2017-03-16T11:43:33Z

-      if (sys.props.getOrElse("spark.submit.deployMode", "client") == "client") {
-        cmd = sys.props.getOrElse("spark.r.driver.command", cmd)
+    val rCommand = maybeConda
+      .map(_ => sys.error("RRunner doesn't support running from a Conda environment yet."))


If there is a need, yes. The total changes should be pretty similar with these ones for python

dansanduleac · 2017-03-16T11:51:15Z



-def exec_sbt(sbt_args=()):
+def exec_sbt(sbt_args=(), exit_on_failure=True):


NOTE: all these changes in this file can be reverted once #129 merges, because I've moved the copyTestReportsToCircle logic inside a bespoke test task for circle

dansanduleac · 2017-03-16T11:52:25Z

+    ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.spark.ml.classification.OneVsRestModel.this"),
+
+    // CondaRunner is meant to own the main() method then delegate to another method
+    ProblemFilters.exclude[FinalMethodProblem]("org.apache.spark.deploy.CondaRunner.main")


makes sense

dansanduleac · 2017-03-16T15:54:59Z

So tests can't run anymore for now because anaconda is throttling us and keep getting timeouts... we are looking for a work-around

justinuang · 2017-03-17T09:50:28Z

@robert3005 @ash211 can one of you please merge this and cut a release =)

ash211 · 2017-03-17T16:24:07Z

@justinuang does middle of next week work for you for a release? We're hoping to have the next release include a batch of k8s stuff that's almost finished with review: #133

ash211 · 2017-03-17T16:34:13Z

@dansanduleac can you guys please file a ticket upstream and send in this PR? We're trying really hard to only merge stuff into this repo that has a path to getting merged upstream

justinuang · 2017-03-17T16:36:44Z

yea, middle of week works. Just found out we can depend on dev builds =)

* Add -DskipTests to dev docs * Remove extraneous skipTests

dansanduleac force-pushed the ds/conda-runner branch 3 times, most recently from 09bcb29 to 530ec4a Compare March 7, 2017 18:04

robert3005 pushed a commit that referenced this pull request Mar 7, 2017

Add -DskipTests to dev docs (#115)

ce5e015

* Add -DskipTests to dev docs * Remove extraneous skipTests

dansanduleac added 8 commits March 10, 2017 11:57

Conda options

97a9c6d

Add addCondaPackage method to python SparkContext

225eeef

Implement CondaRunner, and rewrite python and R runner to extend it. …

63d5d87

…Also install the conda environment on python executors

Fix PythonRunner unnecessary RedirectThread, redirect both as inherit

64de33b

Exclude CondaRunner.main from mima, it's complaining about main metho…

bf2aec7

…d being final for some reason

Add conda test

cc23dfb

Set up miniconda on circle

528d12c

PythonRunner should just say no to setting PYSPARK_*_PYTHON

c0440dc

dansanduleac force-pushed the ds/conda-runner branch from a2b6204 to f6460e1 Compare March 10, 2017 12:13

dansanduleac added 3 commits March 10, 2017 17:02

Also on the executor, crash if there is a conda env and PYSPARK_PYTHO…

78b7fea

…N is set.

Cache the stuff that build/mvn always downloads

a4f2c72

Fix weirdness re: how scala SparkConf.get(x) gets called from python …

b395ed1

…SparkConf.get(x)

dansanduleac force-pushed the ds/conda-runner branch from 0dc941c to b395ed1 Compare March 10, 2017 17:03

dansanduleac added 4 commits March 11, 2017 02:01

Added yarn conda test

3094034

Hack CondaEnvironmentManager to use a condarc, and symlink real envdi…

1b79cde

…r into /tmp to get shorter path

Longer timeout for yarn conda tests

5c7ebe5

Fix run-tests.py prematurely exiting before copying the reports to ci…

36cef20

…rcle

justinuang suggested changes Mar 12, 2017

View reviewed changes

dansanduleac force-pushed the ds/conda-runner branch 2 times, most recently from 23ef54a to a167c76 Compare March 12, 2017 19:03

dansanduleac added 4 commits March 13, 2017 12:20

Download miniconda through pyenv, and create envs for the tests rathe…

9633678

…r than using pip

Undo RedirectThread change

20d1227

Disable RRunner conda for now, it needs more changes and tests anyway…

7b12736

…, will do that later

Use CONDA_BIN for tests rather than hard-coding where it is

aabe5fc

dansanduleac force-pushed the ds/conda-runner branch from a167c76 to aabe5fc Compare March 13, 2017 12:27

ash211 reviewed Mar 15, 2017

View reviewed changes

justinuang reviewed Mar 15, 2017

View reviewed changes

justinuang approved these changes Mar 15, 2017

View reviewed changes

dansanduleac added 7 commits March 16, 2017 11:24

Undo sensible changes for fewer merge conflicts

84c8da4

Cache the whole (root) build directory rather than individual zinc/ma…

8faea04

…ven/scala dirs

Improve logging with provenance (where a forbidden python exec settin…

075e65a

…g comes from)

get rid of needless changes

f9ee56b

Revert changes to RRunner altogether

f03a35c

move CondaRunner mima exclude to the middle so there is less chance o…

ea05895

…f merge conflict

get rid of some more unnecessary refactor

9b9b579

dansanduleac commented Mar 16, 2017

View reviewed changes

dansanduleac added 2 commits March 16, 2017 12:28

Attempt to cache the miniconda install again

930186f

scalastyle

7570365

dansanduleac added 2 commits March 16, 2017 16:20

Don't show channel urls, might divulge token info

4de6f96

Merge remote-tracking branch 'origin/master' into ds/conda-runner

7832976

Address @aash's comments

2430478

ash211 changed the title ~~Conda Runner & full Python conda support~~ [SPARK-20001] Conda Runner & full Python conda support Mar 17, 2017

ash211 approved these changes Mar 17, 2017

View reviewed changes

ash211 merged commit ab03652 into master Mar 17, 2017

ash211 deleted the ds/conda-runner branch March 17, 2017 20:22

robert3005 mentioned this pull request Mar 18, 2017

Test flake: SparkSubmitSuite and HiveSparkSubmitSuite #134

Closed

dansanduleac mentioned this pull request Mar 22, 2017

Get rid of default channels in conda #148

Merged

mccheah pushed a commit that referenced this pull request Apr 27, 2017

Add -DskipTests to dev docs (#115)

043cdd9

* Add -DskipTests to dev docs * Remove extraneous skipTests



		def exec_sbt(sbt_args=()):
		def exec_sbt(sbt_args=(), exit_on_failure=True):

Conversation

dansanduleac commented Mar 6, 2017 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

What changes were proposed in this pull request?

How was this patch tested?

End-to-end python test that

End-to-end yarn test that

Uh oh!

justinuang commented Mar 10, 2017

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

dansanduleac Mar 13, 2017 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

ash211 left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

dansanduleac commented Mar 6, 2017 •

edited

Loading

dansanduleac Mar 13, 2017 •

edited

Loading