Merge pull request apache-spark-on-k8s#284 from palantir/aash/resync-apache

[NOSQUASH] Resync from Apache

Author: ash211

R/pkg/NAMESPACE

@@ -232,6 +232,7 @@ exportMethods("%<=>%",
               "date_sub",
               "datediff",
               "dayofmonth",
+              "dayofweek",
               "dayofyear",
               "decode",
               "dense_rank",

R/pkg/R/functions.R

@@ -696,7 +696,7 @@ setMethod("hash",
 #'
 #' \dontrun{
 #' head(select(df, df$time, year(df$time), quarter(df$time), month(df$time),
-#'            dayofmonth(df$time), dayofyear(df$time), weekofyear(df$time)))
+#'             dayofmonth(df$time), dayofweek(df$time), dayofyear(df$time), weekofyear(df$time)))
 #' head(agg(groupBy(df, year(df$time)), count(df$y), avg(df$y)))
 #' head(agg(groupBy(df, month(df$time)), avg(df$y)))}
 #' @note dayofmonth since 1.5.0
@@ -707,6 +707,21 @@ setMethod("dayofmonth",
             column(jc)
           })
 
+#' @details
+#' \code{dayofweek}: Extracts the day of the week as an integer from a
+#' given date/timestamp/string.
+#'
+#' @rdname column_datetime_functions
+#' @aliases dayofweek dayofweek,Column-method
+#' @export
+#' @note dayofweek since 2.3.0
+setMethod("dayofweek",
+          signature(x = "Column"),
+          function(x) {
+            jc <- callJStatic("org.apache.spark.sql.functions", "dayofweek", x@jc)
+            column(jc)
+          })
+
 #' @details
 #' \code{dayofyear}: Extracts the day of the year as an integer from a
 #' given date/timestamp/string.

R/pkg/R/generics.R

@@ -1048,6 +1048,11 @@ setGeneric("date_sub", function(y, x) { standardGeneric("date_sub") })
 #' @name NULL
 setGeneric("dayofmonth", function(x) { standardGeneric("dayofmonth") })
 
+#' @rdname column_datetime_functions
+#' @export
+#' @name NULL
+setGeneric("dayofweek", function(x) { standardGeneric("dayofweek") })
+
 #' @rdname column_datetime_functions
 #' @export
 #' @name NULL

R/pkg/R/install.R

@@ -152,6 +152,11 @@ install.spark <- function(hadoopVersion = "2.7", mirrorUrl = NULL,
                      })
   if (!tarExists || overwrite || !success) {
     unlink(packageLocalPath)
+    if (success) {
+      # if tar file was not there before (or it was, but we are told to overwrite it),
+      # and untar is successful - set a flag that we have downloaded (and untar) Spark package.
+      assign(".sparkDownloaded", TRUE, envir = .sparkREnv)
+    }
   }
   if (!success) stop("Extract archive failed.")
   message("DONE.")
