Skip to content

Commit 1737598

Browse files
yu-iskwyu-iskw
committed
[SPARK-8549][SparkR] Fix the line length of SparkR
1 parent d983819 commit 1737598

7 files changed

Lines changed: 51 additions & 31 deletions

File tree

‎R/pkg/R/generics.R‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -20,7 +20,8 @@
2020
# @rdname aggregateRDD
2121
# @seealso reduce
2222
# @export
23-
setGeneric("aggregateRDD", function(x, zeroValue, seqOp, combOp) { standardGeneric("aggregateRDD") })
23+
setGeneric("aggregateRDD",
24+
function(x, zeroValue, seqOp, combOp) { standardGeneric("aggregateRDD") })
2425

2526
# @rdname cache-methods
2627
# @export

‎R/pkg/R/pairRDD.R‎

Lines changed: 6 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -560,8 +560,8 @@ setMethod("join",
560560
# Left outer join two RDDs
561561
#
562562
# @description
563-
# \code{leftouterjoin} This function left-outer-joins two RDDs where every element is of the form list(K, V).
564-
# The key types of the two RDDs should be the same.
563+
# \code{leftouterjoin} This function left-outer-joins two RDDs where every element is of
564+
# the form list(K, V). The key types of the two RDDs should be the same.
565565
#
566566
# @param x An RDD to be joined. Should be an RDD where each element is
567567
# list(K, V).
@@ -597,8 +597,8 @@ setMethod("leftOuterJoin",
597597
# Right outer join two RDDs
598598
#
599599
# @description
600-
# \code{rightouterjoin} This function right-outer-joins two RDDs where every element is of the form list(K, V).
601-
# The key types of the two RDDs should be the same.
600+
# \code{rightouterjoin} This function right-outer-joins two RDDs where every element is of
601+
# the form list(K, V). The key types of the two RDDs should be the same.
602602
#
603603
# @param x An RDD to be joined. Should be an RDD where each element is
604604
# list(K, V).
@@ -634,8 +634,8 @@ setMethod("rightOuterJoin",
634634
# Full outer join two RDDs
635635
#
636636
# @description
637-
# \code{fullouterjoin} This function full-outer-joins two RDDs where every element is of the form list(K, V).
638-
# The key types of the two RDDs should be the same.
637+
# \code{fullouterjoin} This function full-outer-joins two RDDs where every element is of
638+
# the form list(K, V). The key types of the two RDDs should be the same.
639639
#
640640
# @param x An RDD to be joined. Should be an RDD where each element is
641641
# list(K, V).

‎R/pkg/R/sparkR.R‎

Lines changed: 6 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -105,7 +105,8 @@ sparkR.init <- function(
105105
sparkPackages = "") {
106106

107107
if (exists(".sparkRjsc", envir = .sparkREnv)) {
108-
cat("Re-using existing Spark Context. Please stop SparkR with sparkR.stop() or restart R to create a new Spark Context\n")
108+
cat(paste("Re-using existing Spark Context.",
109+
"Please stop SparkR with sparkR.stop() or restart R to create a new Spark Context\n"))
109110
return(get(".sparkRjsc", envir = .sparkREnv))
110111
}
111112

@@ -180,14 +181,16 @@ sparkR.init <- function(
180181

181182
sparkExecutorEnvMap <- new.env()
182183
if (!any(names(sparkExecutorEnv) == "LD_LIBRARY_PATH")) {
183-
sparkExecutorEnvMap[["LD_LIBRARY_PATH"]] <- paste0("$LD_LIBRARY_PATH:",Sys.getenv("LD_LIBRARY_PATH"))
184+
sparkExecutorEnvMap[["LD_LIBRARY_PATH"]] <-
185+
paste0("$LD_LIBRARY_PATH:",Sys.getenv("LD_LIBRARY_PATH"))
184186
}
185187
for (varname in names(sparkExecutorEnv)) {
186188
sparkExecutorEnvMap[[varname]] <- sparkExecutorEnv[[varname]]
187189
}
188190

189191
nonEmptyJars <- Filter(function(x) { x != "" }, jars)
190-
localJarPaths <- sapply(nonEmptyJars, function(j) { utils::URLencode(paste("file:", uriSep, j, sep = "")) })
192+
localJarPaths <- sapply(nonEmptyJars,
193+
function(j) { utils::URLencode(paste("file:", uriSep, j, sep = "")) })
191194

192195
# Set the start time to identify jobjs
193196
# Seconds resolution is good enough for this purpose, so use ints

‎R/pkg/R/utils.R‎

Lines changed: 18 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -334,18 +334,21 @@ getStorageLevel <- function(newLevel = c("DISK_ONLY",
334334
"MEMORY_ONLY_SER_2",
335335
"OFF_HEAP")) {
336336
match.arg(newLevel)
337+
strageLevelClass <- "org.apache.spark.storage.StorageLevel"
337338
storageLevel <- switch(newLevel,
338-
"DISK_ONLY" = callJStatic("org.apache.spark.storage.StorageLevel", "DISK_ONLY"),
339-
"DISK_ONLY_2" = callJStatic("org.apache.spark.storage.StorageLevel", "DISK_ONLY_2"),
340-
"MEMORY_AND_DISK" = callJStatic("org.apache.spark.storage.StorageLevel", "MEMORY_AND_DISK"),
341-
"MEMORY_AND_DISK_2" = callJStatic("org.apache.spark.storage.StorageLevel", "MEMORY_AND_DISK_2"),
342-
"MEMORY_AND_DISK_SER" = callJStatic("org.apache.spark.storage.StorageLevel", "MEMORY_AND_DISK_SER"),
343-
"MEMORY_AND_DISK_SER_2" = callJStatic("org.apache.spark.storage.StorageLevel", "MEMORY_AND_DISK_SER_2"),
344-
"MEMORY_ONLY" = callJStatic("org.apache.spark.storage.StorageLevel", "MEMORY_ONLY"),
345-
"MEMORY_ONLY_2" = callJStatic("org.apache.spark.storage.StorageLevel", "MEMORY_ONLY_2"),
346-
"MEMORY_ONLY_SER" = callJStatic("org.apache.spark.storage.StorageLevel", "MEMORY_ONLY_SER"),
347-
"MEMORY_ONLY_SER_2" = callJStatic("org.apache.spark.storage.StorageLevel", "MEMORY_ONLY_SER_2"),
348-
"OFF_HEAP" = callJStatic("org.apache.spark.storage.StorageLevel", "OFF_HEAP"))
339+
"DISK_ONLY" = callJStatic(strageLevelClass, "DISK_ONLY"),
340+
"DISK_ONLY_2" = callJStatic(strageLevelClass, "DISK_ONLY_2"),
341+
"MEMORY_AND_DISK" = callJStatic(strageLevelClass, "MEMORY_AND_DISK"),
342+
"MEMORY_AND_DISK_2" = callJStatic(strageLevelClass, "MEMORY_AND_DISK_2"),
343+
"MEMORY_AND_DISK_SER" = callJStatic(strageLevelClass,
344+
"MEMORY_AND_DISK_SER"),
345+
"MEMORY_AND_DISK_SER_2" = callJStatic(strageLevelClass,
346+
"MEMORY_AND_DISK_SER_2"),
347+
"MEMORY_ONLY" = callJStatic(strageLevelClass, "MEMORY_ONLY"),
348+
"MEMORY_ONLY_2" = callJStatic(strageLevelClass, "MEMORY_ONLY_2"),
349+
"MEMORY_ONLY_SER" = callJStatic(strageLevelClass, "MEMORY_ONLY_SER"),
350+
"MEMORY_ONLY_SER_2" = callJStatic(strageLevelClass, "MEMORY_ONLY_SER_2"),
351+
"OFF_HEAP" = callJStatic(strageLevelClass, "OFF_HEAP"))
349352
}
350353

351354
# Utility function for functions where an argument needs to be integer but we want to allow
@@ -545,9 +548,11 @@ mergePartitions <- function(rdd, zip) {
545548
lengthOfKeys <- part[[len - lengthOfValues]]
546549
stopifnot(len == lengthOfKeys + lengthOfValues)
547550

548-
# For zip operation, check if corresponding partitions of both RDDs have the same number of elements.
551+
# For zip operation, check if corresponding partitions
552+
# of both RDDs have the same number of elements.
549553
if (zip && lengthOfKeys != lengthOfValues) {
550-
stop("Can only zip RDDs with same number of elements in each pair of corresponding partitions.")
554+
stop(paste("Can only zip RDDs with same number of elements",
555+
"in each pair of corresponding partitions."))
551556
}
552557

553558
if (lengthOfKeys > 1) {

‎R/pkg/inst/tests/test_includeJAR.R‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -18,8 +18,8 @@ context("include an external JAR in SparkContext")
1818

1919
runScript <- function() {
2020
sparkHome <- Sys.getenv("SPARK_HOME")
21-
jarPath <- paste("--jars",
22-
shQuote(file.path(sparkHome, "R/lib/SparkR/test_support/sparktestjar_2.10-1.0.jar")))
21+
sparkTestJarPath <- "R/lib/SparkR/test_support/sparktestjar_2.10-1.0.jar"
22+
jarPath <- paste("--jars", shQuote(file.path(sparkHome, sparkTestJarPath)))
2323
scriptPath <- file.path(sparkHome, "R/lib/SparkR/tests/jarTest.R")
2424
submitPath <- file.path(sparkHome, "bin/spark-submit")
2525
res <- system2(command = submitPath,

‎R/pkg/inst/tests/test_rdd.R‎

Lines changed: 8 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -669,27 +669,31 @@ test_that("fullOuterJoin() on pairwise RDDs", {
669669
rdd1 <- parallelize(sc, list(list(1,2), list(1,3), list(3,3)))
670670
rdd2 <- parallelize(sc, list(list(1,1), list(2,4)))
671671
actual <- collect(fullOuterJoin(rdd1, rdd2, 2L))
672-
expected <- list(list(1, list(2, 1)), list(1, list(3, 1)), list(2, list(NULL, 4)), list(3, list(3, NULL)))
672+
expected <- list(list(1, list(2, 1)), list(1, list(3, 1)),
673+
list(2, list(NULL, 4)), list(3, list(3, NULL)))
673674
expect_equal(sortKeyValueList(actual), sortKeyValueList(expected))
674675

675676
rdd1 <- parallelize(sc, list(list("a",2), list("a",3), list("c", 1)))
676677
rdd2 <- parallelize(sc, list(list("a",1), list("b",4)))
677678
actual <- collect(fullOuterJoin(rdd1, rdd2, 2L))
678-
expected <- list(list("b", list(NULL, 4)), list("a", list(2, 1)), list("a", list(3, 1)), list("c", list(1, NULL)))
679+
expected <- list(list("b", list(NULL, 4)), list("a", list(2, 1)),
680+
list("a", list(3, 1)), list("c", list(1, NULL)))
679681
expect_equal(sortKeyValueList(actual),
680682
sortKeyValueList(expected))
681683

682684
rdd1 <- parallelize(sc, list(list(1,1), list(2,2)))
683685
rdd2 <- parallelize(sc, list(list(3,3), list(4,4)))
684686
actual <- collect(fullOuterJoin(rdd1, rdd2, 2L))
685687
expect_equal(sortKeyValueList(actual),
686-
sortKeyValueList(list(list(1, list(1, NULL)), list(2, list(2, NULL)), list(3, list(NULL, 3)), list(4, list(NULL, 4)))))
688+
sortKeyValueList(list(list(1, list(1, NULL)), list(2, list(2, NULL)),
689+
list(3, list(NULL, 3)), list(4, list(NULL, 4)))))
687690

688691
rdd1 <- parallelize(sc, list(list("a",1), list("b",2)))
689692
rdd2 <- parallelize(sc, list(list("c",3), list("d",4)))
690693
actual <- collect(fullOuterJoin(rdd1, rdd2, 2L))
691694
expect_equal(sortKeyValueList(actual),
692-
sortKeyValueList(list(list("a", list(1, NULL)), list("b", list(2, NULL)), list("d", list(NULL, 4)), list("c", list(NULL, 3)))))
695+
sortKeyValueList(list(list("a", list(1, NULL)), list("b", list(2, NULL)),
696+
list("d", list(NULL, 4)), list("c", list(NULL, 3)))))
693697
})
694698

695699
test_that("sortByKey() on pairwise RDDs", {

‎R/pkg/inst/tests/test_sparkSQL.R‎

Lines changed: 9 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -391,7 +391,7 @@ test_that("collect() and take() on a DataFrame return the same number of rows an
391391
expect_equal(ncol(collect(df)), ncol(take(df, 10)))
392392
})
393393

394-
test_that("multiple pipeline transformations starting with a DataFrame result in an RDD with the correct values", {
394+
test_that("multiple pipeline transformations result in an RDD with the correct values", {
395395
df <- jsonFile(sqlContext, jsonPath)
396396
first <- lapply(df, function(row) {
397397
row$age <- row$age + 5
@@ -756,7 +756,14 @@ test_that("toJSON() returns an RDD of the correct values", {
756756
test_that("showDF()", {
757757
df <- jsonFile(sqlContext, jsonPath)
758758
s <- capture.output(showDF(df))
759-
expect_output(s , "+----+-------+\n| age| name|\n+----+-------+\n|null|Michael|\n| 30| Andy|\n| 19| Justin|\n+----+-------+\n")
759+
expected <- paste("+----+-------+\n",
760+
"| age| name|\n",
761+
"+----+-------+\n",
762+
"|null|Michael|\n",
763+
"| 30| Andy|\n",
764+
"| 19| Justin|\n",
765+
"+----+-------+\n", sep="")
766+
expect_output(s , expected)
760767
})
761768

762769
test_that("isLocal()", {

0 commit comments

Comments
 (0)