spark git commit: [SPARK-20098][PYSPARK] dataType's typeName fix
Repository: spark Updated Branches: refs/heads/master f76790557 -> 520d92a19 [SPARK-20098][PYSPARK] dataType's typeName fix ## What changes were proposed in this pull request? `typeName` classmethod has been fixed by using type -> typeName map. ## How was this patch tested? local build Author: Peter Szalai Closes #17435 from szalai1/datatype-gettype-fix. Project: http://git-wip-us.apache.org/repos/asf/spark/repo Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/520d92a1 Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/520d92a1 Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/520d92a1 Branch: refs/heads/master Commit: 520d92a191c3148498087d751aeeddd683055622 Parents: f767905 Author: Peter Szalai Authored: Sun Sep 10 17:47:45 2017 +0900 Committer: hyukjinkwon Committed: Sun Sep 10 17:47:45 2017 +0900 -- python/pyspark/sql/tests.py | 4 python/pyspark/sql/types.py | 5 + 2 files changed, 9 insertions(+) -- http://git-wip-us.apache.org/repos/asf/spark/blob/520d92a1/python/pyspark/sql/tests.py -- diff --git a/python/pyspark/sql/tests.py b/python/pyspark/sql/tests.py index 4d65abc..6e7ddf9 100644 --- a/python/pyspark/sql/tests.py +++ b/python/pyspark/sql/tests.py @@ -209,6 +209,10 @@ class DataTypeTests(unittest.TestCase): row = Row() self.assertEqual(len(row), 0) +def test_struct_field_type_name(self): +struct_field = StructField("a", IntegerType()) +self.assertRaises(TypeError, struct_field.typeName) + class SQLTests(ReusedPySparkTestCase): http://git-wip-us.apache.org/repos/asf/spark/blob/520d92a1/python/pyspark/sql/types.py -- diff --git a/python/pyspark/sql/types.py b/python/pyspark/sql/types.py index 51bf7be..920cf00 100644 --- a/python/pyspark/sql/types.py +++ b/python/pyspark/sql/types.py @@ -440,6 +440,11 @@ class StructField(DataType): def fromInternal(self, obj): return self.dataType.fromInternal(obj) +def typeName(self): +raise TypeError( +"StructField does not have typeName. " +"Use typeName on its type explicitly instead.") + class StructType(DataType): """Struct type, consisting of a list of :class:`StructField`. - To unsubscribe, e-mail: commits-unsubscr...@spark.apache.org For additional commands, e-mail: commits-h...@spark.apache.org
spark git commit: [SPARK-20098][PYSPARK] dataType's typeName fix
Repository: spark Updated Branches: refs/heads/branch-2.2 182478e03 -> b1b5a7fdc [SPARK-20098][PYSPARK] dataType's typeName fix ## What changes were proposed in this pull request? `typeName` classmethod has been fixed by using type -> typeName map. ## How was this patch tested? local build Author: Peter Szalai Closes #17435 from szalai1/datatype-gettype-fix. (cherry picked from commit 520d92a191c3148498087d751aeeddd683055622) Signed-off-by: hyukjinkwon Project: http://git-wip-us.apache.org/repos/asf/spark/repo Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/b1b5a7fd Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/b1b5a7fd Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/b1b5a7fd Branch: refs/heads/branch-2.2 Commit: b1b5a7fdc0f8fabfb235f0b31bde0f1bfb71591a Parents: 182478e Author: Peter Szalai Authored: Sun Sep 10 17:47:45 2017 +0900 Committer: hyukjinkwon Committed: Sun Sep 10 17:48:00 2017 +0900 -- python/pyspark/sql/tests.py | 4 python/pyspark/sql/types.py | 5 + 2 files changed, 9 insertions(+) -- http://git-wip-us.apache.org/repos/asf/spark/blob/b1b5a7fd/python/pyspark/sql/tests.py -- diff --git a/python/pyspark/sql/tests.py b/python/pyspark/sql/tests.py index a100dc0..39655a5 100644 --- a/python/pyspark/sql/tests.py +++ b/python/pyspark/sql/tests.py @@ -188,6 +188,10 @@ class DataTypeTests(unittest.TestCase): row = Row() self.assertEqual(len(row), 0) +def test_struct_field_type_name(self): +struct_field = StructField("a", IntegerType()) +self.assertRaises(TypeError, struct_field.typeName) + class SQLTests(ReusedPySparkTestCase): http://git-wip-us.apache.org/repos/asf/spark/blob/b1b5a7fd/python/pyspark/sql/types.py -- diff --git a/python/pyspark/sql/types.py b/python/pyspark/sql/types.py index 26b54a7..d9206dd 100644 --- a/python/pyspark/sql/types.py +++ b/python/pyspark/sql/types.py @@ -438,6 +438,11 @@ class StructField(DataType): def fromInternal(self, obj): return self.dataType.fromInternal(obj) +def typeName(self): +raise TypeError( +"StructField does not have typeName. " +"Use typeName on its type explicitly instead.") + class StructType(DataType): """Struct type, consisting of a list of :class:`StructField`. - To unsubscribe, e-mail: commits-unsubscr...@spark.apache.org For additional commands, e-mail: commits-h...@spark.apache.org
[3/9] spark git commit: [SPARKR][BACKPORT-2.1] backporting package and test changes
http://git-wip-us.apache.org/repos/asf/spark/blob/ae4e8ae4/R/pkg/tests/fulltests/test_shuffle.R -- diff --git a/R/pkg/tests/fulltests/test_shuffle.R b/R/pkg/tests/fulltests/test_shuffle.R new file mode 100644 index 000..d38efab --- /dev/null +++ b/R/pkg/tests/fulltests/test_shuffle.R @@ -0,0 +1,224 @@ +# +# Licensed to the Apache Software Foundation (ASF) under one or more +# contributor license agreements. See the NOTICE file distributed with +# this work for additional information regarding copyright ownership. +# The ASF licenses this file to You under the Apache License, Version 2.0 +# (the "License"); you may not use this file except in compliance with +# the License. You may obtain a copy of the License at +# +#http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# + +context("partitionBy, groupByKey, reduceByKey etc.") + +# JavaSparkContext handle +sparkSession <- sparkR.session(enableHiveSupport = FALSE) +sc <- callJStatic("org.apache.spark.sql.api.r.SQLUtils", "getJavaSparkContext", sparkSession) + +# Data +intPairs <- list(list(1L, -1), list(2L, 100), list(2L, 1), list(1L, 200)) +intRdd <- parallelize(sc, intPairs, 2L) + +doublePairs <- list(list(1.5, -1), list(2.5, 100), list(2.5, 1), list(1.5, 200)) +doubleRdd <- parallelize(sc, doublePairs, 2L) + +numPairs <- list(list(1L, 100), list(2L, 200), list(4L, -1), list(3L, 1), + list(3L, 0)) +numPairsRdd <- parallelize(sc, numPairs, length(numPairs)) + +strList <- list("Dexter Morgan: Blood. Sometimes it sets my teeth on edge and ", +"Dexter Morgan: Harry and Dorris Morgan did a wonderful job ") +strListRDD <- parallelize(sc, strList, 4) + +test_that("groupByKey for integers", { + grouped <- groupByKey(intRdd, 2L) + + actual <- collectRDD(grouped) + + expected <- list(list(2L, list(100, 1)), list(1L, list(-1, 200))) + expect_equal(sortKeyValueList(actual), sortKeyValueList(expected)) +}) + +test_that("groupByKey for doubles", { + grouped <- groupByKey(doubleRdd, 2L) + + actual <- collectRDD(grouped) + + expected <- list(list(1.5, list(-1, 200)), list(2.5, list(100, 1))) + expect_equal(sortKeyValueList(actual), sortKeyValueList(expected)) +}) + +test_that("reduceByKey for ints", { + reduced <- reduceByKey(intRdd, "+", 2L) + + actual <- collectRDD(reduced) + + expected <- list(list(2L, 101), list(1L, 199)) + expect_equal(sortKeyValueList(actual), sortKeyValueList(expected)) +}) + +test_that("reduceByKey for doubles", { + reduced <- reduceByKey(doubleRdd, "+", 2L) + actual <- collectRDD(reduced) + + expected <- list(list(1.5, 199), list(2.5, 101)) + expect_equal(sortKeyValueList(actual), sortKeyValueList(expected)) +}) + +test_that("combineByKey for ints", { + reduced <- combineByKey(intRdd, function(x) { x }, "+", "+", 2L) + + actual <- collectRDD(reduced) + + expected <- list(list(2L, 101), list(1L, 199)) + expect_equal(sortKeyValueList(actual), sortKeyValueList(expected)) +}) + +test_that("combineByKey for doubles", { + reduced <- combineByKey(doubleRdd, function(x) { x }, "+", "+", 2L) + actual <- collectRDD(reduced) + + expected <- list(list(1.5, 199), list(2.5, 101)) + expect_equal(sortKeyValueList(actual), sortKeyValueList(expected)) +}) + +test_that("combineByKey for characters", { + stringKeyRDD <- parallelize(sc, + list(list("max", 1L), list("min", 2L), + list("other", 3L), list("max", 4L)), 2L) + reduced <- combineByKey(stringKeyRDD, + function(x) { x }, "+", "+", 2L) + actual <- collectRDD(reduced) + + expected <- list(list("max", 5L), list("min", 2L), list("other", 3L)) + expect_equal(sortKeyValueList(actual), sortKeyValueList(expected)) +}) + +test_that("aggregateByKey", { + # test aggregateByKey for int keys + rdd <- parallelize(sc, list(list(1, 1), list(1, 2), list(2, 3), list(2, 4))) + + zeroValue <- list(0, 0) + seqOp <- function(x, y) { list(x[[1]] + y, x[[2]] + 1) } + combOp <- function(x, y) { list(x[[1]] + y[[1]], x[[2]] + y[[2]]) } + aggregatedRDD <- aggregateByKey(rdd, zeroValue, seqOp, combOp, 2L) + + actual <- collectRDD(aggregatedRDD) + + expected <- list(list(1, list(3, 2)), list(2, list(7, 2))) + expect_equal(sortKeyValueList(actual), sortKeyValueList(expected)) + + # test aggregateByKey for string keys + rdd <- parallelize(sc, list(list("a", 1), list("a", 2), list("b", 3), list("b", 4))) + + zeroValue <- list(0, 0) + seqOp <- function(x, y) { list(x[[1]] + y, x[[2]] + 1) } + combOp <- function(x, y) { list(x[[1]] + y[[1]], x[[2]] + y[[2]]) } + aggregatedRDD <- aggregateByK
[1/9] spark git commit: [SPARKR][BACKPORT-2.1] backporting package and test changes
Repository: spark Updated Branches: refs/heads/branch-2.1 6a8a726f3 -> ae4e8ae41 http://git-wip-us.apache.org/repos/asf/spark/blob/ae4e8ae4/R/pkg/tests/fulltests/test_take.R -- diff --git a/R/pkg/tests/fulltests/test_take.R b/R/pkg/tests/fulltests/test_take.R new file mode 100644 index 000..aaa5328 --- /dev/null +++ b/R/pkg/tests/fulltests/test_take.R @@ -0,0 +1,69 @@ +# +# Licensed to the Apache Software Foundation (ASF) under one or more +# contributor license agreements. See the NOTICE file distributed with +# this work for additional information regarding copyright ownership. +# The ASF licenses this file to You under the Apache License, Version 2.0 +# (the "License"); you may not use this file except in compliance with +# the License. You may obtain a copy of the License at +# +#http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# + +context("tests RDD function take()") + +# Mock data +numVector <- c(-10:97) +numList <- list(sqrt(1), sqrt(2), sqrt(3), 4 ** 10) +strVector <- c("Dexter Morgan: I suppose I should be upset, even feel", + "violated, but I'm not. No, in fact, I think this is a friendly", + "message, like \"Hey, wanna play?\" and yes, I want to play. ", + "I really, really do.") +strList <- list("Dexter Morgan: Blood. Sometimes it sets my teeth on edge, ", +"other times it helps me control the chaos.", +"Dexter Morgan: Harry and Dorris Morgan did a wonderful job ", +"raising me. But they're both dead now. I didn't kill them. Honest.") + +# JavaSparkContext handle +sparkSession <- sparkR.session(enableHiveSupport = FALSE) +sc <- callJStatic("org.apache.spark.sql.api.r.SQLUtils", "getJavaSparkContext", sparkSession) + +test_that("take() gives back the original elements in correct count and order", { + numVectorRDD <- parallelize(sc, numVector, 10) + # case: number of elements to take is less than the size of the first partition + expect_equal(takeRDD(numVectorRDD, 1), as.list(head(numVector, n = 1))) + # case: number of elements to take is the same as the size of the first partition + expect_equal(takeRDD(numVectorRDD, 11), as.list(head(numVector, n = 11))) + # case: number of elements to take is greater than all elements + expect_equal(takeRDD(numVectorRDD, length(numVector)), as.list(numVector)) + expect_equal(takeRDD(numVectorRDD, length(numVector) + 1), as.list(numVector)) + + numListRDD <- parallelize(sc, numList, 1) + numListRDD2 <- parallelize(sc, numList, 4) + expect_equal(takeRDD(numListRDD, 3), takeRDD(numListRDD2, 3)) + expect_equal(takeRDD(numListRDD, 5), takeRDD(numListRDD2, 5)) + expect_equal(takeRDD(numListRDD, 1), as.list(head(numList, n = 1))) + expect_equal(takeRDD(numListRDD2, 999), numList) + + strVectorRDD <- parallelize(sc, strVector, 2) + strVectorRDD2 <- parallelize(sc, strVector, 3) + expect_equal(takeRDD(strVectorRDD, 4), as.list(strVector)) + expect_equal(takeRDD(strVectorRDD2, 2), as.list(head(strVector, n = 2))) + + strListRDD <- parallelize(sc, strList, 4) + strListRDD2 <- parallelize(sc, strList, 1) + expect_equal(takeRDD(strListRDD, 3), as.list(head(strList, n = 3))) + expect_equal(takeRDD(strListRDD2, 1), as.list(head(strList, n = 1))) + + expect_equal(length(takeRDD(strListRDD, 0)), 0) + expect_equal(length(takeRDD(strVectorRDD, 0)), 0) + expect_equal(length(takeRDD(numListRDD, 0)), 0) + expect_equal(length(takeRDD(numVectorRDD, 0)), 0) +}) + +sparkR.session.stop() http://git-wip-us.apache.org/repos/asf/spark/blob/ae4e8ae4/R/pkg/tests/fulltests/test_textFile.R -- diff --git a/R/pkg/tests/fulltests/test_textFile.R b/R/pkg/tests/fulltests/test_textFile.R new file mode 100644 index 000..3b46606 --- /dev/null +++ b/R/pkg/tests/fulltests/test_textFile.R @@ -0,0 +1,164 @@ +# +# Licensed to the Apache Software Foundation (ASF) under one or more +# contributor license agreements. See the NOTICE file distributed with +# this work for additional information regarding copyright ownership. +# The ASF licenses this file to You under the Apache License, Version 2.0 +# (the "License"); you may not use this file except in compliance with +# the License. You may obtain a copy of the License at +# +#http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the spe
[6/9] spark git commit: [SPARKR][BACKPORT-2.1] backporting package and test changes
http://git-wip-us.apache.org/repos/asf/spark/blob/ae4e8ae4/R/pkg/inst/tests/testthat/test_sparkSQL.R -- diff --git a/R/pkg/inst/tests/testthat/test_sparkSQL.R b/R/pkg/inst/tests/testthat/test_sparkSQL.R deleted file mode 100644 index 6285124..000 --- a/R/pkg/inst/tests/testthat/test_sparkSQL.R +++ /dev/null @@ -1,2883 +0,0 @@ -# -# Licensed to the Apache Software Foundation (ASF) under one or more -# contributor license agreements. See the NOTICE file distributed with -# this work for additional information regarding copyright ownership. -# The ASF licenses this file to You under the Apache License, Version 2.0 -# (the "License"); you may not use this file except in compliance with -# the License. You may obtain a copy of the License at -# -#http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on an "AS IS" BASIS, -# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -# See the License for the specific language governing permissions and -# limitations under the License. -# - -library(testthat) - -context("SparkSQL functions") - -# Utility function for easily checking the values of a StructField -checkStructField <- function(actual, expectedName, expectedType, expectedNullable) { - expect_equal(class(actual), "structField") - expect_equal(actual$name(), expectedName) - expect_equal(actual$dataType.toString(), expectedType) - expect_equal(actual$nullable(), expectedNullable) -} - -markUtf8 <- function(s) { - Encoding(s) <- "UTF-8" - s -} - -setHiveContext <- function(sc) { - if (exists(".testHiveSession", envir = .sparkREnv)) { -hiveSession <- get(".testHiveSession", envir = .sparkREnv) - } else { -# initialize once and reuse -ssc <- callJMethod(sc, "sc") -hiveCtx <- tryCatch({ - newJObject("org.apache.spark.sql.hive.test.TestHiveContext", ssc, FALSE) -}, -error = function(err) { - skip("Hive is not build with SparkSQL, skipped") -}) -hiveSession <- callJMethod(hiveCtx, "sparkSession") - } - previousSession <- get(".sparkRsession", envir = .sparkREnv) - assign(".sparkRsession", hiveSession, envir = .sparkREnv) - assign(".prevSparkRsession", previousSession, envir = .sparkREnv) - hiveSession -} - -unsetHiveContext <- function() { - previousSession <- get(".prevSparkRsession", envir = .sparkREnv) - assign(".sparkRsession", previousSession, envir = .sparkREnv) - remove(".prevSparkRsession", envir = .sparkREnv) -} - -# Tests for SparkSQL functions in SparkR - -filesBefore <- list.files(path = sparkRDir, all.files = TRUE) -sparkSession <- sparkR.session() -sc <- callJStatic("org.apache.spark.sql.api.r.SQLUtils", "getJavaSparkContext", sparkSession) - -mockLines <- c("{\"name\":\"Michael\"}", - "{\"name\":\"Andy\", \"age\":30}", - "{\"name\":\"Justin\", \"age\":19}") -jsonPath <- tempfile(pattern = "sparkr-test", fileext = ".tmp") -parquetPath <- tempfile(pattern = "sparkr-test", fileext = ".parquet") -orcPath <- tempfile(pattern = "sparkr-test", fileext = ".orc") -writeLines(mockLines, jsonPath) - -# For test nafunctions, like dropna(), fillna(),... -mockLinesNa <- c("{\"name\":\"Bob\",\"age\":16,\"height\":176.5}", - "{\"name\":\"Alice\",\"age\":null,\"height\":164.3}", - "{\"name\":\"David\",\"age\":60,\"height\":null}", - "{\"name\":\"Amy\",\"age\":null,\"height\":null}", - "{\"name\":null,\"age\":null,\"height\":null}") -jsonPathNa <- tempfile(pattern = "sparkr-test", fileext = ".tmp") -writeLines(mockLinesNa, jsonPathNa) - -# For test complex types in DataFrame -mockLinesComplexType <- - c("{\"c1\":[1, 2, 3], \"c2\":[\"a\", \"b\", \"c\"], \"c3\":[1.0, 2.0, 3.0]}", -"{\"c1\":[4, 5, 6], \"c2\":[\"d\", \"e\", \"f\"], \"c3\":[4.0, 5.0, 6.0]}", -"{\"c1\":[7, 8, 9], \"c2\":[\"g\", \"h\", \"i\"], \"c3\":[7.0, 8.0, 9.0]}") -complexTypeJsonPath <- tempfile(pattern = "sparkr-test", fileext = ".tmp") -writeLines(mockLinesComplexType, complexTypeJsonPath) - -test_that("calling sparkRSQL.init returns existing SQL context", { - sqlContext <- suppressWarnings(sparkRSQL.init(sc)) - expect_equal(suppressWarnings(sparkRSQL.init(sc)), sqlContext) -}) - -test_that("calling sparkRSQL.init returns existing SparkSession", { - expect_equal(suppressWarnings(sparkRSQL.init(sc)), sparkSession) -}) - -test_that("calling sparkR.session returns existing SparkSession", { - expect_equal(sparkR.session(), sparkSession) -}) - -test_that("infer types and check types", { - expect_equal(infer_type(1L), "integer") - expect_equal(infer_type(1.0), "double") - expect_equal(infer_type("abc"), "string") - expect_equal(infer_type(TRUE), "boolean") - expect_equal(infer_type(as.Date("2015-03-11")), "date") - expect_equal(infer_type(as.POSIXlt("2015-03-11 12:13:04.043")), "time
[2/9] spark git commit: [SPARKR][BACKPORT-2.1] backporting package and test changes
http://git-wip-us.apache.org/repos/asf/spark/blob/ae4e8ae4/R/pkg/tests/fulltests/test_sparkSQL.R -- diff --git a/R/pkg/tests/fulltests/test_sparkSQL.R b/R/pkg/tests/fulltests/test_sparkSQL.R new file mode 100644 index 000..07de45b --- /dev/null +++ b/R/pkg/tests/fulltests/test_sparkSQL.R @@ -0,0 +1,2887 @@ +# +# Licensed to the Apache Software Foundation (ASF) under one or more +# contributor license agreements. See the NOTICE file distributed with +# this work for additional information regarding copyright ownership. +# The ASF licenses this file to You under the Apache License, Version 2.0 +# (the "License"); you may not use this file except in compliance with +# the License. You may obtain a copy of the License at +# +#http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# + +library(testthat) + +context("SparkSQL functions") + +# Utility function for easily checking the values of a StructField +checkStructField <- function(actual, expectedName, expectedType, expectedNullable) { + expect_equal(class(actual), "structField") + expect_equal(actual$name(), expectedName) + expect_equal(actual$dataType.toString(), expectedType) + expect_equal(actual$nullable(), expectedNullable) +} + +markUtf8 <- function(s) { + Encoding(s) <- "UTF-8" + s +} + +setHiveContext <- function(sc) { + if (exists(".testHiveSession", envir = .sparkREnv)) { +hiveSession <- get(".testHiveSession", envir = .sparkREnv) + } else { +# initialize once and reuse +ssc <- callJMethod(sc, "sc") +hiveCtx <- tryCatch({ + newJObject("org.apache.spark.sql.hive.test.TestHiveContext", ssc, FALSE) +}, +error = function(err) { + skip("Hive is not build with SparkSQL, skipped") +}) +hiveSession <- callJMethod(hiveCtx, "sparkSession") + } + previousSession <- get(".sparkRsession", envir = .sparkREnv) + assign(".sparkRsession", hiveSession, envir = .sparkREnv) + assign(".prevSparkRsession", previousSession, envir = .sparkREnv) + hiveSession +} + +unsetHiveContext <- function() { + previousSession <- get(".prevSparkRsession", envir = .sparkREnv) + assign(".sparkRsession", previousSession, envir = .sparkREnv) + remove(".prevSparkRsession", envir = .sparkREnv) +} + +# Tests for SparkSQL functions in SparkR + +filesBefore <- list.files(path = sparkRDir, all.files = TRUE) +sparkSession <- sparkR.session() +sc <- callJStatic("org.apache.spark.sql.api.r.SQLUtils", "getJavaSparkContext", sparkSession) + +mockLines <- c("{\"name\":\"Michael\"}", + "{\"name\":\"Andy\", \"age\":30}", + "{\"name\":\"Justin\", \"age\":19}") +jsonPath <- tempfile(pattern = "sparkr-test", fileext = ".tmp") +parquetPath <- tempfile(pattern = "sparkr-test", fileext = ".parquet") +orcPath <- tempfile(pattern = "sparkr-test", fileext = ".orc") +writeLines(mockLines, jsonPath) + +# For test nafunctions, like dropna(), fillna(),... +mockLinesNa <- c("{\"name\":\"Bob\",\"age\":16,\"height\":176.5}", + "{\"name\":\"Alice\",\"age\":null,\"height\":164.3}", + "{\"name\":\"David\",\"age\":60,\"height\":null}", + "{\"name\":\"Amy\",\"age\":null,\"height\":null}", + "{\"name\":null,\"age\":null,\"height\":null}") +jsonPathNa <- tempfile(pattern = "sparkr-test", fileext = ".tmp") +writeLines(mockLinesNa, jsonPathNa) + +# For test complex types in DataFrame +mockLinesComplexType <- + c("{\"c1\":[1, 2, 3], \"c2\":[\"a\", \"b\", \"c\"], \"c3\":[1.0, 2.0, 3.0]}", +"{\"c1\":[4, 5, 6], \"c2\":[\"d\", \"e\", \"f\"], \"c3\":[4.0, 5.0, 6.0]}", +"{\"c1\":[7, 8, 9], \"c2\":[\"g\", \"h\", \"i\"], \"c3\":[7.0, 8.0, 9.0]}") +complexTypeJsonPath <- tempfile(pattern = "sparkr-test", fileext = ".tmp") +writeLines(mockLinesComplexType, complexTypeJsonPath) + +if (.Platform$OS.type == "windows") { + Sys.setenv(TZ = "GMT") +} + +test_that("calling sparkRSQL.init returns existing SQL context", { + sqlContext <- suppressWarnings(sparkRSQL.init(sc)) + expect_equal(suppressWarnings(sparkRSQL.init(sc)), sqlContext) +}) + +test_that("calling sparkRSQL.init returns existing SparkSession", { + expect_equal(suppressWarnings(sparkRSQL.init(sc)), sparkSession) +}) + +test_that("calling sparkR.session returns existing SparkSession", { + expect_equal(sparkR.session(), sparkSession) +}) + +test_that("infer types and check types", { + expect_equal(infer_type(1L), "integer") + expect_equal(infer_type(1.0), "double") + expect_equal(infer_type("abc"), "string") + expect_equal(infer_type(TRUE), "boolean") + expect_equal(infer_type(as.Date("2015-03-11")), "date") + expect_equal(infer_
[9/9] spark git commit: [SPARKR][BACKPORT-2.1] backporting package and test changes
[SPARKR][BACKPORT-2.1] backporting package and test changes ## What changes were proposed in this pull request? cherrypick or manually porting changes to 2.1 ## How was this patch tested? Jenkins Author: Felix Cheung Author: hyukjinkwon Author: Wayne Zhang Closes #19165 from felixcheung/rbackportpkg21. Project: http://git-wip-us.apache.org/repos/asf/spark/repo Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/ae4e8ae4 Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/ae4e8ae4 Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/ae4e8ae4 Branch: refs/heads/branch-2.1 Commit: ae4e8ae41d1ba135159afe9ffb1302971343efd1 Parents: 6a8a726 Author: Felix Cheung Authored: Sun Sep 10 10:24:46 2017 -0700 Committer: Felix Cheung Committed: Sun Sep 10 10:24:46 2017 -0700 -- R/pkg/.Rbuildignore |2 + R/pkg/DESCRIPTION |2 +- R/pkg/R/install.R |6 +- R/pkg/inst/tests/testthat/jarTest.R | 32 - R/pkg/inst/tests/testthat/packageInAJarTest.R | 30 - R/pkg/inst/tests/testthat/test_Serde.R | 79 - R/pkg/inst/tests/testthat/test_Windows.R| 27 - R/pkg/inst/tests/testthat/test_basic.R | 72 + R/pkg/inst/tests/testthat/test_binaryFile.R | 92 - .../inst/tests/testthat/test_binary_function.R | 104 - R/pkg/inst/tests/testthat/test_broadcast.R | 51 - R/pkg/inst/tests/testthat/test_client.R | 43 - R/pkg/inst/tests/testthat/test_context.R| 210 -- R/pkg/inst/tests/testthat/test_includePackage.R | 60 - R/pkg/inst/tests/testthat/test_jvm_api.R| 36 - R/pkg/inst/tests/testthat/test_mllib.R | 1205 .../tests/testthat/test_parallelize_collect.R | 112 - R/pkg/inst/tests/testthat/test_rdd.R| 804 - R/pkg/inst/tests/testthat/test_shuffle.R| 224 -- R/pkg/inst/tests/testthat/test_sparkR.R | 46 - R/pkg/inst/tests/testthat/test_sparkSQL.R | 2883 - R/pkg/inst/tests/testthat/test_take.R | 69 - R/pkg/inst/tests/testthat/test_textFile.R | 164 - R/pkg/inst/tests/testthat/test_utils.R | 242 -- R/pkg/tests/fulltests/jarTest.R | 32 + R/pkg/tests/fulltests/packageInAJarTest.R | 30 + R/pkg/tests/fulltests/test_Serde.R | 79 + R/pkg/tests/fulltests/test_Windows.R| 27 + R/pkg/tests/fulltests/test_binaryFile.R | 92 + R/pkg/tests/fulltests/test_binary_function.R| 104 + R/pkg/tests/fulltests/test_broadcast.R | 51 + R/pkg/tests/fulltests/test_client.R | 43 + R/pkg/tests/fulltests/test_context.R| 210 ++ R/pkg/tests/fulltests/test_includePackage.R | 60 + R/pkg/tests/fulltests/test_jvm_api.R| 36 + R/pkg/tests/fulltests/test_mllib.R | 1205 .../tests/fulltests/test_parallelize_collect.R | 112 + R/pkg/tests/fulltests/test_rdd.R| 804 + R/pkg/tests/fulltests/test_shuffle.R| 224 ++ R/pkg/tests/fulltests/test_sparkR.R | 46 + R/pkg/tests/fulltests/test_sparkSQL.R | 2887 ++ R/pkg/tests/fulltests/test_take.R | 69 + R/pkg/tests/fulltests/test_textFile.R | 164 + R/pkg/tests/fulltests/test_utils.R | 242 ++ R/pkg/tests/run-all.R | 22 +- R/pkg/vignettes/sparkr-vignettes.Rmd| 47 +- R/run-tests.sh |2 +- appveyor.yml|3 + .../apache/spark/deploy/SparkSubmitSuite.scala |6 +- 49 files changed, 6653 insertions(+), 6539 deletions(-) -- http://git-wip-us.apache.org/repos/asf/spark/blob/ae4e8ae4/R/pkg/.Rbuildignore -- diff --git a/R/pkg/.Rbuildignore b/R/pkg/.Rbuildignore index f12f8c2..280ab58 100644 --- a/R/pkg/.Rbuildignore +++ b/R/pkg/.Rbuildignore @@ -6,3 +6,5 @@ ^README\.Rmd$ ^src-native$ ^html$ +^tests/fulltests/* + http://git-wip-us.apache.org/repos/asf/spark/blob/ae4e8ae4/R/pkg/DESCRIPTION -- diff --git a/R/pkg/DESCRIPTION b/R/pkg/DESCRIPTION index 2d461ca..899d410 100644 --- a/R/pkg/DESCRIPTION +++ b/R/pkg/DESCRIPTION @@ -2,7 +2,7 @@ Package: SparkR Type: Package Version: 2.1.2 Title: R Frontend for Apache Spark -Description: The SparkR package provides an R Frontend for Apache Spark. +Description: Provides an R Frontend for Apache Spark. Authors@R: c(person("Shivaram", "Venkataraman", role = c("aut", "cre"), email = "shiva...@cs.berkeley.edu"), person("Xiangrui", "Meng", role = "aut", http://git-wip-us.apache.org/repo
[5/9] spark git commit: [SPARKR][BACKPORT-2.1] backporting package and test changes
http://git-wip-us.apache.org/repos/asf/spark/blob/ae4e8ae4/R/pkg/inst/tests/testthat/test_take.R -- diff --git a/R/pkg/inst/tests/testthat/test_take.R b/R/pkg/inst/tests/testthat/test_take.R deleted file mode 100644 index aaa5328..000 --- a/R/pkg/inst/tests/testthat/test_take.R +++ /dev/null @@ -1,69 +0,0 @@ -# -# Licensed to the Apache Software Foundation (ASF) under one or more -# contributor license agreements. See the NOTICE file distributed with -# this work for additional information regarding copyright ownership. -# The ASF licenses this file to You under the Apache License, Version 2.0 -# (the "License"); you may not use this file except in compliance with -# the License. You may obtain a copy of the License at -# -#http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on an "AS IS" BASIS, -# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -# See the License for the specific language governing permissions and -# limitations under the License. -# - -context("tests RDD function take()") - -# Mock data -numVector <- c(-10:97) -numList <- list(sqrt(1), sqrt(2), sqrt(3), 4 ** 10) -strVector <- c("Dexter Morgan: I suppose I should be upset, even feel", - "violated, but I'm not. No, in fact, I think this is a friendly", - "message, like \"Hey, wanna play?\" and yes, I want to play. ", - "I really, really do.") -strList <- list("Dexter Morgan: Blood. Sometimes it sets my teeth on edge, ", -"other times it helps me control the chaos.", -"Dexter Morgan: Harry and Dorris Morgan did a wonderful job ", -"raising me. But they're both dead now. I didn't kill them. Honest.") - -# JavaSparkContext handle -sparkSession <- sparkR.session(enableHiveSupport = FALSE) -sc <- callJStatic("org.apache.spark.sql.api.r.SQLUtils", "getJavaSparkContext", sparkSession) - -test_that("take() gives back the original elements in correct count and order", { - numVectorRDD <- parallelize(sc, numVector, 10) - # case: number of elements to take is less than the size of the first partition - expect_equal(takeRDD(numVectorRDD, 1), as.list(head(numVector, n = 1))) - # case: number of elements to take is the same as the size of the first partition - expect_equal(takeRDD(numVectorRDD, 11), as.list(head(numVector, n = 11))) - # case: number of elements to take is greater than all elements - expect_equal(takeRDD(numVectorRDD, length(numVector)), as.list(numVector)) - expect_equal(takeRDD(numVectorRDD, length(numVector) + 1), as.list(numVector)) - - numListRDD <- parallelize(sc, numList, 1) - numListRDD2 <- parallelize(sc, numList, 4) - expect_equal(takeRDD(numListRDD, 3), takeRDD(numListRDD2, 3)) - expect_equal(takeRDD(numListRDD, 5), takeRDD(numListRDD2, 5)) - expect_equal(takeRDD(numListRDD, 1), as.list(head(numList, n = 1))) - expect_equal(takeRDD(numListRDD2, 999), numList) - - strVectorRDD <- parallelize(sc, strVector, 2) - strVectorRDD2 <- parallelize(sc, strVector, 3) - expect_equal(takeRDD(strVectorRDD, 4), as.list(strVector)) - expect_equal(takeRDD(strVectorRDD2, 2), as.list(head(strVector, n = 2))) - - strListRDD <- parallelize(sc, strList, 4) - strListRDD2 <- parallelize(sc, strList, 1) - expect_equal(takeRDD(strListRDD, 3), as.list(head(strList, n = 3))) - expect_equal(takeRDD(strListRDD2, 1), as.list(head(strList, n = 1))) - - expect_equal(length(takeRDD(strListRDD, 0)), 0) - expect_equal(length(takeRDD(strVectorRDD, 0)), 0) - expect_equal(length(takeRDD(numListRDD, 0)), 0) - expect_equal(length(takeRDD(numVectorRDD, 0)), 0) -}) - -sparkR.session.stop() http://git-wip-us.apache.org/repos/asf/spark/blob/ae4e8ae4/R/pkg/inst/tests/testthat/test_textFile.R -- diff --git a/R/pkg/inst/tests/testthat/test_textFile.R b/R/pkg/inst/tests/testthat/test_textFile.R deleted file mode 100644 index 3b46606..000 --- a/R/pkg/inst/tests/testthat/test_textFile.R +++ /dev/null @@ -1,164 +0,0 @@ -# -# Licensed to the Apache Software Foundation (ASF) under one or more -# contributor license agreements. See the NOTICE file distributed with -# this work for additional information regarding copyright ownership. -# The ASF licenses this file to You under the Apache License, Version 2.0 -# (the "License"); you may not use this file except in compliance with -# the License. You may obtain a copy of the License at -# -#http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on an "AS IS" BASIS, -# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -# See the License for the specific language governing permissions and -# l
[7/9] spark git commit: [SPARKR][BACKPORT-2.1] backporting package and test changes
http://git-wip-us.apache.org/repos/asf/spark/blob/ae4e8ae4/R/pkg/inst/tests/testthat/test_shuffle.R -- diff --git a/R/pkg/inst/tests/testthat/test_shuffle.R b/R/pkg/inst/tests/testthat/test_shuffle.R deleted file mode 100644 index d38efab..000 --- a/R/pkg/inst/tests/testthat/test_shuffle.R +++ /dev/null @@ -1,224 +0,0 @@ -# -# Licensed to the Apache Software Foundation (ASF) under one or more -# contributor license agreements. See the NOTICE file distributed with -# this work for additional information regarding copyright ownership. -# The ASF licenses this file to You under the Apache License, Version 2.0 -# (the "License"); you may not use this file except in compliance with -# the License. You may obtain a copy of the License at -# -#http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on an "AS IS" BASIS, -# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -# See the License for the specific language governing permissions and -# limitations under the License. -# - -context("partitionBy, groupByKey, reduceByKey etc.") - -# JavaSparkContext handle -sparkSession <- sparkR.session(enableHiveSupport = FALSE) -sc <- callJStatic("org.apache.spark.sql.api.r.SQLUtils", "getJavaSparkContext", sparkSession) - -# Data -intPairs <- list(list(1L, -1), list(2L, 100), list(2L, 1), list(1L, 200)) -intRdd <- parallelize(sc, intPairs, 2L) - -doublePairs <- list(list(1.5, -1), list(2.5, 100), list(2.5, 1), list(1.5, 200)) -doubleRdd <- parallelize(sc, doublePairs, 2L) - -numPairs <- list(list(1L, 100), list(2L, 200), list(4L, -1), list(3L, 1), - list(3L, 0)) -numPairsRdd <- parallelize(sc, numPairs, length(numPairs)) - -strList <- list("Dexter Morgan: Blood. Sometimes it sets my teeth on edge and ", -"Dexter Morgan: Harry and Dorris Morgan did a wonderful job ") -strListRDD <- parallelize(sc, strList, 4) - -test_that("groupByKey for integers", { - grouped <- groupByKey(intRdd, 2L) - - actual <- collectRDD(grouped) - - expected <- list(list(2L, list(100, 1)), list(1L, list(-1, 200))) - expect_equal(sortKeyValueList(actual), sortKeyValueList(expected)) -}) - -test_that("groupByKey for doubles", { - grouped <- groupByKey(doubleRdd, 2L) - - actual <- collectRDD(grouped) - - expected <- list(list(1.5, list(-1, 200)), list(2.5, list(100, 1))) - expect_equal(sortKeyValueList(actual), sortKeyValueList(expected)) -}) - -test_that("reduceByKey for ints", { - reduced <- reduceByKey(intRdd, "+", 2L) - - actual <- collectRDD(reduced) - - expected <- list(list(2L, 101), list(1L, 199)) - expect_equal(sortKeyValueList(actual), sortKeyValueList(expected)) -}) - -test_that("reduceByKey for doubles", { - reduced <- reduceByKey(doubleRdd, "+", 2L) - actual <- collectRDD(reduced) - - expected <- list(list(1.5, 199), list(2.5, 101)) - expect_equal(sortKeyValueList(actual), sortKeyValueList(expected)) -}) - -test_that("combineByKey for ints", { - reduced <- combineByKey(intRdd, function(x) { x }, "+", "+", 2L) - - actual <- collectRDD(reduced) - - expected <- list(list(2L, 101), list(1L, 199)) - expect_equal(sortKeyValueList(actual), sortKeyValueList(expected)) -}) - -test_that("combineByKey for doubles", { - reduced <- combineByKey(doubleRdd, function(x) { x }, "+", "+", 2L) - actual <- collectRDD(reduced) - - expected <- list(list(1.5, 199), list(2.5, 101)) - expect_equal(sortKeyValueList(actual), sortKeyValueList(expected)) -}) - -test_that("combineByKey for characters", { - stringKeyRDD <- parallelize(sc, - list(list("max", 1L), list("min", 2L), - list("other", 3L), list("max", 4L)), 2L) - reduced <- combineByKey(stringKeyRDD, - function(x) { x }, "+", "+", 2L) - actual <- collectRDD(reduced) - - expected <- list(list("max", 5L), list("min", 2L), list("other", 3L)) - expect_equal(sortKeyValueList(actual), sortKeyValueList(expected)) -}) - -test_that("aggregateByKey", { - # test aggregateByKey for int keys - rdd <- parallelize(sc, list(list(1, 1), list(1, 2), list(2, 3), list(2, 4))) - - zeroValue <- list(0, 0) - seqOp <- function(x, y) { list(x[[1]] + y, x[[2]] + 1) } - combOp <- function(x, y) { list(x[[1]] + y[[1]], x[[2]] + y[[2]]) } - aggregatedRDD <- aggregateByKey(rdd, zeroValue, seqOp, combOp, 2L) - - actual <- collectRDD(aggregatedRDD) - - expected <- list(list(1, list(3, 2)), list(2, list(7, 2))) - expect_equal(sortKeyValueList(actual), sortKeyValueList(expected)) - - # test aggregateByKey for string keys - rdd <- parallelize(sc, list(list("a", 1), list("a", 2), list("b", 3), list("b", 4))) - - zeroValue <- list(0, 0) - seqOp <- function(x, y) { list(x[[1]] + y, x[[2]] + 1) } - combOp <- function(x, y) { list(x[[1]] + y[[1]], x[[2]] + y[[2]]) } - aggregate
[8/9] spark git commit: [SPARKR][BACKPORT-2.1] backporting package and test changes
http://git-wip-us.apache.org/repos/asf/spark/blob/ae4e8ae4/R/pkg/inst/tests/testthat/test_mllib.R -- diff --git a/R/pkg/inst/tests/testthat/test_mllib.R b/R/pkg/inst/tests/testthat/test_mllib.R deleted file mode 100644 index 8fe3a87..000 --- a/R/pkg/inst/tests/testthat/test_mllib.R +++ /dev/null @@ -1,1205 +0,0 @@ -# -# Licensed to the Apache Software Foundation (ASF) under one or more -# contributor license agreements. See the NOTICE file distributed with -# this work for additional information regarding copyright ownership. -# The ASF licenses this file to You under the Apache License, Version 2.0 -# (the "License"); you may not use this file except in compliance with -# the License. You may obtain a copy of the License at -# -#http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on an "AS IS" BASIS, -# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -# See the License for the specific language governing permissions and -# limitations under the License. -# - -library(testthat) - -context("MLlib functions") - -# Tests for MLlib functions in SparkR -sparkSession <- sparkR.session(enableHiveSupport = FALSE) - -absoluteSparkPath <- function(x) { - sparkHome <- sparkR.conf("spark.home") - file.path(sparkHome, x) -} - -test_that("formula of spark.glm", { - training <- suppressWarnings(createDataFrame(iris)) - # directly calling the spark API - # dot minus and intercept vs native glm - model <- spark.glm(training, Sepal_Width ~ . - Species + 0) - vals <- collect(select(predict(model, training), "prediction")) - rVals <- predict(glm(Sepal.Width ~ . - Species + 0, data = iris), iris) - expect_true(all(abs(rVals - vals) < 1e-6), rVals - vals) - - # feature interaction vs native glm - model <- spark.glm(training, Sepal_Width ~ Species:Sepal_Length) - vals <- collect(select(predict(model, training), "prediction")) - rVals <- predict(glm(Sepal.Width ~ Species:Sepal.Length, data = iris), iris) - expect_true(all(abs(rVals - vals) < 1e-6), rVals - vals) - - # glm should work with long formula - training <- suppressWarnings(createDataFrame(iris)) - training$LongLongLongLongLongName <- training$Sepal_Width - training$VeryLongLongLongLonLongName <- training$Sepal_Length - training$AnotherLongLongLongLongName <- training$Species - model <- spark.glm(training, LongLongLongLongLongName ~ VeryLongLongLongLonLongName + -AnotherLongLongLongLongName) - vals <- collect(select(predict(model, training), "prediction")) - rVals <- predict(glm(Sepal.Width ~ Sepal.Length + Species, data = iris), iris) - expect_true(all(abs(rVals - vals) < 1e-6), rVals - vals) -}) - -test_that("spark.glm and predict", { - training <- suppressWarnings(createDataFrame(iris)) - # gaussian family - model <- spark.glm(training, Sepal_Width ~ Sepal_Length + Species) - prediction <- predict(model, training) - expect_equal(typeof(take(select(prediction, "prediction"), 1)$prediction), "double") - vals <- collect(select(prediction, "prediction")) - rVals <- predict(glm(Sepal.Width ~ Sepal.Length + Species, data = iris), iris) - expect_true(all(abs(rVals - vals) < 1e-6), rVals - vals) - - # poisson family - model <- spark.glm(training, Sepal_Width ~ Sepal_Length + Species, - family = poisson(link = identity)) - prediction <- predict(model, training) - expect_equal(typeof(take(select(prediction, "prediction"), 1)$prediction), "double") - vals <- collect(select(prediction, "prediction")) - rVals <- suppressWarnings(predict(glm(Sepal.Width ~ Sepal.Length + Species, - data = iris, family = poisson(link = identity)), iris)) - expect_true(all(abs(rVals - vals) < 1e-6), rVals - vals) - - # Gamma family - x <- runif(100, -1, 1) - y <- rgamma(100, rate = 10 / exp(0.5 + 1.2 * x), shape = 10) - df <- as.DataFrame(as.data.frame(list(x = x, y = y))) - model <- glm(y ~ x, family = Gamma, df) - out <- capture.output(print(summary(model))) - expect_true(any(grepl("Dispersion parameter for gamma family", out))) - - # Test stats::predict is working - x <- rnorm(15) - y <- x + rnorm(15) - expect_equal(length(predict(lm(y ~ x))), 15) -}) - -test_that("spark.glm summary", { - # gaussian family - training <- suppressWarnings(createDataFrame(iris)) - stats <- summary(spark.glm(training, Sepal_Width ~ Sepal_Length + Species)) - - rStats <- summary(glm(Sepal.Width ~ Sepal.Length + Species, data = iris)) - - coefs <- unlist(stats$coefficients) - rCoefs <- unlist(rStats$coefficients) - expect_true(all(abs(rCoefs - coefs) < 1e-4)) - expect_true(all( -rownames(stats$coefficients) == -c("(Intercept)", "Sepal_Length", "Species_versicolor", "Species_virginica"))) - expect_equal(stats$dispersion, rStats$dispersion) - expect_equal(stats$null.deviance, rStats$null.deviance) - expect_equal(stats$deviance, rSta
[4/9] spark git commit: [SPARKR][BACKPORT-2.1] backporting package and test changes
http://git-wip-us.apache.org/repos/asf/spark/blob/ae4e8ae4/R/pkg/tests/fulltests/test_mllib.R -- diff --git a/R/pkg/tests/fulltests/test_mllib.R b/R/pkg/tests/fulltests/test_mllib.R new file mode 100644 index 000..8fe3a87 --- /dev/null +++ b/R/pkg/tests/fulltests/test_mllib.R @@ -0,0 +1,1205 @@ +# +# Licensed to the Apache Software Foundation (ASF) under one or more +# contributor license agreements. See the NOTICE file distributed with +# this work for additional information regarding copyright ownership. +# The ASF licenses this file to You under the Apache License, Version 2.0 +# (the "License"); you may not use this file except in compliance with +# the License. You may obtain a copy of the License at +# +#http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# + +library(testthat) + +context("MLlib functions") + +# Tests for MLlib functions in SparkR +sparkSession <- sparkR.session(enableHiveSupport = FALSE) + +absoluteSparkPath <- function(x) { + sparkHome <- sparkR.conf("spark.home") + file.path(sparkHome, x) +} + +test_that("formula of spark.glm", { + training <- suppressWarnings(createDataFrame(iris)) + # directly calling the spark API + # dot minus and intercept vs native glm + model <- spark.glm(training, Sepal_Width ~ . - Species + 0) + vals <- collect(select(predict(model, training), "prediction")) + rVals <- predict(glm(Sepal.Width ~ . - Species + 0, data = iris), iris) + expect_true(all(abs(rVals - vals) < 1e-6), rVals - vals) + + # feature interaction vs native glm + model <- spark.glm(training, Sepal_Width ~ Species:Sepal_Length) + vals <- collect(select(predict(model, training), "prediction")) + rVals <- predict(glm(Sepal.Width ~ Species:Sepal.Length, data = iris), iris) + expect_true(all(abs(rVals - vals) < 1e-6), rVals - vals) + + # glm should work with long formula + training <- suppressWarnings(createDataFrame(iris)) + training$LongLongLongLongLongName <- training$Sepal_Width + training$VeryLongLongLongLonLongName <- training$Sepal_Length + training$AnotherLongLongLongLongName <- training$Species + model <- spark.glm(training, LongLongLongLongLongName ~ VeryLongLongLongLonLongName + +AnotherLongLongLongLongName) + vals <- collect(select(predict(model, training), "prediction")) + rVals <- predict(glm(Sepal.Width ~ Sepal.Length + Species, data = iris), iris) + expect_true(all(abs(rVals - vals) < 1e-6), rVals - vals) +}) + +test_that("spark.glm and predict", { + training <- suppressWarnings(createDataFrame(iris)) + # gaussian family + model <- spark.glm(training, Sepal_Width ~ Sepal_Length + Species) + prediction <- predict(model, training) + expect_equal(typeof(take(select(prediction, "prediction"), 1)$prediction), "double") + vals <- collect(select(prediction, "prediction")) + rVals <- predict(glm(Sepal.Width ~ Sepal.Length + Species, data = iris), iris) + expect_true(all(abs(rVals - vals) < 1e-6), rVals - vals) + + # poisson family + model <- spark.glm(training, Sepal_Width ~ Sepal_Length + Species, + family = poisson(link = identity)) + prediction <- predict(model, training) + expect_equal(typeof(take(select(prediction, "prediction"), 1)$prediction), "double") + vals <- collect(select(prediction, "prediction")) + rVals <- suppressWarnings(predict(glm(Sepal.Width ~ Sepal.Length + Species, + data = iris, family = poisson(link = identity)), iris)) + expect_true(all(abs(rVals - vals) < 1e-6), rVals - vals) + + # Gamma family + x <- runif(100, -1, 1) + y <- rgamma(100, rate = 10 / exp(0.5 + 1.2 * x), shape = 10) + df <- as.DataFrame(as.data.frame(list(x = x, y = y))) + model <- glm(y ~ x, family = Gamma, df) + out <- capture.output(print(summary(model))) + expect_true(any(grepl("Dispersion parameter for gamma family", out))) + + # Test stats::predict is working + x <- rnorm(15) + y <- x + rnorm(15) + expect_equal(length(predict(lm(y ~ x))), 15) +}) + +test_that("spark.glm summary", { + # gaussian family + training <- suppressWarnings(createDataFrame(iris)) + stats <- summary(spark.glm(training, Sepal_Width ~ Sepal_Length + Species)) + + rStats <- summary(glm(Sepal.Width ~ Sepal.Length + Species, data = iris)) + + coefs <- unlist(stats$coefficients) + rCoefs <- unlist(rStats$coefficients) + expect_true(all(abs(rCoefs - coefs) < 1e-4)) + expect_true(all( +rownames(stats$coefficients) == +c("(Intercept)", "Sepal_Length", "Species_versicolor", "Species_virginica"))) + expect_equal(stats$dispersion, rStats$dispersion) + expect_equal(stats$null.deviance, rStats$null.deviance) + expect_equal(stats$deviance, rStats$deviance) + expe
spark git commit: [SPARK-21610][SQL] Corrupt records are not handled properly when creating a dataframe from a file
Repository: spark Updated Branches: refs/heads/master 520d92a19 -> 6273a711b [SPARK-21610][SQL] Corrupt records are not handled properly when creating a dataframe from a file ## What changes were proposed in this pull request? ``` echo '{"field": 1} {"field": 2} {"field": "3"}' >/tmp/sample.json ``` ```scala import org.apache.spark.sql.types._ val schema = new StructType() .add("field", ByteType) .add("_corrupt_record", StringType) val file = "/tmp/sample.json" val dfFromFile = spark.read.schema(schema).json(file) scala> dfFromFile.show(false) +-+---+ |field|_corrupt_record| +-+---+ |1|null | |2|null | |null |{"field": "3"} | +-+---+ scala> dfFromFile.filter($"_corrupt_record".isNotNull).count() res1: Long = 0 scala> dfFromFile.filter($"_corrupt_record".isNull).count() res2: Long = 3 ``` When the `requiredSchema` only contains `_corrupt_record`, the derived `actualSchema` is empty and the `_corrupt_record` are all null for all rows. This PR captures above situation and raise an exception with a reasonable workaround messag so that users can know what happened and how to fix the query. ## How was this patch tested? Added test case. Author: Jen-Ming Chung Closes #18865 from jmchung/SPARK-21610. Project: http://git-wip-us.apache.org/repos/asf/spark/repo Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/6273a711 Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/6273a711 Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/6273a711 Branch: refs/heads/master Commit: 6273a711b69139ef0210f59759030a0b4a26b118 Parents: 520d92a Author: Jen-Ming Chung Authored: Sun Sep 10 17:26:43 2017 -0700 Committer: gatorsmile Committed: Sun Sep 10 17:26:43 2017 -0700 -- docs/sql-programming-guide.md | 4 +++ .../datasources/json/JsonFileFormat.scala | 14 ++ .../execution/datasources/json/JsonSuite.scala | 29 3 files changed, 47 insertions(+) -- http://git-wip-us.apache.org/repos/asf/spark/blob/6273a711/docs/sql-programming-guide.md -- diff --git a/docs/sql-programming-guide.md b/docs/sql-programming-guide.md index 45ba4d1..0a8acbb 100644 --- a/docs/sql-programming-guide.md +++ b/docs/sql-programming-guide.md @@ -1543,6 +1543,10 @@ options. # Migration Guide +## Upgrading From Spark SQL 2.2 to 2.3 + + - Since Spark 2.3, the queries from raw JSON/CSV files are disallowed when the referenced columns only include the internal corrupt record column (named `_corrupt_record` by default). For example, `spark.read.schema(schema).json(file).filter($"_corrupt_record".isNotNull).count()` and `spark.read.schema(schema).json(file).select("_corrupt_record").show()`. Instead, you can cache or save the parsed results and then send the same query. For example, `val df = spark.read.schema(schema).json(file).cache()` and then `df.filter($"_corrupt_record".isNotNull).count()`. + ## Upgrading From Spark SQL 2.1 to 2.2 - Spark 2.1.1 introduced a new configuration key: `spark.sql.hive.caseSensitiveInferenceMode`. It had a default setting of `NEVER_INFER`, which kept behavior identical to 2.1.0. However, Spark 2.2.0 changes this setting's default value to `INFER_AND_SAVE` to restore compatibility with reading Hive metastore tables whose underlying file schema have mixed-case column names. With the `INFER_AND_SAVE` configuration value, on first access Spark will perform schema inference on any Hive metastore table for which it has not already saved an inferred schema. Note that schema inference can be a very time consuming operation for tables with thousands of partitions. If compatibility with mixed-case column names is not a concern, you can safely set `spark.sql.hive.caseSensitiveInferenceMode` to `NEVER_INFER` to avoid the initial overhead of schema inference. Note that with the new default `INFER_AND_SAVE` setting, the results of the schema inference are saved as a metastore key for future use . Therefore, the initial schema inference occurs only at a table's first access. http://git-wip-us.apache.org/repos/asf/spark/blob/6273a711/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/json/JsonFileFormat.scala -- diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/json/JsonFileFormat.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/json/JsonFileFormat.scala index 53d62d8..b5ed6e4 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/json/JsonFileFormat.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/json/JsonFileFormat.scala @@ -113,6 +113,20
spark git commit: [BUILD][TEST][SPARKR] add sparksubmitsuite to appveyor tests
Repository: spark Updated Branches: refs/heads/master 6273a711b -> 828fab035 [BUILD][TEST][SPARKR] add sparksubmitsuite to appveyor tests ## What changes were proposed in this pull request? more file regex ## How was this patch tested? Jenkins, AppVeyor Author: Felix Cheung Closes #19177 from felixcheung/rmoduletotest. Project: http://git-wip-us.apache.org/repos/asf/spark/repo Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/828fab03 Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/828fab03 Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/828fab03 Branch: refs/heads/master Commit: 828fab03567ecc245a65c4d295a677ce0ba26c19 Parents: 6273a71 Author: Felix Cheung Authored: Mon Sep 11 09:32:25 2017 +0900 Committer: hyukjinkwon Committed: Mon Sep 11 09:32:25 2017 +0900 -- appveyor.yml | 1 + 1 file changed, 1 insertion(+) -- http://git-wip-us.apache.org/repos/asf/spark/blob/828fab03/appveyor.yml -- diff --git a/appveyor.yml b/appveyor.yml index 43dad9b..dc2d81f 100644 --- a/appveyor.yml +++ b/appveyor.yml @@ -32,6 +32,7 @@ only_commits: - sql/core/src/main/scala/org/apache/spark/sql/api/r/ - core/src/main/scala/org/apache/spark/api/r/ - mllib/src/main/scala/org/apache/spark/ml/r/ +- core/src/test/scala/org/apache/spark/deploy/SparkSubmitSuite.scala cache: - C:\Users\appveyor\.m2 - To unsubscribe, e-mail: commits-unsubscr...@spark.apache.org For additional commands, e-mail: commits-h...@spark.apache.org