Add conda support for R (apache#261)

Author: dansanduleac

R/pkg/DESCRIPTION

@@ -57,5 +57,5 @@ Collate:
     'types.R'
     'utils.R'
     'window.R'
-RoxygenNote: 5.0.1
+RoxygenNote: 6.0.1
 VignetteBuilder: knitr

R/pkg/R/RDD.R

@@ -168,6 +168,7 @@ setMethod("getJRDD", signature(rdd = "PipelinedRDD"),
                                    serializedFuncArr,
                                    rdd@env$prev_serializedMode,
                                    packageNamesArr,
+                                   spark.buildCondaInstructions(),
                                    broadcastArr,
                                    callJMethod(prev_jrdd, "classTag"))
             } else {
@@ -177,6 +178,7 @@ setMethod("getJRDD", signature(rdd = "PipelinedRDD"),
                                    rdd@env$prev_serializedMode,
                                    serializedMode,
                                    packageNamesArr,
+                                   spark.buildCondaInstructions(),
                                    broadcastArr,
                                    callJMethod(prev_jrdd, "classTag"))
             }

R/pkg/R/context.R

@@ -319,6 +319,13 @@ spark.addFile <- function(path, recursive = FALSE) {
   invisible(callJMethod(sc, "addFile", suppressWarnings(normalizePath(path)), recursive))
 }
 
+#' Construct condaBuildInstructions used to re-create the driver's conda
+#' environment on executors.
+spark.buildCondaInstructions <- function() {
+  sc <- callJMethod(getSparkContext(), "sc")
+  callJMethod(sc, "buildCondaInstructions")
+}
+
 #' Get the root directory that contains files added through spark.addFile.
 #'
 #' @rdname spark.getSparkFilesRootDirectory

R/pkg/tests/fulltests/test_includePackage.R

@@ -18,7 +18,10 @@
 context("include R packages")
 
 # JavaSparkContext handle
-sparkSession <- sparkR.session(master = sparkRTestMaster, enableHiveSupport = FALSE)
+config <- list(spark.conda.channelUrls = "https://repo.continuum.io/pkgs/r/",
+  spark.conda.bootstrapPackages = "r,plyr")
+sparkSession <- sparkR.session(master = sparkRTestMaster,
+  enableHiveSupport = FALSE, sparkConfig = config)
 sc <- callJStatic("org.apache.spark.sql.api.r.SQLUtils", "getJavaSparkContext", sparkSession)
 
 # Partitioned data
