diff --git a/R/pkg/R/utils.R b/R/pkg/R/utils.R
index d29af00affb983cf33ffa4fe7601ca40739d70df..ea45e394500e8c8f8c4a824d84ce0a31d30bca58 100644
--- a/R/pkg/R/utils.R
+++ b/R/pkg/R/utils.R
@@ -907,3 +907,19 @@ basenameSansExtFromUrl <- function(url) {
isAtomicLengthOne <- function(x) {
is.atomic(x) && length(x) == 1
}
+
+is_cran <- function() {
+ !identical(Sys.getenv("NOT_CRAN"), "true")
+}
+
+is_windows <- function() {
+ .Platform$OS.type == "windows"
+}
+
+hadoop_home_set <- function() {
+ !identical(Sys.getenv("HADOOP_HOME"), "")
+}
+
+not_cran_or_windows_with_hadoop <- function() {
+ !is_cran() && (!is_windows() || hadoop_home_set())
+}
diff --git a/R/pkg/inst/tests/testthat/test_mllib_classification.R b/R/pkg/inst/tests/testthat/test_mllib_classification.R
index f3eaeb381afc48cf4d14e3526c85d9c1cb1d9824..abf8bb25581fd02aab8dc43ac389bb68333b5af3 100644
--- a/R/pkg/inst/tests/testthat/test_mllib_classification.R
+++ b/R/pkg/inst/tests/testthat/test_mllib_classification.R
@@ -50,15 +50,17 @@ test_that("spark.svmLinear", {
expect_equal(sort(as.list(take(select(prediction, "prediction"), 10))[[1]]), expected)
# Test model save and load
- modelPath <- tempfile(pattern = "spark-svm-linear", fileext = ".tmp")
- write.ml(model, modelPath)
- expect_error(write.ml(model, modelPath))
- write.ml(model, modelPath, overwrite = TRUE)
- model2 <- read.ml(modelPath)
- coefs <- summary(model)$coefficients
- coefs2 <- summary(model2)$coefficients
- expect_equal(coefs, coefs2)
- unlink(modelPath)
+ if (not_cran_or_windows_with_hadoop()) {
+ modelPath <- tempfile(pattern = "spark-svm-linear", fileext = ".tmp")
+ write.ml(model, modelPath)
+ expect_error(write.ml(model, modelPath))
+ write.ml(model, modelPath, overwrite = TRUE)
+ model2 <- read.ml(modelPath)
+ coefs <- summary(model)$coefficients
+ coefs2 <- summary(model2)$coefficients
+ expect_equal(coefs, coefs2)
+ unlink(modelPath)
+ }
# Test prediction with numeric label
label <- c(0.0, 0.0, 0.0, 1.0, 1.0)
@@ -128,15 +130,17 @@ test_that("spark.logit", {
expect_true(all(abs(setosaCoefs - setosaCoefs) < 0.1))
# Test model save and load
- modelPath <- tempfile(pattern = "spark-logit", fileext = ".tmp")
- write.ml(model, modelPath)
- expect_error(write.ml(model, modelPath))
- write.ml(model, modelPath, overwrite = TRUE)
- model2 <- read.ml(modelPath)
- coefs <- summary(model)$coefficients
- coefs2 <- summary(model2)$coefficients
- expect_equal(coefs, coefs2)
- unlink(modelPath)
+ if (not_cran_or_windows_with_hadoop()) {
+ modelPath <- tempfile(pattern = "spark-logit", fileext = ".tmp")
+ write.ml(model, modelPath)
+ expect_error(write.ml(model, modelPath))
+ write.ml(model, modelPath, overwrite = TRUE)
+ model2 <- read.ml(modelPath)
+ coefs <- summary(model)$coefficients
+ coefs2 <- summary(model2)$coefficients
+ expect_equal(coefs, coefs2)
+ unlink(modelPath)
+ }
# R code to reproduce the result.
# nolint start
@@ -243,19 +247,21 @@ test_that("spark.mlp", {
expect_equal(head(mlpPredictions$prediction, 6), c("1.0", "0.0", "0.0", "0.0", "0.0", "0.0"))
# Test model save/load
- modelPath <- tempfile(pattern = "spark-mlp", fileext = ".tmp")
- write.ml(model, modelPath)
- expect_error(write.ml(model, modelPath))
- write.ml(model, modelPath, overwrite = TRUE)
- model2 <- read.ml(modelPath)
- summary2 <- summary(model2)
-
- expect_equal(summary2$numOfInputs, 4)
- expect_equal(summary2$numOfOutputs, 3)
- expect_equal(summary2$layers, c(4, 5, 4, 3))
- expect_equal(length(summary2$weights), 64)
-
- unlink(modelPath)
+ if (not_cran_or_windows_with_hadoop()) {
+ modelPath <- tempfile(pattern = "spark-mlp", fileext = ".tmp")
+ write.ml(model, modelPath)
+ expect_error(write.ml(model, modelPath))
+ write.ml(model, modelPath, overwrite = TRUE)
+ model2 <- read.ml(modelPath)
+ summary2 <- summary(model2)
+
+ expect_equal(summary2$numOfInputs, 4)
+ expect_equal(summary2$numOfOutputs, 3)
+ expect_equal(summary2$layers, c(4, 5, 4, 3))
+ expect_equal(length(summary2$weights), 64)
+
+ unlink(modelPath)
+ }
# Test default parameter
model <- spark.mlp(df, label ~ features, layers = c(4, 5, 4, 3))
@@ -354,16 +360,18 @@ test_that("spark.naiveBayes", {
"Yes", "Yes", "No", "No"))
# Test model save/load
- modelPath <- tempfile(pattern = "spark-naiveBayes", fileext = ".tmp")
- write.ml(m, modelPath)
- expect_error(write.ml(m, modelPath))
- write.ml(m, modelPath, overwrite = TRUE)
- m2 <- read.ml(modelPath)
- s2 <- summary(m2)
- expect_equal(s$apriori, s2$apriori)
- expect_equal(s$tables, s2$tables)
-
- unlink(modelPath)
+ if (not_cran_or_windows_with_hadoop()) {
+ modelPath <- tempfile(pattern = "spark-naiveBayes", fileext = ".tmp")
+ write.ml(m, modelPath)
+ expect_error(write.ml(m, modelPath))
+ write.ml(m, modelPath, overwrite = TRUE)
+ m2 <- read.ml(modelPath)
+ s2 <- summary(m2)
+ expect_equal(s$apriori, s2$apriori)
+ expect_equal(s$tables, s2$tables)
+
+ unlink(modelPath)
+ }
# Test e1071::naiveBayes
if (requireNamespace("e1071", quietly = TRUE)) {
diff --git a/R/pkg/inst/tests/testthat/test_mllib_clustering.R b/R/pkg/inst/tests/testthat/test_mllib_clustering.R
index df8e5968b27f450bddb1ab9f1af123791f9e1cea..8f71de1cbc7b5da07439d3f6b792b9a1507a9bed 100644
--- a/R/pkg/inst/tests/testthat/test_mllib_clustering.R
+++ b/R/pkg/inst/tests/testthat/test_mllib_clustering.R
@@ -53,18 +53,20 @@ test_that("spark.bisectingKmeans", {
c(0, 1, 2, 3))
# Test model save/load
- modelPath <- tempfile(pattern = "spark-bisectingkmeans", fileext = ".tmp")
- write.ml(model, modelPath)
- expect_error(write.ml(model, modelPath))
- write.ml(model, modelPath, overwrite = TRUE)
- model2 <- read.ml(modelPath)
- summary2 <- summary(model2)
- expect_equal(sort(unlist(summary.model$size)), sort(unlist(summary2$size)))
- expect_equal(summary.model$coefficients, summary2$coefficients)
- expect_true(!summary.model$is.loaded)
- expect_true(summary2$is.loaded)
-
- unlink(modelPath)
+ if (not_cran_or_windows_with_hadoop()) {
+ modelPath <- tempfile(pattern = "spark-bisectingkmeans", fileext = ".tmp")
+ write.ml(model, modelPath)
+ expect_error(write.ml(model, modelPath))
+ write.ml(model, modelPath, overwrite = TRUE)
+ model2 <- read.ml(modelPath)
+ summary2 <- summary(model2)
+ expect_equal(sort(unlist(summary.model$size)), sort(unlist(summary2$size)))
+ expect_equal(summary.model$coefficients, summary2$coefficients)
+ expect_true(!summary.model$is.loaded)
+ expect_true(summary2$is.loaded)
+
+ unlink(modelPath)
+ }
})
test_that("spark.gaussianMixture", {
@@ -125,18 +127,20 @@ test_that("spark.gaussianMixture", {
expect_equal(p$prediction, c(0, 0, 0, 0, 0, 0, 0, 1, 1, 1, 1, 1, 1, 1, 1))
# Test model save/load
- modelPath <- tempfile(pattern = "spark-gaussianMixture", fileext = ".tmp")
- write.ml(model, modelPath)
- expect_error(write.ml(model, modelPath))
- write.ml(model, modelPath, overwrite = TRUE)
- model2 <- read.ml(modelPath)
- stats2 <- summary(model2)
- expect_equal(stats$lambda, stats2$lambda)
- expect_equal(unlist(stats$mu), unlist(stats2$mu))
- expect_equal(unlist(stats$sigma), unlist(stats2$sigma))
- expect_equal(unlist(stats$loglik), unlist(stats2$loglik))
-
- unlink(modelPath)
+ if (not_cran_or_windows_with_hadoop()) {
+ modelPath <- tempfile(pattern = "spark-gaussianMixture", fileext = ".tmp")
+ write.ml(model, modelPath)
+ expect_error(write.ml(model, modelPath))
+ write.ml(model, modelPath, overwrite = TRUE)
+ model2 <- read.ml(modelPath)
+ stats2 <- summary(model2)
+ expect_equal(stats$lambda, stats2$lambda)
+ expect_equal(unlist(stats$mu), unlist(stats2$mu))
+ expect_equal(unlist(stats$sigma), unlist(stats2$sigma))
+ expect_equal(unlist(stats$loglik), unlist(stats2$loglik))
+
+ unlink(modelPath)
+ }
})
test_that("spark.kmeans", {
@@ -171,18 +175,20 @@ test_that("spark.kmeans", {
expect_true(class(summary.model$coefficients[1, ]) == "numeric")
# Test model save/load
- modelPath <- tempfile(pattern = "spark-kmeans", fileext = ".tmp")
- write.ml(model, modelPath)
- expect_error(write.ml(model, modelPath))
- write.ml(model, modelPath, overwrite = TRUE)
- model2 <- read.ml(modelPath)
- summary2 <- summary(model2)
- expect_equal(sort(unlist(summary.model$size)), sort(unlist(summary2$size)))
- expect_equal(summary.model$coefficients, summary2$coefficients)
- expect_true(!summary.model$is.loaded)
- expect_true(summary2$is.loaded)
-
- unlink(modelPath)
+ if (not_cran_or_windows_with_hadoop()) {
+ modelPath <- tempfile(pattern = "spark-kmeans", fileext = ".tmp")
+ write.ml(model, modelPath)
+ expect_error(write.ml(model, modelPath))
+ write.ml(model, modelPath, overwrite = TRUE)
+ model2 <- read.ml(modelPath)
+ summary2 <- summary(model2)
+ expect_equal(sort(unlist(summary.model$size)), sort(unlist(summary2$size)))
+ expect_equal(summary.model$coefficients, summary2$coefficients)
+ expect_true(!summary.model$is.loaded)
+ expect_true(summary2$is.loaded)
+
+ unlink(modelPath)
+ }
# Test Kmeans on dataset that is sensitive to seed value
col1 <- c(1, 2, 3, 4, 0, 1, 2, 3, 4, 0)
@@ -236,22 +242,24 @@ test_that("spark.lda with libsvm", {
expect_true(logPrior <= 0 & !is.na(logPrior))
# Test model save/load
- modelPath <- tempfile(pattern = "spark-lda", fileext = ".tmp")
- write.ml(model, modelPath)
- expect_error(write.ml(model, modelPath))
- write.ml(model, modelPath, overwrite = TRUE)
- model2 <- read.ml(modelPath)
- stats2 <- summary(model2)
-
- expect_true(stats2$isDistributed)
- expect_equal(logLikelihood, stats2$logLikelihood)
- expect_equal(logPerplexity, stats2$logPerplexity)
- expect_equal(vocabSize, stats2$vocabSize)
- expect_equal(vocabulary, stats2$vocabulary)
- expect_equal(trainingLogLikelihood, stats2$trainingLogLikelihood)
- expect_equal(logPrior, stats2$logPrior)
-
- unlink(modelPath)
+ if (not_cran_or_windows_with_hadoop()) {
+ modelPath <- tempfile(pattern = "spark-lda", fileext = ".tmp")
+ write.ml(model, modelPath)
+ expect_error(write.ml(model, modelPath))
+ write.ml(model, modelPath, overwrite = TRUE)
+ model2 <- read.ml(modelPath)
+ stats2 <- summary(model2)
+
+ expect_true(stats2$isDistributed)
+ expect_equal(logLikelihood, stats2$logLikelihood)
+ expect_equal(logPerplexity, stats2$logPerplexity)
+ expect_equal(vocabSize, stats2$vocabSize)
+ expect_equal(vocabulary, stats2$vocabulary)
+ expect_equal(trainingLogLikelihood, stats2$trainingLogLikelihood)
+ expect_equal(logPrior, stats2$logPrior)
+
+ unlink(modelPath)
+ }
})
test_that("spark.lda with text input", {
diff --git a/R/pkg/inst/tests/testthat/test_mllib_fpm.R b/R/pkg/inst/tests/testthat/test_mllib_fpm.R
index 1fa5375f9da31b5b2e9dee223de8886c07e4cd18..4e10ca1e4f50bdafcae9bbe588b59a71e36b7b55 100644
--- a/R/pkg/inst/tests/testthat/test_mllib_fpm.R
+++ b/R/pkg/inst/tests/testthat/test_mllib_fpm.R
@@ -62,15 +62,17 @@ test_that("spark.fpGrowth", {
expect_equivalent(expected_predictions, collect(predict(model, new_data)))
- modelPath <- tempfile(pattern = "spark-fpm", fileext = ".tmp")
- write.ml(model, modelPath, overwrite = TRUE)
- loaded_model <- read.ml(modelPath)
+ if (not_cran_or_windows_with_hadoop()) {
+ modelPath <- tempfile(pattern = "spark-fpm", fileext = ".tmp")
+ write.ml(model, modelPath, overwrite = TRUE)
+ loaded_model <- read.ml(modelPath)
- expect_equivalent(
- itemsets,
- collect(spark.freqItemsets(loaded_model)))
+ expect_equivalent(
+ itemsets,
+ collect(spark.freqItemsets(loaded_model)))
- unlink(modelPath)
+ unlink(modelPath)
+ }
model_without_numpartitions <- spark.fpGrowth(data, minSupport = 0.3, minConfidence = 0.8)
expect_equal(
diff --git a/R/pkg/inst/tests/testthat/test_mllib_recommendation.R b/R/pkg/inst/tests/testthat/test_mllib_recommendation.R
index e3e2b15c71361eccb23ed6616933136b6171da84..cc8064f88d27a3caf1b4e5a742267ba4b83b0a15 100644
--- a/R/pkg/inst/tests/testthat/test_mllib_recommendation.R
+++ b/R/pkg/inst/tests/testthat/test_mllib_recommendation.R
@@ -37,29 +37,31 @@ test_that("spark.als", {
tolerance = 1e-4)
# Test model save/load
- modelPath <- tempfile(pattern = "spark-als", fileext = ".tmp")
- write.ml(model, modelPath)
- expect_error(write.ml(model, modelPath))
- write.ml(model, modelPath, overwrite = TRUE)
- model2 <- read.ml(modelPath)
- stats2 <- summary(model2)
- expect_equal(stats2$rating, "score")
- userFactors <- collect(stats$userFactors)
- itemFactors <- collect(stats$itemFactors)
- userFactors2 <- collect(stats2$userFactors)
- itemFactors2 <- collect(stats2$itemFactors)
+ if (not_cran_or_windows_with_hadoop()) {
+ modelPath <- tempfile(pattern = "spark-als", fileext = ".tmp")
+ write.ml(model, modelPath)
+ expect_error(write.ml(model, modelPath))
+ write.ml(model, modelPath, overwrite = TRUE)
+ model2 <- read.ml(modelPath)
+ stats2 <- summary(model2)
+ expect_equal(stats2$rating, "score")
+ userFactors <- collect(stats$userFactors)
+ itemFactors <- collect(stats$itemFactors)
+ userFactors2 <- collect(stats2$userFactors)
+ itemFactors2 <- collect(stats2$itemFactors)
- orderUser <- order(userFactors$id)
- orderUser2 <- order(userFactors2$id)
- expect_equal(userFactors$id[orderUser], userFactors2$id[orderUser2])
- expect_equal(userFactors$features[orderUser], userFactors2$features[orderUser2])
+ orderUser <- order(userFactors$id)
+ orderUser2 <- order(userFactors2$id)
+ expect_equal(userFactors$id[orderUser], userFactors2$id[orderUser2])
+ expect_equal(userFactors$features[orderUser], userFactors2$features[orderUser2])
- orderItem <- order(itemFactors$id)
- orderItem2 <- order(itemFactors2$id)
- expect_equal(itemFactors$id[orderItem], itemFactors2$id[orderItem2])
- expect_equal(itemFactors$features[orderItem], itemFactors2$features[orderItem2])
+ orderItem <- order(itemFactors$id)
+ orderItem2 <- order(itemFactors2$id)
+ expect_equal(itemFactors$id[orderItem], itemFactors2$id[orderItem2])
+ expect_equal(itemFactors$features[orderItem], itemFactors2$features[orderItem2])
- unlink(modelPath)
+ unlink(modelPath)
+ }
})
sparkR.session.stop()
diff --git a/R/pkg/inst/tests/testthat/test_mllib_regression.R b/R/pkg/inst/tests/testthat/test_mllib_regression.R
index 44c98be906d815c7fd9c8f14bb625f4f850075f6..b05fdd350ca2824078d0d0a8e6a418fe10818a9c 100644
--- a/R/pkg/inst/tests/testthat/test_mllib_regression.R
+++ b/R/pkg/inst/tests/testthat/test_mllib_regression.R
@@ -401,14 +401,16 @@ test_that("spark.isoreg", {
expect_equal(predict_result$prediction, c(7.0, 7.0, 6.0, 5.5, 5.0, 4.0, 1.0))
# Test model save/load
- modelPath <- tempfile(pattern = "spark-isoreg", fileext = ".tmp")
- write.ml(model, modelPath)
- expect_error(write.ml(model, modelPath))
- write.ml(model, modelPath, overwrite = TRUE)
- model2 <- read.ml(modelPath)
- expect_equal(result, summary(model2))
-
- unlink(modelPath)
+ if (not_cran_or_windows_with_hadoop()) {
+ modelPath <- tempfile(pattern = "spark-isoreg", fileext = ".tmp")
+ write.ml(model, modelPath)
+ expect_error(write.ml(model, modelPath))
+ write.ml(model, modelPath, overwrite = TRUE)
+ model2 <- read.ml(modelPath)
+ expect_equal(result, summary(model2))
+
+ unlink(modelPath)
+ }
})
test_that("spark.survreg", {
@@ -450,17 +452,19 @@ test_that("spark.survreg", {
2.390146, 2.891269, 2.891269), tolerance = 1e-4)
# Test model save/load
- modelPath <- tempfile(pattern = "spark-survreg", fileext = ".tmp")
- write.ml(model, modelPath)
- expect_error(write.ml(model, modelPath))
- write.ml(model, modelPath, overwrite = TRUE)
- model2 <- read.ml(modelPath)
- stats2 <- summary(model2)
- coefs2 <- as.vector(stats2$coefficients[, 1])
- expect_equal(coefs, coefs2)
- expect_equal(rownames(stats$coefficients), rownames(stats2$coefficients))
-
- unlink(modelPath)
+ if (not_cran_or_windows_with_hadoop()) {
+ modelPath <- tempfile(pattern = "spark-survreg", fileext = ".tmp")
+ write.ml(model, modelPath)
+ expect_error(write.ml(model, modelPath))
+ write.ml(model, modelPath, overwrite = TRUE)
+ model2 <- read.ml(modelPath)
+ stats2 <- summary(model2)
+ coefs2 <- as.vector(stats2$coefficients[, 1])
+ expect_equal(coefs, coefs2)
+ expect_equal(rownames(stats$coefficients), rownames(stats2$coefficients))
+
+ unlink(modelPath)
+ }
# Test survival::survreg
if (requireNamespace("survival", quietly = TRUE)) {
diff --git a/R/pkg/inst/tests/testthat/test_mllib_tree.R b/R/pkg/inst/tests/testthat/test_mllib_tree.R
index b283e734cec53f2da21d034f1db91468c510a91c..5fd6a38ecb4fab38c889da4a9d9ba2d0deeb6488 100644
--- a/R/pkg/inst/tests/testthat/test_mllib_tree.R
+++ b/R/pkg/inst/tests/testthat/test_mllib_tree.R
@@ -44,21 +44,23 @@ test_that("spark.gbt", {
expect_equal(stats$numFeatures, 6)
expect_equal(length(stats$treeWeights), 20)
- modelPath <- tempfile(pattern = "spark-gbtRegression", fileext = ".tmp")
- write.ml(model, modelPath)
- expect_error(write.ml(model, modelPath))
- write.ml(model, modelPath, overwrite = TRUE)
- model2 <- read.ml(modelPath)
- stats2 <- summary(model2)
- expect_equal(stats$formula, stats2$formula)
- expect_equal(stats$numFeatures, stats2$numFeatures)
- expect_equal(stats$features, stats2$features)
- expect_equal(stats$featureImportances, stats2$featureImportances)
- expect_equal(stats$maxDepth, stats2$maxDepth)
- expect_equal(stats$numTrees, stats2$numTrees)
- expect_equal(stats$treeWeights, stats2$treeWeights)
-
- unlink(modelPath)
+ if (not_cran_or_windows_with_hadoop()) {
+ modelPath <- tempfile(pattern = "spark-gbtRegression", fileext = ".tmp")
+ write.ml(model, modelPath)
+ expect_error(write.ml(model, modelPath))
+ write.ml(model, modelPath, overwrite = TRUE)
+ model2 <- read.ml(modelPath)
+ stats2 <- summary(model2)
+ expect_equal(stats$formula, stats2$formula)
+ expect_equal(stats$numFeatures, stats2$numFeatures)
+ expect_equal(stats$features, stats2$features)
+ expect_equal(stats$featureImportances, stats2$featureImportances)
+ expect_equal(stats$maxDepth, stats2$maxDepth)
+ expect_equal(stats$numTrees, stats2$numTrees)
+ expect_equal(stats$treeWeights, stats2$treeWeights)
+
+ unlink(modelPath)
+ }
# classification
# label must be binary - GBTClassifier currently only supports binary classification.
@@ -76,17 +78,19 @@ test_that("spark.gbt", {
expect_equal(length(grep("setosa", predictions)), 50)
expect_equal(length(grep("versicolor", predictions)), 50)
- modelPath <- tempfile(pattern = "spark-gbtClassification", fileext = ".tmp")
- write.ml(model, modelPath)
- expect_error(write.ml(model, modelPath))
- write.ml(model, modelPath, overwrite = TRUE)
- model2 <- read.ml(modelPath)
- stats2 <- summary(model2)
- expect_equal(stats$depth, stats2$depth)
- expect_equal(stats$numNodes, stats2$numNodes)
- expect_equal(stats$numClasses, stats2$numClasses)
-
- unlink(modelPath)
+ if (not_cran_or_windows_with_hadoop()) {
+ modelPath <- tempfile(pattern = "spark-gbtClassification", fileext = ".tmp")
+ write.ml(model, modelPath)
+ expect_error(write.ml(model, modelPath))
+ write.ml(model, modelPath, overwrite = TRUE)
+ model2 <- read.ml(modelPath)
+ stats2 <- summary(model2)
+ expect_equal(stats$depth, stats2$depth)
+ expect_equal(stats$numNodes, stats2$numNodes)
+ expect_equal(stats$numClasses, stats2$numClasses)
+
+ unlink(modelPath)
+ }
iris2$NumericSpecies <- ifelse(iris2$Species == "setosa", 0, 1)
df <- suppressWarnings(createDataFrame(iris2))
@@ -136,21 +140,23 @@ test_that("spark.randomForest", {
expect_equal(stats$numTrees, 20)
expect_equal(stats$maxDepth, 5)
- modelPath <- tempfile(pattern = "spark-randomForestRegression", fileext = ".tmp")
- write.ml(model, modelPath)
- expect_error(write.ml(model, modelPath))
- write.ml(model, modelPath, overwrite = TRUE)
- model2 <- read.ml(modelPath)
- stats2 <- summary(model2)
- expect_equal(stats$formula, stats2$formula)
- expect_equal(stats$numFeatures, stats2$numFeatures)
- expect_equal(stats$features, stats2$features)
- expect_equal(stats$featureImportances, stats2$featureImportances)
- expect_equal(stats$numTrees, stats2$numTrees)
- expect_equal(stats$maxDepth, stats2$maxDepth)
- expect_equal(stats$treeWeights, stats2$treeWeights)
-
- unlink(modelPath)
+ if (not_cran_or_windows_with_hadoop()) {
+ modelPath <- tempfile(pattern = "spark-randomForestRegression", fileext = ".tmp")
+ write.ml(model, modelPath)
+ expect_error(write.ml(model, modelPath))
+ write.ml(model, modelPath, overwrite = TRUE)
+ model2 <- read.ml(modelPath)
+ stats2 <- summary(model2)
+ expect_equal(stats$formula, stats2$formula)
+ expect_equal(stats$numFeatures, stats2$numFeatures)
+ expect_equal(stats$features, stats2$features)
+ expect_equal(stats$featureImportances, stats2$featureImportances)
+ expect_equal(stats$numTrees, stats2$numTrees)
+ expect_equal(stats$maxDepth, stats2$maxDepth)
+ expect_equal(stats$treeWeights, stats2$treeWeights)
+
+ unlink(modelPath)
+ }
# classification
data <- suppressWarnings(createDataFrame(iris))
@@ -168,17 +174,19 @@ test_that("spark.randomForest", {
expect_equal(length(grep("setosa", predictions)), 50)
expect_equal(length(grep("versicolor", predictions)), 50)
- modelPath <- tempfile(pattern = "spark-randomForestClassification", fileext = ".tmp")
- write.ml(model, modelPath)
- expect_error(write.ml(model, modelPath))
- write.ml(model, modelPath, overwrite = TRUE)
- model2 <- read.ml(modelPath)
- stats2 <- summary(model2)
- expect_equal(stats$depth, stats2$depth)
- expect_equal(stats$numNodes, stats2$numNodes)
- expect_equal(stats$numClasses, stats2$numClasses)
-
- unlink(modelPath)
+ if (not_cran_or_windows_with_hadoop()) {
+ modelPath <- tempfile(pattern = "spark-randomForestClassification", fileext = ".tmp")
+ write.ml(model, modelPath)
+ expect_error(write.ml(model, modelPath))
+ write.ml(model, modelPath, overwrite = TRUE)
+ model2 <- read.ml(modelPath)
+ stats2 <- summary(model2)
+ expect_equal(stats$depth, stats2$depth)
+ expect_equal(stats$numNodes, stats2$numNodes)
+ expect_equal(stats$numClasses, stats2$numClasses)
+
+ unlink(modelPath)
+ }
# Test numeric response variable
labelToIndex <- function(species) {
diff --git a/R/pkg/inst/tests/testthat/test_sparkSQL.R b/R/pkg/inst/tests/testthat/test_sparkSQL.R
index b633b78d5bb4dbec4ff9c4541f1c845228a49cfa..9fc6e5dabecc32909c7c0543726b03601519f5cd 100644
--- a/R/pkg/inst/tests/testthat/test_sparkSQL.R
+++ b/R/pkg/inst/tests/testthat/test_sparkSQL.R
@@ -61,7 +61,11 @@ unsetHiveContext <- function() {
# Tests for SparkSQL functions in SparkR
filesBefore <- list.files(path = sparkRDir, all.files = TRUE)
-sparkSession <- sparkR.session(master = sparkRTestMaster)
+sparkSession <- if (not_cran_or_windows_with_hadoop()) {
+ sparkR.session(master = sparkRTestMaster)
+ } else {
+ sparkR.session(master = sparkRTestMaster, enableHiveSupport = FALSE)
+ }
sc <- callJStatic("org.apache.spark.sql.api.r.SQLUtils", "getJavaSparkContext", sparkSession)
mockLines <- c("{\"name\":\"Michael\"}",
@@ -326,51 +330,53 @@ test_that("createDataFrame uses files for large objects", {
})
test_that("read/write csv as DataFrame", {
- csvPath <- tempfile(pattern = "sparkr-test", fileext = ".csv")
- mockLinesCsv <- c("year,make,model,comment,blank",
- "\"2012\",\"Tesla\",\"S\",\"No comment\",",
- "1997,Ford,E350,\"Go get one now they are going fast\",",
- "2015,Chevy,Volt",
- "NA,Dummy,Placeholder")
- writeLines(mockLinesCsv, csvPath)
-
- # default "header" is false, inferSchema to handle "year" as "int"
- df <- read.df(csvPath, "csv", header = "true", inferSchema = "true")
- expect_equal(count(df), 4)
- expect_equal(columns(df), c("year", "make", "model", "comment", "blank"))
- expect_equal(sort(unlist(collect(where(df, df$year == 2015)))),
- sort(unlist(list(year = 2015, make = "Chevy", model = "Volt"))))
-
- # since "year" is "int", let's skip the NA values
- withoutna <- na.omit(df, how = "any", cols = "year")
- expect_equal(count(withoutna), 3)
-
- unlink(csvPath)
- csvPath <- tempfile(pattern = "sparkr-test", fileext = ".csv")
- mockLinesCsv <- c("year,make,model,comment,blank",
- "\"2012\",\"Tesla\",\"S\",\"No comment\",",
- "1997,Ford,E350,\"Go get one now they are going fast\",",
- "2015,Chevy,Volt",
- "Empty,Dummy,Placeholder")
- writeLines(mockLinesCsv, csvPath)
-
- df2 <- read.df(csvPath, "csv", header = "true", inferSchema = "true", na.strings = "Empty")
- expect_equal(count(df2), 4)
- withoutna2 <- na.omit(df2, how = "any", cols = "year")
- expect_equal(count(withoutna2), 3)
- expect_equal(count(where(withoutna2, withoutna2$make == "Dummy")), 0)
-
- # writing csv file
- csvPath2 <- tempfile(pattern = "csvtest2", fileext = ".csv")
- write.df(df2, path = csvPath2, "csv", header = "true")
- df3 <- read.df(csvPath2, "csv", header = "true")
- expect_equal(nrow(df3), nrow(df2))
- expect_equal(colnames(df3), colnames(df2))
- csv <- read.csv(file = list.files(csvPath2, pattern = "^part", full.names = T)[[1]])
- expect_equal(colnames(df3), colnames(csv))
-
- unlink(csvPath)
- unlink(csvPath2)
+ if (not_cran_or_windows_with_hadoop()) {
+ csvPath <- tempfile(pattern = "sparkr-test", fileext = ".csv")
+ mockLinesCsv <- c("year,make,model,comment,blank",
+ "\"2012\",\"Tesla\",\"S\",\"No comment\",",
+ "1997,Ford,E350,\"Go get one now they are going fast\",",
+ "2015,Chevy,Volt",
+ "NA,Dummy,Placeholder")
+ writeLines(mockLinesCsv, csvPath)
+
+ # default "header" is false, inferSchema to handle "year" as "int"
+ df <- read.df(csvPath, "csv", header = "true", inferSchema = "true")
+ expect_equal(count(df), 4)
+ expect_equal(columns(df), c("year", "make", "model", "comment", "blank"))
+ expect_equal(sort(unlist(collect(where(df, df$year == 2015)))),
+ sort(unlist(list(year = 2015, make = "Chevy", model = "Volt"))))
+
+ # since "year" is "int", let's skip the NA values
+ withoutna <- na.omit(df, how = "any", cols = "year")
+ expect_equal(count(withoutna), 3)
+
+ unlink(csvPath)
+ csvPath <- tempfile(pattern = "sparkr-test", fileext = ".csv")
+ mockLinesCsv <- c("year,make,model,comment,blank",
+ "\"2012\",\"Tesla\",\"S\",\"No comment\",",
+ "1997,Ford,E350,\"Go get one now they are going fast\",",
+ "2015,Chevy,Volt",
+ "Empty,Dummy,Placeholder")
+ writeLines(mockLinesCsv, csvPath)
+
+ df2 <- read.df(csvPath, "csv", header = "true", inferSchema = "true", na.strings = "Empty")
+ expect_equal(count(df2), 4)
+ withoutna2 <- na.omit(df2, how = "any", cols = "year")
+ expect_equal(count(withoutna2), 3)
+ expect_equal(count(where(withoutna2, withoutna2$make == "Dummy")), 0)
+
+ # writing csv file
+ csvPath2 <- tempfile(pattern = "csvtest2", fileext = ".csv")
+ write.df(df2, path = csvPath2, "csv", header = "true")
+ df3 <- read.df(csvPath2, "csv", header = "true")
+ expect_equal(nrow(df3), nrow(df2))
+ expect_equal(colnames(df3), colnames(df2))
+ csv <- read.csv(file = list.files(csvPath2, pattern = "^part", full.names = T)[[1]])
+ expect_equal(colnames(df3), colnames(csv))
+
+ unlink(csvPath)
+ unlink(csvPath2)
+ }
})
test_that("Support other types for options", {
@@ -601,48 +607,50 @@ test_that("Collect DataFrame with complex types", {
})
test_that("read/write json files", {
- # Test read.df
- df <- read.df(jsonPath, "json")
- expect_is(df, "SparkDataFrame")
- expect_equal(count(df), 3)
-
- # Test read.df with a user defined schema
- schema <- structType(structField("name", type = "string"),
- structField("age", type = "double"))
-
- df1 <- read.df(jsonPath, "json", schema)
- expect_is(df1, "SparkDataFrame")
- expect_equal(dtypes(df1), list(c("name", "string"), c("age", "double")))
-
- # Test loadDF
- df2 <- loadDF(jsonPath, "json", schema)
- expect_is(df2, "SparkDataFrame")
- expect_equal(dtypes(df2), list(c("name", "string"), c("age", "double")))
-
- # Test read.json
- df <- read.json(jsonPath)
- expect_is(df, "SparkDataFrame")
- expect_equal(count(df), 3)
-
- # Test write.df
- jsonPath2 <- tempfile(pattern = "jsonPath2", fileext = ".json")
- write.df(df, jsonPath2, "json", mode = "overwrite")
-
- # Test write.json
- jsonPath3 <- tempfile(pattern = "jsonPath3", fileext = ".json")
- write.json(df, jsonPath3)
-
- # Test read.json()/jsonFile() works with multiple input paths
- jsonDF1 <- read.json(c(jsonPath2, jsonPath3))
- expect_is(jsonDF1, "SparkDataFrame")
- expect_equal(count(jsonDF1), 6)
- # Suppress warnings because jsonFile is deprecated
- jsonDF2 <- suppressWarnings(jsonFile(c(jsonPath2, jsonPath3)))
- expect_is(jsonDF2, "SparkDataFrame")
- expect_equal(count(jsonDF2), 6)
-
- unlink(jsonPath2)
- unlink(jsonPath3)
+ if (not_cran_or_windows_with_hadoop()) {
+ # Test read.df
+ df <- read.df(jsonPath, "json")
+ expect_is(df, "SparkDataFrame")
+ expect_equal(count(df), 3)
+
+ # Test read.df with a user defined schema
+ schema <- structType(structField("name", type = "string"),
+ structField("age", type = "double"))
+
+ df1 <- read.df(jsonPath, "json", schema)
+ expect_is(df1, "SparkDataFrame")
+ expect_equal(dtypes(df1), list(c("name", "string"), c("age", "double")))
+
+ # Test loadDF
+ df2 <- loadDF(jsonPath, "json", schema)
+ expect_is(df2, "SparkDataFrame")
+ expect_equal(dtypes(df2), list(c("name", "string"), c("age", "double")))
+
+ # Test read.json
+ df <- read.json(jsonPath)
+ expect_is(df, "SparkDataFrame")
+ expect_equal(count(df), 3)
+
+ # Test write.df
+ jsonPath2 <- tempfile(pattern = "jsonPath2", fileext = ".json")
+ write.df(df, jsonPath2, "json", mode = "overwrite")
+
+ # Test write.json
+ jsonPath3 <- tempfile(pattern = "jsonPath3", fileext = ".json")
+ write.json(df, jsonPath3)
+
+ # Test read.json()/jsonFile() works with multiple input paths
+ jsonDF1 <- read.json(c(jsonPath2, jsonPath3))
+ expect_is(jsonDF1, "SparkDataFrame")
+ expect_equal(count(jsonDF1), 6)
+ # Suppress warnings because jsonFile is deprecated
+ jsonDF2 <- suppressWarnings(jsonFile(c(jsonPath2, jsonPath3)))
+ expect_is(jsonDF2, "SparkDataFrame")
+ expect_equal(count(jsonDF2), 6)
+
+ unlink(jsonPath2)
+ unlink(jsonPath3)
+ }
})
test_that("read/write json files - compression option", {
@@ -736,33 +744,35 @@ test_that("test cache, uncache and clearCache", {
})
test_that("insertInto() on a registered table", {
- df <- read.df(jsonPath, "json")
- write.df(df, parquetPath, "parquet", "overwrite")
- dfParquet <- read.df(parquetPath, "parquet")
-
- lines <- c("{\"name\":\"Bob\", \"age\":24}",
- "{\"name\":\"James\", \"age\":35}")
- jsonPath2 <- tempfile(pattern = "jsonPath2", fileext = ".tmp")
- parquetPath2 <- tempfile(pattern = "parquetPath2", fileext = ".parquet")
- writeLines(lines, jsonPath2)
- df2 <- read.df(jsonPath2, "json")
- write.df(df2, parquetPath2, "parquet", "overwrite")
- dfParquet2 <- read.df(parquetPath2, "parquet")
-
- createOrReplaceTempView(dfParquet, "table1")
- insertInto(dfParquet2, "table1")
- expect_equal(count(sql("select * from table1")), 5)
- expect_equal(first(sql("select * from table1 order by age"))$name, "Michael")
- expect_true(dropTempView("table1"))
-
- createOrReplaceTempView(dfParquet, "table1")
- insertInto(dfParquet2, "table1", overwrite = TRUE)
- expect_equal(count(sql("select * from table1")), 2)
- expect_equal(first(sql("select * from table1 order by age"))$name, "Bob")
- expect_true(dropTempView("table1"))
-
- unlink(jsonPath2)
- unlink(parquetPath2)
+ if (not_cran_or_windows_with_hadoop()) {
+ df <- read.df(jsonPath, "json")
+ write.df(df, parquetPath, "parquet", "overwrite")
+ dfParquet <- read.df(parquetPath, "parquet")
+
+ lines <- c("{\"name\":\"Bob\", \"age\":24}",
+ "{\"name\":\"James\", \"age\":35}")
+ jsonPath2 <- tempfile(pattern = "jsonPath2", fileext = ".tmp")
+ parquetPath2 <- tempfile(pattern = "parquetPath2", fileext = ".parquet")
+ writeLines(lines, jsonPath2)
+ df2 <- read.df(jsonPath2, "json")
+ write.df(df2, parquetPath2, "parquet", "overwrite")
+ dfParquet2 <- read.df(parquetPath2, "parquet")
+
+ createOrReplaceTempView(dfParquet, "table1")
+ insertInto(dfParquet2, "table1")
+ expect_equal(count(sql("select * from table1")), 5)
+ expect_equal(first(sql("select * from table1 order by age"))$name, "Michael")
+ expect_true(dropTempView("table1"))
+
+ createOrReplaceTempView(dfParquet, "table1")
+ insertInto(dfParquet2, "table1", overwrite = TRUE)
+ expect_equal(count(sql("select * from table1")), 2)
+ expect_equal(first(sql("select * from table1 order by age"))$name, "Bob")
+ expect_true(dropTempView("table1"))
+
+ unlink(jsonPath2)
+ unlink(parquetPath2)
+ }
})
test_that("tableToDF() returns a new DataFrame", {
@@ -954,14 +964,16 @@ test_that("cache(), storageLevel(), persist(), and unpersist() on a DataFrame",
})
test_that("setCheckpointDir(), checkpoint() on a DataFrame", {
- checkpointDir <- file.path(tempdir(), "cproot")
- expect_true(length(list.files(path = checkpointDir, all.files = TRUE)) == 0)
-
- setCheckpointDir(checkpointDir)
- df <- read.json(jsonPath)
- df <- checkpoint(df)
- expect_is(df, "SparkDataFrame")
- expect_false(length(list.files(path = checkpointDir, all.files = TRUE)) == 0)
+ if (not_cran_or_windows_with_hadoop()) {
+ checkpointDir <- file.path(tempdir(), "cproot")
+ expect_true(length(list.files(path = checkpointDir, all.files = TRUE)) == 0)
+
+ setCheckpointDir(checkpointDir)
+ df <- read.json(jsonPath)
+ df <- checkpoint(df)
+ expect_is(df, "SparkDataFrame")
+ expect_false(length(list.files(path = checkpointDir, all.files = TRUE)) == 0)
+ }
})
test_that("schema(), dtypes(), columns(), names() return the correct values/format", {
@@ -1329,45 +1341,47 @@ test_that("column calculation", {
})
test_that("test HiveContext", {
- setHiveContext(sc)
-
- schema <- structType(structField("name", "string"), structField("age", "integer"),
- structField("height", "float"))
- createTable("people", source = "json", schema = schema)
- df <- read.df(jsonPathNa, "json", schema)
- insertInto(df, "people")
- expect_equal(collect(sql("SELECT age from people WHERE name = 'Bob'"))$age, c(16))
- sql("DROP TABLE people")
-
- df <- createTable("json", jsonPath, "json")
- expect_is(df, "SparkDataFrame")
- expect_equal(count(df), 3)
- df2 <- sql("select * from json")
- expect_is(df2, "SparkDataFrame")
- expect_equal(count(df2), 3)
-
- jsonPath2 <- tempfile(pattern = "sparkr-test", fileext = ".tmp")
- saveAsTable(df, "json2", "json", "append", path = jsonPath2)
- df3 <- sql("select * from json2")
- expect_is(df3, "SparkDataFrame")
- expect_equal(count(df3), 3)
- unlink(jsonPath2)
-
- hivetestDataPath <- tempfile(pattern = "sparkr-test", fileext = ".tmp")
- saveAsTable(df, "hivetestbl", path = hivetestDataPath)
- df4 <- sql("select * from hivetestbl")
- expect_is(df4, "SparkDataFrame")
- expect_equal(count(df4), 3)
- unlink(hivetestDataPath)
-
- parquetDataPath <- tempfile(pattern = "sparkr-test", fileext = ".tmp")
- saveAsTable(df, "parquetest", "parquet", mode = "overwrite", path = parquetDataPath)
- df5 <- sql("select * from parquetest")
- expect_is(df5, "SparkDataFrame")
- expect_equal(count(df5), 3)
- unlink(parquetDataPath)
-
- unsetHiveContext()
+ if (not_cran_or_windows_with_hadoop()) {
+ setHiveContext(sc)
+
+ schema <- structType(structField("name", "string"), structField("age", "integer"),
+ structField("height", "float"))
+ createTable("people", source = "json", schema = schema)
+ df <- read.df(jsonPathNa, "json", schema)
+ insertInto(df, "people")
+ expect_equal(collect(sql("SELECT age from people WHERE name = 'Bob'"))$age, c(16))
+ sql("DROP TABLE people")
+
+ df <- createTable("json", jsonPath, "json")
+ expect_is(df, "SparkDataFrame")
+ expect_equal(count(df), 3)
+ df2 <- sql("select * from json")
+ expect_is(df2, "SparkDataFrame")
+ expect_equal(count(df2), 3)
+
+ jsonPath2 <- tempfile(pattern = "sparkr-test", fileext = ".tmp")
+ saveAsTable(df, "json2", "json", "append", path = jsonPath2)
+ df3 <- sql("select * from json2")
+ expect_is(df3, "SparkDataFrame")
+ expect_equal(count(df3), 3)
+ unlink(jsonPath2)
+
+ hivetestDataPath <- tempfile(pattern = "sparkr-test", fileext = ".tmp")
+ saveAsTable(df, "hivetestbl", path = hivetestDataPath)
+ df4 <- sql("select * from hivetestbl")
+ expect_is(df4, "SparkDataFrame")
+ expect_equal(count(df4), 3)
+ unlink(hivetestDataPath)
+
+ parquetDataPath <- tempfile(pattern = "sparkr-test", fileext = ".tmp")
+ saveAsTable(df, "parquetest", "parquet", mode = "overwrite", path = parquetDataPath)
+ df5 <- sql("select * from parquetest")
+ expect_is(df5, "SparkDataFrame")
+ expect_equal(count(df5), 3)
+ unlink(parquetDataPath)
+
+ unsetHiveContext()
+ }
})
test_that("column operators", {
@@ -2420,34 +2434,36 @@ test_that("read/write ORC files - compression option", {
})
test_that("read/write Parquet files", {
- df <- read.df(jsonPath, "json")
- # Test write.df and read.df
- write.df(df, parquetPath, "parquet", mode = "overwrite")
- df2 <- read.df(parquetPath, "parquet")
- expect_is(df2, "SparkDataFrame")
- expect_equal(count(df2), 3)
-
- # Test write.parquet/saveAsParquetFile and read.parquet/parquetFile
- parquetPath2 <- tempfile(pattern = "parquetPath2", fileext = ".parquet")
- write.parquet(df, parquetPath2)
- parquetPath3 <- tempfile(pattern = "parquetPath3", fileext = ".parquet")
- suppressWarnings(saveAsParquetFile(df, parquetPath3))
- parquetDF <- read.parquet(c(parquetPath2, parquetPath3))
- expect_is(parquetDF, "SparkDataFrame")
- expect_equal(count(parquetDF), count(df) * 2)
- parquetDF2 <- suppressWarnings(parquetFile(parquetPath2, parquetPath3))
- expect_is(parquetDF2, "SparkDataFrame")
- expect_equal(count(parquetDF2), count(df) * 2)
-
- # Test if varargs works with variables
- saveMode <- "overwrite"
- mergeSchema <- "true"
- parquetPath4 <- tempfile(pattern = "parquetPath3", fileext = ".parquet")
- write.df(df, parquetPath3, "parquet", mode = saveMode, mergeSchema = mergeSchema)
-
- unlink(parquetPath2)
- unlink(parquetPath3)
- unlink(parquetPath4)
+ if (not_cran_or_windows_with_hadoop()) {
+ df <- read.df(jsonPath, "json")
+ # Test write.df and read.df
+ write.df(df, parquetPath, "parquet", mode = "overwrite")
+ df2 <- read.df(parquetPath, "parquet")
+ expect_is(df2, "SparkDataFrame")
+ expect_equal(count(df2), 3)
+
+ # Test write.parquet/saveAsParquetFile and read.parquet/parquetFile
+ parquetPath2 <- tempfile(pattern = "parquetPath2", fileext = ".parquet")
+ write.parquet(df, parquetPath2)
+ parquetPath3 <- tempfile(pattern = "parquetPath3", fileext = ".parquet")
+ suppressWarnings(saveAsParquetFile(df, parquetPath3))
+ parquetDF <- read.parquet(c(parquetPath2, parquetPath3))
+ expect_is(parquetDF, "SparkDataFrame")
+ expect_equal(count(parquetDF), count(df) * 2)
+ parquetDF2 <- suppressWarnings(parquetFile(parquetPath2, parquetPath3))
+ expect_is(parquetDF2, "SparkDataFrame")
+ expect_equal(count(parquetDF2), count(df) * 2)
+
+ # Test if varargs works with variables
+ saveMode <- "overwrite"
+ mergeSchema <- "true"
+ parquetPath4 <- tempfile(pattern = "parquetPath3", fileext = ".parquet")
+ write.df(df, parquetPath3, "parquet", mode = saveMode, mergeSchema = mergeSchema)
+
+ unlink(parquetPath2)
+ unlink(parquetPath3)
+ unlink(parquetPath4)
+ }
})
test_that("read/write Parquet files - compression option/mode", {