Always offer File type conda instructions to executors using driver's transitive packages (#710)

* use verbosity flag in install

* always pass down file instruction for executors conda env

* check style

* reuse yarn test to double check executor conda env

* fix yarn cluster test python file

* better documentation, tests

* update doc w/ checkstyle

Co-authored-by: Willi Raschkowski <[email protected]>

* remove unnecessary import test

Co-authored-by: Willi Raschkowski <[email protected]>

Author: Arnoldosmium

core/src/main/scala/org/apache/spark/api/conda/CondaEnvironment.scala

@@ -92,6 +92,7 @@ final class CondaEnvironment(
     manager.runCondaProcess(rootPath,
       List("install", "-n", envName, "-y")
         ::: extraArgs.toList
+        ::: manager.verbosityFlags
         ::: "--" :: packages.toList,
       description = s"install dependencies in conda env $condaEnvDir",
       channels = channels.iterator.map(_.url).toList,
@@ -113,17 +114,35 @@ final class CondaEnvironment(
   }
 
   /**
-   * This is for sending the instructions to the executors so they can replicate the same steps.
+   * This is for sending the instructions to the executors so they can replicate the same conda
+   * environment.
+   * <p><ul>
+   *   <li>In {@code File} mode, re-use the same specfile on executors.</li>
+   *   <li>In {@code Solve} mode, list resolved packages into a specfile
+   *    and use that on executors.</li>
+   * </ul>
    */
   def buildSetupInstructions: CondaSetupInstructions = {
-    CondaSetupInstructions(
-      bootstrapMode,
-      packages.toList,
-      bootstrapPackageUrls,
-      bootstrapPackageUrlsUserInfo,
-      channels.toList,
-      extraArgs,
-      envVars)
+    bootstrapMode match {
+      case CondaBootstrapMode.Solve =>
+        CondaSetupInstructions(
+          CondaBootstrapMode.File,
+          packages.toList,
+          getTransitivePackageUrls(),
+          bootstrapPackageUrlsUserInfo,
+          channels.toList,
+          extraArgs,
+          envVars)
+      case CondaBootstrapMode.File =>
+        CondaSetupInstructions(
+          bootstrapMode,
+          packages.toList,
+          bootstrapPackageUrls,
+          bootstrapPackageUrlsUserInfo,
+          channels.toList,
+          extraArgs,
+          envVars)
+    }
   }
 }
 

core/src/main/scala/org/apache/spark/api/conda/CondaEnvironmentManager.scala

@@ -71,6 +71,10 @@ final class CondaEnvironmentManager(condaBinaryPath: String,
     defaultInfo("pkgs_dirs").extract[List[String]]
   }
 
+  private[conda] lazy val verbosityFlags: List[String] = {
+    0.until(verbosity).map(_ => "-v").toList
+  }
+
   def listPackagesExplicit(envDir: String): List[String] = {
     logInfo("Retrieving a conda environment's list of installed packages")
     val command = Process(List(condaBinaryPath, "list", "-p", envDir, "--explicit"), None)
@@ -90,7 +94,13 @@ final class CondaEnvironmentManager(condaBinaryPath: String,
       condaEnvVars: Map[String, String] = Map.empty): CondaEnvironment = {
     condaMode match {
       case CondaBootstrapMode.Solve =>
-        create(baseDir, condaPackages, condaChannelUrls, condaExtraArgs, condaEnvVars)
+        create(
+          baseDir,
+          condaPackages,
+          condaPackageUrlsUserInfo,
+          condaChannelUrls,
+          condaExtraArgs,
+          condaEnvVars)
       case CondaBootstrapMode.File =>
         createWithFile(
           baseDir, condaPackageUrls, condaPackageUrlsUserInfo, condaExtraArgs, condaEnvVars)
@@ -100,6 +110,7 @@ final class CondaEnvironmentManager(condaBinaryPath: String,
   def create(
       baseDir: String,
       condaPackages: Seq[String],
+      condaPackageUrlsUserInfo: Option[String],
       condaChannelUrls: Seq[String],
       condaExtraArgs: Seq[String] = Nil,
       condaEnvVars: Map[String, String] = Map.empty): CondaEnvironment = {
@@ -114,8 +125,6 @@ final class CondaEnvironmentManager(condaBinaryPath: String,
     logInfo(s"Creating symlink $linkedBaseDir -> $baseDir")
     Files.createSymbolicLink(linkedBaseDir, Paths.get(baseDir))
 
-    val verbosityFlags = 0.until(verbosity).map(_ => "-v").toList
-
     // Attempt to create environment
     runCondaProcess(
       linkedBaseDir,
@@ -135,7 +144,7 @@ final class CondaEnvironmentManager(condaBinaryPath: String,
       CondaBootstrapMode.Solve,
       condaPackages,
       Nil,
-      None,
+      condaPackageUrlsUserInfo,
       condaChannelUrls,
       condaExtraArgs)
   }
@@ -159,8 +168,6 @@ final class CondaEnvironmentManager(condaBinaryPath: String,
     logInfo(s"Creating symlink $linkedBaseDir -> $baseDir")
     Files.createSymbolicLink(linkedBaseDir, Paths.get(baseDir))
 
-    val verbosityFlags = 0.until(verbosity).map(_ => "-v").toList
-
     // Authenticate URLs if we have a UserInfo argument
     val finalCondaPackageUrls = if (condaPackageUrlsUserInfo.isDefined) {
       condaPackageUrls.map { packageUrl =>

resource-managers/yarn/src/test/scala/org/apache/spark/deploy/yarn/YarnClusterSuite.scala

@@ -160,6 +160,35 @@ class YarnClusterSuite extends BaseYarnClusterSuite {
     |    sc.stop()
   """.stripMargin
 
+  private val TEST_CONDA_DRIVER_INIT_PYFILE = """
+    |import os
+    |import sys
+    |
+    |from pyspark import SparkConf , SparkContext
+    |if __name__ == "__main__":
+    |    if len(sys.argv) != 2:
+    |        print >> sys.stderr, "Usage: test.py [result file]"
+    |        exit(-1)
+    |    sc = SparkContext(conf=SparkConf())
+    |
+    |    status = open(sys.argv[1],'w')
+    |
+    |    def numpy_multiply(x):
+    |        # Ensure executors have the same packages of the driver.
+    |        import numpy
+    |        return numpy.multiply(x, 2)
+    |
+    |    rdd = sc.parallelize(range(10)).map(numpy_multiply)
+    |    rdd_sum = rdd.sum()
+    |    if rdd_sum == 90:       # sum(0:9) * 2 = 90
+    |        result = "success"
+    |    else:
+    |        result = "failure"
+    |    status.write(result)
+    |    status.close()
+    |    sc.stop()
+  """.stripMargin
+
   private val TEST_PYMODULE = """
     |def func():
     |    return 42
@@ -282,6 +311,20 @@ class YarnClusterSuite extends BaseYarnClusterSuite {
     testCondaPySparkAllModes(clientMode = false)
   }
 
+  test("Python application within Conda in yarn-cluster mode has necessary pkgs") {
+    // run conda driver
+    val extraConfForCreate: Map[String, String] = Map(
+      "spark.conda.binaryPath" -> sys.env("CONDA_BIN"),
+      "spark.conda.channelUrls" -> "https://repo.continuum.io/pkgs/main",
+      "spark.conda.bootstrapPackages" -> "python=3.6,numpy=1.14.0"
+    )
+
+    testCondaPySpark(
+      clientMode = false,
+      TEST_CONDA_DRIVER_INIT_PYFILE,
+      extraConf = extraConfForCreate)
+  }
+
   test("run Python application in yarn-cluster mode using " +
     "spark.yarn.appMasterEnv to override local envvar") {
     testPySpark(