@@ -266,6 +271,7 @@ hadoopVersionName <- function(hadoopVersion) {
 
 # The implementation refers to appdirs package: https://pypi.python.org/pypi/appdirs and
 # adapt to Spark context
+# see also sparkCacheRelPathLength()
 sparkCachePath <- function() {
   if (is_windows()) {
     winAppPath <- Sys.getenv("LOCALAPPDATA", unset = NA)
@@ -282,7 +288,7 @@ sparkCachePath <- function() {
     }
   } else if (.Platform$OS.type == "unix") {
     if (Sys.info()["sysname"] == "Darwin") {
-      path <- file.path(Sys.getenv("HOME"), "Library/Caches", "spark")
+      path <- file.path(Sys.getenv("HOME"), "Library", "Caches", "spark")
     } else {
       path <- file.path(
         Sys.getenv("XDG_CACHE_HOME", file.path(Sys.getenv("HOME"), ".cache")), "spark")
@@ -293,6 +299,16 @@ sparkCachePath <- function() {
   normalizePath(path, mustWork = FALSE)
 }
 
+# Length of the Spark cache specific relative path segments for each platform
+# eg. "Apache\Spark\Cache" is 3 in Windows, or "spark" is 1 in unix
+# Must match sparkCachePath() exactly.
+sparkCacheRelPathLength <- function() {
+  if (is_windows()) {
+    3
+  } else {
+    1
+  }
+}
 
 installInstruction <- function(mode) {
   if (mode == "remote") {
@@ -310,3 +326,22 @@ installInstruction <- function(mode) {
     stop(paste0("No instruction found for ", mode, " mode."))
   }
 }
+
+uninstallDownloadedSpark <- function() {
+  # clean up if Spark was downloaded
+  sparkDownloaded <- getOne(".sparkDownloaded",
+                            envir = .sparkREnv,
+                            inherits = TRUE,
+                            ifnotfound = FALSE)
+  sparkDownloadedDir <- Sys.getenv("SPARK_HOME")
+  if (sparkDownloaded && nchar(sparkDownloadedDir) > 0) {
+    unlink(sparkDownloadedDir, recursive = TRUE, force = TRUE)
+
+    dirs <- traverseParentDirs(sparkCachePath(), sparkCacheRelPathLength())
+    lapply(dirs, function(d) {
+      if (length(list.files(d, all.files = TRUE, include.dirs = TRUE, no.. = TRUE)) == 0) {
+        unlink(d, recursive = TRUE, force = TRUE)
+      }
+    })
+  }
+}

R/pkg/R/utils.R

@@ -910,3 +910,16 @@ hadoop_home_set <- function() {
 windows_with_hadoop <- function() {
   !is_windows() || hadoop_home_set()
 }
+
+# get0 not supported before R 3.2.0
+getOne <- function(x, envir, inherits = TRUE, ifnotfound = NULL) {
+  mget(x[1L], envir = envir, inherits = inherits, ifnotfound = list(ifnotfound))[[1L]]
+}
+
+# Returns a vector of parent directories, traversing up count times, starting with a full path
+# eg. traverseParentDirs("/Users/user/Library/Caches/spark/spark2.2", 1) should return
+# this "/Users/user/Library/Caches/spark/spark2.2"
+# and  "/Users/user/Library/Caches/spark"
+traverseParentDirs <- function(x, count) {
+  if (dirname(x) == x || count <= 0) x else c(x, Recall(dirname(x), count - 1))
+}

R/pkg/tests/fulltests/test_mllib_classification.R

@@ -66,7 +66,7 @@ test_that("spark.svmLinear", {
   feature <- c(1.1419053, 0.9194079, -0.9498666, -1.1069903, 0.2809776)
   data <- as.data.frame(cbind(label, feature))
   df <- createDataFrame(data)
-  model <- spark.svmLinear(df, label ~ feature, regParam = 0.1)
+  model <- spark.svmLinear(df, label ~ feature, regParam = 0.1, maxIter = 5)
   prediction <- collect(select(predict(model, df), "prediction"))
   expect_equal(sort(prediction$prediction), c("0.0", "0.0", "0.0", "1.0", "1.0"))
 
@@ -77,10 +77,11 @@ test_that("spark.svmLinear", {
   trainidxs <- base::sample(nrow(data), nrow(data) * 0.7)
   traindf <- as.DataFrame(data[trainidxs, ])
   testdf <- as.DataFrame(rbind(data[-trainidxs, ], c(0, "the other")))
-  model <- spark.svmLinear(traindf, clicked ~ ., regParam = 0.1)
+  model <- spark.svmLinear(traindf, clicked ~ ., regParam = 0.1, maxIter = 5)
   predictions <- predict(model, testdf)
   expect_error(collect(predictions))
-  model <- spark.svmLinear(traindf, clicked ~ ., regParam = 0.1, handleInvalid = "skip")
+  model <- spark.svmLinear(traindf, clicked ~ ., regParam = 0.1,
+                           handleInvalid = "skip", maxIter = 5)
   predictions <- predict(model, testdf)
   expect_equal(class(collect(predictions)$clicked[1]), "list")
 

R/pkg/tests/fulltests/test_sparkSQL.R

@@ -733,7 +733,7 @@ test_that("test cache, uncache and clearCache", {
   expect_true(dropTempView("table1"))
 
   expect_error(uncacheTable("foo"),
-      "Error in uncacheTable : no such table - Table or view 'foo' not found in database 'default'")
+      "Error in uncacheTable : analysis error - Table or view not found: foo")
 })
 
 test_that("insertInto() on a registered table", {
@@ -1699,6 +1699,7 @@ test_that("date functions on a DataFrame", {
             list(a = 2L, b = as.Date("2013-12-14")),
             list(a = 3L, b = as.Date("2014-12-15")))
   df <- createDataFrame(l)
+  expect_equal(collect(select(df, dayofweek(df$b)))[, 1], c(5, 7, 2))
   expect_equal(collect(select(df, dayofmonth(df$b)))[, 1], c(13, 14, 15))
   expect_equal(collect(select(df, dayofyear(df$b)))[, 1], c(348, 348, 349))
   expect_equal(collect(select(df, weekofyear(df$b)))[, 1], c(50, 50, 51))

R/pkg/tests/fulltests/test_utils.R

@@ -228,4 +228,29 @@ test_that("basenameSansExtFromUrl", {
   expect_equal(basenameSansExtFromUrl(z), "spark-2.1.0--hive")
 })
 
+test_that("getOne", {
+  dummy <- getOne(".dummyValue", envir = new.env(), ifnotfound = FALSE)
+  expect_equal(dummy, FALSE)
+})
+
+test_that("traverseParentDirs", {
+  if (is_windows()) {
+    # original path is included as-is, otherwise dirname() replaces \\ with / on windows
+    dirs <- traverseParentDirs("c:\\Users\\user\\AppData\\Local\\Apache\\Spark\\Cache\\spark2.2", 3)
+    expect <- c("c:\\Users\\user\\AppData\\Local\\Apache\\Spark\\Cache\\spark2.2",
+                "c:/Users/user/AppData/Local/Apache/Spark/Cache",
+                "c:/Users/user/AppData/Local/Apache/Spark",
+                "c:/Users/user/AppData/Local/Apache")
+    expect_equal(dirs, expect)
+  } else {
+    dirs <- traverseParentDirs("/Users/user/Library/Caches/spark/spark2.2", 1)
+    expect <- c("/Users/user/Library/Caches/spark/spark2.2", "/Users/user/Library/Caches/spark")
+    expect_equal(dirs, expect)
+
+    dirs <- traverseParentDirs("/home/u/.cache/spark/spark2.2", 1)
+    expect <- c("/home/u/.cache/spark/spark2.2", "/home/u/.cache/spark")
+    expect_equal(dirs, expect)
+  }
+})
+
 sparkR.session.stop()

R/pkg/tests/run-all.R

@@ -46,7 +46,7 @@ if (identical(Sys.getenv("NOT_CRAN"), "true")) {
   tmpDir <- tempdir()
   tmpArg <- paste0("-Djava.io.tmpdir=", tmpDir)
   sparkRTestConfig <- list(spark.driver.extraJavaOptions = tmpArg,
-                            spark.executor.extraJavaOptions = tmpArg)
+                           spark.executor.extraJavaOptions = tmpArg)
 }
 
 test_package("SparkR")
@@ -60,3 +60,5 @@ if (identical(Sys.getenv("NOT_CRAN"), "true")) {
                        NULL,
                        "summary")
 }
+
+SparkR:::uninstallDownloadedSpark()

R/pkg/vignettes/sparkr-vignettes.Rmd

@@ -37,7 +37,7 @@ opts_hooks$set(eval = function(options) {
   options
 })
 r_tmp_dir <- tempdir()
-tmp_arg <- paste("-Djava.io.tmpdir=", r_tmp_dir, sep = "")
+tmp_arg <- paste0("-Djava.io.tmpdir=", r_tmp_dir)
 sparkSessionConfig <- list(spark.driver.extraJavaOptions = tmp_arg,
                            spark.executor.extraJavaOptions = tmp_arg)
 old_java_opt <- Sys.getenv("_JAVA_OPTIONS")
@@ -1183,3 +1183,7 @@ env | map
 ```{r, echo=FALSE}
 sparkR.session.stop()
 ```
+
+```{r cleanup, include=FALSE}
+SparkR:::uninstallDownloadedSpark()
+```