Distributing R Computations

Overview

sparklyr provides support to run arbitrary R code at scale within your Spark Cluster through spark_apply(). This is especially useful where there is a need to use functionality available only in R or R packages that is not available in Apache Spark nor Spark Packages.

spark_apply() applies an R function to a Spark object (typically, a Spark DataFrame). Spark objects are partitioned so they can be distributed across a cluster. You can use spark_apply() with the default partitions or you can define your own partitions with the group_by() argument. Your R function must return another Spark DataFrame. spark_apply() will run your R function on each partition and output a single Spark DataFrame.

Apply an R function to a Spark Object

Lets run a simple example. We will apply the identify function, I(), over a list of numbers we created with the sdf_len() function.

library(sparklyr)

sc <- spark_connect(master = "local")

sdf_len(sc, 5, repartition = 1) %>%
  spark_apply(function(e) I(e))
#> # Source: spark<?> [?? x 1]
#>      id
#>   <int>
#> 1     1
#> 2     2
#> 3     3
#> 4     4
#> 5     5

Your R function should be designed to operate on an R data frame. The R function passed to spark_apply() expects a DataFrame and will return an object that can be cast as a DataFrame. We can use the class function to verify the class of the data.

sdf_len(sc, 10, repartition = 1) %>%
  spark_apply(function(e) class(e))
#> # Source: spark<?> [?? x 1]
#>   result    
#>   <chr>     
#> 1 tbl_df    
#> 2 tbl       
#> 3 data.frame

Spark will partition your data by hash or range so it can be distributed across a cluster. In the following example we create two partitions and count the number of rows in each partition. Then we print the first record in each partition.

trees_tbl <- sdf_copy_to(sc, trees, repartition = 2)

spark_apply(
  trees_tbl,
  function(e) nrow(e), names = "n"
  )
#> # Source: spark<?> [?? x 1]
#>       n
#>   <int>
#> 1    15
#> 2    16

spark_apply(trees_tbl, function(e) head(e, 1))
#> # Source: spark<?> [?? x 3]
#>   Girth Height Volume
#>   <dbl>  <dbl>  <dbl>
#> 1   8.3     70   10.3
#> 2  12.9     74   22.2

We can apply any arbitrary function to the partitions in the Spark DataFrame. For instance, we can scale or jitter the columns. Notice that spark_apply() applies the R function to all partitions and returns a single DataFrame.

spark_apply(trees_tbl, function(e) scale(e))
#> # Source: spark<?> [?? x 3]
#>      Girth Height  Volume
#>      <dbl>  <dbl>   <dbl>
#>  1 -2.05   -0.607 -1.69  
#>  2 -1.79   -1.43  -1.69  
#>  3 -1.62   -1.76  -1.71  
#>  4 -0.134  -0.276 -0.339 
#>  5  0.0407  1.21   0.191 
#>  6  0.128   1.54   0.390 
#>  7  0.302  -1.27  -0.515 
#>  8  0.302   0.221  0.0589
#>  9  0.389   1.05   1.03  
#> 10  0.477   0.221  0.434 
#> # … with more rows

spark_apply(trees_tbl, function(e) lapply(e, jitter))
#> # Source: spark<?> [?? x 3]
#>    Girth Height Volume
#>    <dbl>  <dbl>  <dbl>
#>  1  8.29   70.0   10.3
#>  2  8.60   65.0   10.3
#>  3  8.82   63.0   10.2
#>  4 10.5    72.1   16.4
#>  5 10.7    80.8   18.8
#>  6 10.8    83.0   19.7
#>  7 11.0    66.1   15.6
#>  8 11.0    75.1   18.2
#>  9 11.1    80.2   22.6
#> 10 11.2    75.2   19.9
#> # … with more rows

By default spark_apply() derives the column names from the input Spark data frame. Use the names argument to rename or add new columns.

spark_apply(
  trees_tbl,
  function(e) data.frame(2.54 * e$Girth, e), names = c("Girth(cm)", colnames(trees))
  )
#> # Source: spark<?> [?? x 4]
#>    `Girth(cm)` Girth Height Volume
#>          <dbl> <dbl>  <dbl>  <dbl>
#>  1        21.1   8.3     70   10.3
#>  2        21.8   8.6     65   10.3
#>  3        22.4   8.8     63   10.2
#>  4        26.7  10.5     72   16.4
#>  5        27.2  10.7     81   18.8
#>  6        27.4  10.8     83   19.7
#>  7        27.9  11       66   15.6
#>  8        27.9  11       75   18.2
#>  9        28.2  11.1     80   22.6
#> 10        28.4  11.2     75   19.9
#> # … with more rows

Group By

In some cases you may want to apply your R function to specific groups in your data. For example, suppose you want to compute regression models against specific subgroups. To solve this, you can specify a group_by() argument. This example counts the number of rows in iris by species and then fits a simple linear model for each species.