@@ -27,18 +30,17 @@ rdd <- parallelize(sc, nums, 2L)
 
 test_that("include inside function", {
   # Only run the test if plyr is installed.
-  if ("plyr" %in% rownames(installed.packages())) {
-    suppressPackageStartupMessages(library(plyr))
-    generateData <- function(x) {
-      suppressPackageStartupMessages(library(plyr))
-      attach(airquality)
-      result <- transform(Ozone, logOzone = log(Ozone))
-      result
-    }
 
-    data <- lapplyPartition(rdd, generateData)
-    actual <- collectRDD(data)
+  suppressPackageStartupMessages(library(plyr))
+  generateData <- function(x) {
+    suppressPackageStartupMessages(library(plyr))
+    attach(airquality)
+    result <- transform(Ozone, logOzone = log(Ozone))
+    result
   }
+
+  data <- lapplyPartition(rdd, generateData)
+  actual <- collectRDD(data)
 })
 
 test_that("use include package", {

R/run-tests.sh

@@ -23,7 +23,7 @@ FAILED=0
 LOGFILE=$FWDIR/unit-tests.out
 rm -f $LOGFILE
 
-SPARK_TESTING=1 NOT_CRAN=true $FWDIR/../bin/spark-submit --driver-java-options "-Dlog4j.configuration=file:$FWDIR/log4j.properties" --conf spark.hadoop.fs.defaultFS="file:///" $FWDIR/pkg/tests/run-all.R 2>&1 | tee -a $LOGFILE
+SPARK_TESTING=1 NOT_CRAN=true $FWDIR/../bin/spark-submit --driver-java-options "-Dlog4j.configuration=file:$FWDIR/log4j.properties" --num-executors 1 --conf spark.hadoop.fs.defaultFS="file:///" --conf spark.conda.binaryPath=$CONDA_BIN --conf spark.conda.bootstrapPackages="r,r-essentials,r-plyr,r-testthat" --conf spark.conda.channelUrls="https://repo.continuum.io/pkgs/r,https://repo.continuum.io/pkgs/main,https://repo.continuum.io/pkgs/free,https://repo.continuum.io/pkgs/pro" $FWDIR/pkg/tests/run-all.R 2>&1 | tee -a $LOGFILE
 FAILED=$((PIPESTATUS[0]||$FAILED))
 
 NUM_TEST_WARNING="$(grep -c -e 'Warnings ----------------' $LOGFILE)"

circle.yml

@@ -2,7 +2,13 @@ machine:
   java:
     version: oraclejdk8
   post:
-    - sudo apt-get --assume-yes install r-base r-base-dev
+    - sudo sh -c 'echo "deb http://cran.rstudio.com/bin/linux/ubuntu trusty/" >> /etc/apt/sources.list'
+    - gpg --keyserver keyserver.ubuntu.com --recv-key E084DAB9
+    - gpg -a --export E084DAB9 | sudo apt-key add -
+    - sudo apt-get update
+    - sudo apt-get --assume-yes install r-base r-base-dev qpdf
+    - sudo chmod 777 /usr/local/lib/R/site-library 
+    - /usr/lib/R/bin/R -e "install.packages(c('knitr', 'rmarkdown', 'testthat', 'e1071', 'survival', 'devtools', 'roxygen2', 'lintr'), repos='http://cran.us.r-project.org', lib='/usr/local/lib/R/site-library')"
     - |
       if [[ ! -d ${CONDA_ROOT} ]]; then
           echo "Installing Miniconda...";

core/src/main/scala/org/apache/spark/SparkEnv.scala

@@ -17,7 +17,7 @@
 
 package org.apache.spark
 
-import java.io.File
+import java.io.{DataOutputStream, File, IOException}
 import java.net.Socket
 import java.util.Locale
 
@@ -76,6 +76,7 @@ class SparkEnv (
   case class PythonWorkerKey(pythonExec: Option[String], envVars: Map[String, String],
                              condaInstructions: Option[CondaSetupInstructions])
   private val pythonWorkers = mutable.HashMap[PythonWorkerKey, PythonWorkerFactory]()
+  private var rDaemonChannel: DataOutputStream = _
 
   // A general, soft-reference map for metadata needed during HadoopRDD split computation
   // (e.g., HadoopFileRDD uses this to cache JobConfs and InputFormats).
@@ -88,6 +89,7 @@ class SparkEnv (
     if (!isStopped) {
       isStopped = true
       pythonWorkers.values.foreach(_.stop())
+      destroyRDaemonChannel()
       mapOutputTracker.stop()
       shuffleManager.stop()
       broadcastManager.stop()
@@ -114,6 +116,34 @@ class SparkEnv (
     }
   }
 
+  private[spark] def setRDaemonChannel(daemonChannel: DataOutputStream) {
+    rDaemonChannel = daemonChannel
+  }
+
+  private[spark] def rDaemonExists(): Boolean = {
+    rDaemonChannel != null
+  }
+
+  private[spark] def destroyRDaemonChannel(): Unit = {
+    if (rDaemonChannel != null) {
+      rDaemonChannel.close()
+      rDaemonChannel = null
+    }
+  }
+
+  private[spark] def createRWorkerFromDaemon(port: Int) {
+    try {
+      rDaemonChannel.writeInt(port)
+      rDaemonChannel.flush()
+    } catch {
+      case e: IOException =>
+        // daemon process died
+        destroyRDaemonChannel()
+        // fail the current task, retry by scheduler
+        throw e
+    }
+  }
+
   private[spark]
   def createPythonWorker(pythonExec: Option[String], envVars: Map[String, String],
                          condaInstructions: Option[CondaSetupInstructions]): java.net.Socket = {

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

@@ -36,7 +36,9 @@ import org.json4s.jackson.Json4sScalaModule
 import org.json4s.jackson.JsonMethods
 
 import org.apache.spark.SparkConf
+import org.apache.spark.SparkEnv
 import org.apache.spark.SparkException
+import org.apache.spark.api.conda.CondaEnvironment.CondaSetupInstructions
 import org.apache.spark.internal.Logging
 import org.apache.spark.internal.config.CONDA_BINARY_PATH
 import org.apache.spark.internal.config.CONDA_GLOBAL_PACKAGE_DIRS
@@ -216,4 +218,23 @@ object CondaEnvironmentManager extends Logging {
     val packageDirs = sparkConf.get(CONDA_GLOBAL_PACKAGE_DIRS)
     new CondaEnvironmentManager(condaBinaryPath, verbosity, packageDirs)
   }
+
+  /**
+   * Helper method to create a conda environment from [[CondaEnvironment.CondaSetupInstructions]].
+   * This is intended to be called on the executor with serialized instructions.
+   */
+  def createCondaEnvironment(instructions: CondaSetupInstructions): CondaEnvironment = {
+    val condaPackages = instructions.packages
+    val env = SparkEnv.get
+    val condaEnvManager = CondaEnvironmentManager.fromConf(env.conf)
+    val envDir = {
+      // Which local dir to create it in?
+      val localDirs = env.blockManager.diskBlockManager.localDirs
+      val hash = Utils.nonNegativeHash(condaPackages)
+      val dirId = hash % localDirs.length
+      Utils.createTempDir(localDirs(dirId).getAbsolutePath, "conda").getAbsolutePath
+    }
+    condaEnvManager.create(envDir, condaPackages, instructions.channels)
+  }
+
 }

core/src/main/scala/org/apache/spark/api/python/PythonWorkerFactory.scala

@@ -29,7 +29,6 @@ import org.apache.spark._
 import org.apache.spark.api.conda.CondaEnvironment.CondaSetupInstructions
 import org.apache.spark.api.conda.CondaEnvironmentManager
 import org.apache.spark.internal.Logging
-import org.apache.spark.internal.config.CONDA_BOOTSTRAP_PACKAGES
 import org.apache.spark.util.{RedirectThread, Utils}
 
 private[spark] class PythonWorkerFactory(requestedPythonExec: Option[String],
@@ -62,20 +61,7 @@ private[spark] class PythonWorkerFactory(requestedPythonExec: Option[String],
 
   private[this] val condaEnv = {
     // Set up conda environment if there are any conda packages requested
-    condaInstructions.map { instructions =>
-      val condaPackages = instructions.packages
-
-      val env = SparkEnv.get
-      val condaEnvManager = CondaEnvironmentManager.fromConf(env.conf)
-      val envDir = {
-        // Which local dir to create it in?
-        val localDirs = env.blockManager.diskBlockManager.localDirs
-        val hash = Utils.nonNegativeHash(condaPackages)
-        val dirId = hash % localDirs.length
-        Utils.createTempDir(localDirs(dirId).getAbsolutePath, "conda").getAbsolutePath
-      }
-      condaEnvManager.create(envDir, condaPackages, instructions.channels)
-    }
+    condaInstructions.map(CondaEnvironmentManager.createCondaEnvironment)
   }
 
   private[this] val envVars: Map[String, String] = {

core/src/main/scala/org/apache/spark/api/r/RRDD.scala

@@ -24,6 +24,7 @@ import scala.collection.JavaConverters._
 import scala.reflect.ClassTag
 
 import org.apache.spark._
+import org.apache.spark.api.conda.CondaEnvironment.CondaSetupInstructions
 import org.apache.spark.api.java.{JavaPairRDD, JavaRDD, JavaSparkContext}
 import org.apache.spark.api.python.PythonRDD
 import org.apache.spark.broadcast.Broadcast
@@ -39,11 +40,17 @@ private abstract class BaseRRDD[T: ClassTag, U: ClassTag](
     packageNames: Array[Byte],
     broadcastVars: Array[Broadcast[Object]])
   extends RDD[U](parent) with Logging {
+
+  /**
+   * Get the conda instructions eagerly - when the RDD is created.
+   */
+  val condaInstructions: Option[CondaSetupInstructions] = context.buildCondaInstructions()
+
   override def getPartitions: Array[Partition] = parent.partitions
 
   override def compute(partition: Partition, context: TaskContext): Iterator[U] = {
     val runner = new RRunner[U](
-      func, deserializer, serializer, packageNames, broadcastVars, numPartitions)
+      func, deserializer, serializer, packageNames, broadcastVars, condaInstructions, numPartitions)
 
     // The parent may be also an RRDD, so we should launch it first.
     val parentIterator = firstParent[T].iterator(partition, context)
@@ -79,6 +86,7 @@ private class RRDD[T: ClassTag](
     deserializer: String,
     serializer: String,
     packageNames: Array[Byte],
+    condaSetupInstructions: Option[CondaSetupInstructions],
     broadcastVars: Array[Object])
   extends BaseRRDD[T, Array[Byte]](
     parent, -1, func, deserializer, serializer, packageNames,
@@ -94,6 +102,7 @@ private class StringRRDD[T: ClassTag](
     func: Array[Byte],
     deserializer: String,
     packageNames: Array[Byte],
+    condaSetupInstructions: Option[CondaSetupInstructions],
     broadcastVars: Array[Object])
   extends BaseRRDD[T, String](
     parent, -1, func, deserializer, SerializationFormats.STRING, packageNames,