Skip to content

SPARK-16785 R dapply doesn't return array or raw columns #14783

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
clarkfitzg wants to merge 21 commits into from
Closed

SPARK-16785 R dapply doesn't return array or raw columns #14783

Show file tree
Hide file tree
Changes from 17 commits
Commits
Show all changes
21 commits
Select commit Hold shift + click to select a range
f8e1920
test for array and byte columns
clarkfitzg Aug 18, 2016
1336605
R createDataFrame.default uses list of dataframes as rows
clarkfitzg Aug 18, 2016
a0e13ff
test is working now
clarkfitzg Aug 18, 2016
5015233
add simple collect() to test
clarkfitzg Aug 19, 2016
d044054
document how tests fail
clarkfitzg Aug 19, 2016
311b554
identified where patch went wrong
clarkfitzg Aug 19, 2016
2d2654d
back to original code for creating data frame
clarkfitzg Aug 19, 2016
ff1a0d0
first pass modifying worker.R
clarkfitzg Aug 19, 2016
1e27ef3
no change in error message, reverting
clarkfitzg Aug 19, 2016
25d0ec1
an experiment modifying dapplyCollect directly
clarkfitzg Aug 19, 2016
77a9822
dapplyCollect worked, it just nested things
clarkfitzg Aug 24, 2016
70b0d44
tests pass!
clarkfitzg Aug 24, 2016
b21a21d
put rbind function in utils.R
clarkfitzg Aug 24, 2016
ba87b06
rename to rbindRaws and put in utils.R
clarkfitzg Aug 24, 2016
528fa6e
just whitespace
clarkfitzg Aug 24, 2016
e0a3894
syntax error in worker.R
clarkfitzg Aug 24, 2016
5871257
satisfy lintr
clarkfitzg Aug 24, 2016
84ef4cc
address sun-rui's comments
clarkfitzg Aug 24, 2016
0c2a215
Change names from list_cols -> binary per felixcheung's feedback
clarkfitzg Aug 31, 2016
77fa9b4
fix R style fail
clarkfitzg Aug 31, 2016
91d69be
tests for a single binary column
clarkfitzg Sep 7, 2016
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 3 additions & 0 deletions R/pkg/R/SQLContext.R
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -183,6 +183,8 @@ getDefaultSqlSource <- function() {
# TODO(davies): support sampling and infer type from NA
createDataFrame.default <- function(data, schema = NULL, samplingRatio = 1.0) {
sparkSession <- getSparkSession()

# Convert dataframes into a list of rows. Each row is a list

@sun-rui sun-rui Aug 24, 2016 •
edited
Loading

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

how about " If the data is a dataframe, convert it into ..."?

if (is.data.frame(data)) {
# get the names of columns, they will be put into RDD
if (is.null(schema)) {
Expand All @@ -208,6 +210,7 @@ createDataFrame.default <- function(data, schema = NULL, samplingRatio = 1.0) {
args <- list(FUN = list, SIMPLIFY = FALSE, USE.NAMES = FALSE)
data <- do.call(mapply, append(args, data))
}

if (is.list(data)) {
sc <- callJStatic("org.apache.spark.sql.api.r.SQLUtils", "getJavaSparkContext", sparkSession)
rdd <- parallelize(sc, data)
Expand Down
15 changes: 15 additions & 0 deletions R/pkg/R/utils.R
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -697,3 +697,18 @@ is_master_local <- function(master) {
is_sparkR_shell <- function() {
grepl(".*shell\\.R$", Sys.getenv("R_PROFILE_USER"), perl = TRUE)
}

# rbind a list of rows with raw (binary) columns
#
# @param inputData a list of rows, with each row a list
# @return data.frame with raw columns as lists
rbindRaws <- function(inputData){
row1 <- inputData[[1]]
rawcolumns <- ("raw" == sapply(row1, class))

listmatrix <- do.call(rbind, inputData)

@shivaram shivaram Sep 7, 2016

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Do you know what happens if we have a mixed set of columns here ? i.e. say one column with "raw", one with "integer" and one with "character" -- From reading some docs it looks like everything is converted to create a character matrix when we use rbind.

I think we have two choices if thats the case
(a) we apply the type conversions after rbind
(b) we only call this method when all columns are raw

@clarkfitzg clarkfitzg Sep 7, 2016

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more. 

> b = serialize(1:10, NULL)
> inputData = list(list(1L, b, 'a'), list(2L, b, 'b'))  # Mixed data types
> listmatrix <- do.call(rbind, inputData)
> listmatrix
     [,1] [,2]   [,3]
[1,] 1    Raw,62 "a"
[2,] 2    Raw,62 "b"
> class(listmatrix)
[1] "matrix"
> typeof(listmatrix)
[1] "list"
> is.character(listmatrix)
[1] FALSE

A little unusual- it's a list matrix. Hence the name. Which docs are you referring to?

The test that's in here now does test for mixed columns, but it doesn't test for a single column of raws. I'll add that now.

@shivaram shivaram Sep 7, 2016

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I was looking at https://stat.ethz.ch/R-manual/R-devel/library/base/html/cbind.html specifically the section Value which says

The type of a matrix result determined from the highest type of any of the inputs in the hierarchy raw < logical < integer < double < complex < character < list .

@felixcheung felixcheung Sep 7, 2016

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think the correct class is maintained:

> sapply(listmatrix, class)
[1] "integer"   "integer"   "raw"       "raw"       "character" "character"
> sapply(listmatrix, typeof)
[1] "integer"   "integer"   "raw"       "raw"       "character" "character"

@shivaram shivaram Sep 7, 2016

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Ah I see - the types are inside the listmatrix. Thanks @clarkfitzg for clarifying. Let us know once you have added the test for a single column of raw as well.

@clarkfitzg clarkfitzg Sep 7, 2016

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Since everything in in inputData is a list this goes straight to the top of hierarchy- same as if you called rbind(list1, list2, ...).

# A dataframe with all list columns
out <- as.data.frame(listmatrix)
out[!rawcolumns] <- lapply(out[!rawcolumns], unlist)
out
}
19 changes: 19 additions & 0 deletions R/pkg/inst/tests/testthat/test_sparkSQL.R
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2248,6 +2248,25 @@ test_that("dapply() and dapplyCollect() on a DataFrame", {
expect_identical(expected, result)
})

test_that("dapplyCollect() on dataframe with list columns", {

df_listcols <- data.frame(key = 1:3)
df_listcols$bytes <- lapply(df_listcols$key, serialize, connection = NULL)

# TODO clarkfitzg: Related issue- The dataframe can't be collected if this
# column is added:
#df_listcols$arr <- lapply(df_listcols$key,
# function(x) seq(0, 1, length.out=15))

df_listcols_spark <- createDataFrame(df_listcols)

result1 <- collect(df_listcols_spark)
expect_identical(df_listcols, result1)

result2 <- dapplyCollect(df_listcols_spark, function(x) x)
expect_equal(df_listcols, result2)
})

test_that("repartition by columns on DataFrame", {
df <- createDataFrame(
list(list(1L, 1, "1", 0.1), list(1L, 2, "2", 0.2), list(3L, 3, "3", 0.3)),
Expand Down
9 changes: 9 additions & 0 deletions R/pkg/inst/tests/testthat/test_utils.R
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -183,4 +183,13 @@ test_that("overrideEnvs", {
expect_equal(config[["config_only"]], "ok")
})

test_that("rbindRaws", {
r <- serialize(1, connection = NULL)
inputData <- list(list(1L, r), list(2L, r), list(3L, r))
expected <- data.frame(V1 = 1:3)
expected$V2 <- list(r, r, r)
result <- SparkR:::rbindRaws(inputData)

@sun-rui sun-rui Aug 24, 2016

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

SparkR::: is not needed

expect_equal(expected, result)
})

sparkR.session.stop()
9 changes: 8 additions & 1 deletion R/pkg/inst/worker/worker.R
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -36,7 +36,14 @@ compute <- function(mode, partition, serializer, deserializer, key,
# available since R 3.2.4. So we set the global option here.
oldOpt <- getOption("stringsAsFactors")
options(stringsAsFactors = FALSE)
inputData <- do.call(rbind.data.frame, inputData)

# Handle binary data types
if ("raw" %in% sapply(inputData[[1]], class)) {
inputData <- SparkR:::rbindRaws(inputData)

@sun-rui sun-rui Aug 24, 2016

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

same as above. SparkR::: is not needed

@clarkfitzg clarkfitzg Aug 24, 2016

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

True, but looking through the rest of worker.R it seems that using ::: is the convention in this file?

@clarkfitzg clarkfitzg Aug 24, 2016

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Running tests locally without it- appears it is necessary here.

@sun-rui sun-rui Aug 24, 2016 •
edited
Loading

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

No it is not a preferred style in worker.R. It seems that they were some changes left slip under some previous code review.

    suppressPackageStartupMessages(library(SparkR))

should be moved to the front of work.R, and thus SparkR::: can be removed. A lot of "SparkR:::" is annoying.

@clarkfitzg clarkfitzg Aug 24, 2016

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

When I remove SparkR::: and run the tests locally I see error with rbindRaws not found.

So it looks like the SparkR::: needs to be there to access private functions in SparkR from the worker node.

@clarkfitzg clarkfitzg Aug 24, 2016

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

And rbindRaws doesn't need to be exported.

@sun-rui sun-rui Aug 24, 2016

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

sure. "SparkR:::" is needed for private functions.

} else {
inputData <- do.call(rbind.data.frame, inputData)
}

options(stringsAsFactors = oldOpt)

names(inputData) <- colNames
Expand Down