iris_tbl <- sdf_copy_to(sc, iris)

spark_apply(iris_tbl, nrow, group_by = "Species")
#> # Source: spark<?> [?? x 2]
#>   Species    result
#>   <chr>       <int>
#> 1 versicolor     50
#> 2 virginica      50
#> 3 setosa         50

iris_tbl %>%
  spark_apply(
    function(e) summary(lm(Petal_Length ~ Petal_Width, e))$r.squared,
    names = "r.squared",
    group_by = "Species"
    )
#> # Source: spark<?> [?? x 2]
#>   Species    r.squared
#>   <chr>          <dbl>
#> 1 versicolor     0.619
#> 2 virginica      0.104
#> 3 setosa         0.110

Distributing Packages

With spark_apply() you can use any R package inside Spark. For instance, you can use the broom package to create a tidy data frame from linear regression output.

spark_apply(
  iris_tbl,
  function(e) broom::tidy(lm(Petal_Length ~ Petal_Width, e)),
  names = c("term", "estimate", "std.error", "statistic", "p.value"),
  group_by = "Species"
  )
#> # Source: spark<?> [?? x 6]
#>   Species    term        estimate std.error stati…¹  p.value
#>   <chr>      <chr>          <dbl>     <dbl>   <dbl>    <dbl>
#> 1 versicolor (Intercept)    1.78     0.284     6.28 9.48e- 8
#> 2 versicolor Petal_Width    1.87     0.212     8.83 1.27e-11
#> 3 virginica  (Intercept)    4.24     0.561     7.56 1.04e- 9
#> 4 virginica  Petal_Width    0.647    0.275     2.36 2.25e- 2
#> 5 setosa     (Intercept)    1.33     0.0600   22.1  7.68e-27
#> 6 setosa     Petal_Width    0.546    0.224     2.44 1.86e- 2
#> # … with abbreviated variable name ¹statistic

To use R packages inside Spark, your packages must be installed on the worker nodes. The first time you call spark_apply() all of the contents in your local .libPaths() will be copied into each Spark worker node via the SparkConf.addFile() function. Packages will only be copied once and will persist as long as the connection remains open. It’s not uncommon for R libraries to be several gigabytes in size, so be prepared for a one-time tax while the R packages are copied over to your Spark cluster. You can disable package distribution by setting packages = FALSE. Note: packages are not copied in local mode (master="local") because the packages already exist on the system.

Handling Errors

It can be more difficult to troubleshoot R issues in a cluster than in local mode. For instance, the following R code causes the distributed execution to fail and suggests you check the logs for details.

spark_apply(iris_tbl, function(e) stop("Make this fail"))

It is worth mentioning that different cluster providers and platforms expose worker logs in different ways. Specific documentation for your environment will point out how to retrieve these logs.

Requirements

The R Runtime is expected to be pre-installed in the cluster for spark_apply() to function. Failure to install the cluster will trigger a Cannot run program, no such file or directory error while attempting to use spark_apply(). Contact your cluster administrator to consider making the R runtime available throughout the entire cluster.

A Homogeneous Cluster is required since the driver node distributes, and potentially compiles, packages to the workers. For instance, the driver and workers must have the same processor architecture, system libraries, etc.

Configuration

The following table describes relevant parameters while making use of spark_apply.

Value	Description
`spark.r.command`	The path to the R binary. Useful to select from multiple R versions.
`sparklyr.worker.gateway.address`	The gateway address to use under each worker node. Defaults to `sparklyr.gateway.address`.
`sparklyr.worker.gateway.port`	The gateway port to use under each worker node. Defaults to `sparklyr.gateway.port`.

For example, one could make use of an specific R version by running:

config <- spark_config()
config[["spark.r.command"]] <- "<path-to-r-version>"
sc <- spark_connect(master = "local", config = config)

sdf_len(sc, 10) %>% spark_apply(function(e) e)

Limitations

Closures

Closures are serialized using serialize, which is described as “A simple low-level interface for serializing to connections.”. One of the current limitations of serialize is that it wont serialize objects being referenced outside of it’s environment. For instance, the following function will error out since the closures references external_value:

external_value <- 1
spark_apply(iris_tbl, function(e) e + external_value)

Livy

Currently, Livy connections do not support distributing packages since the client machine where the libraries are pre-compiled might not have the same processor architecture, not operating systems that the cluster machines.

Computing over Groups

While performing computations over groups, spark_apply() will provide partitions over the selected column; however, this implies that each partition can fit into a worker node, if this is not the case an exception will be thrown. To perform operations over groups that exceed the resources of a single node, one can consider partitioning to smaller units or use dplyr::do() which is currently optimized for large partitions.

Package Installation

Since packages are copied only once for the duration of the spark_connect() connection, installing additional packages is not supported while the connection is active. Therefore, if a new package needs to be installed, spark_disconnect() the connection, modify packages and reconnect.