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))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))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)
trees_tbl %>%
spark_apply(function(e) nrow(e), names = "n")
trees_tbl %>%
spark_apply(function(e) head(e, 1))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.
trees_tbl %>%
spark_apply(function(e) scale(e))
trees_tbl %>%
spark_apply(function(e) lapply(e, jitter))By default spark_apply() derives the column names from the input Spark data frame. Use the names argument to rename or add new columns.
trees_tbl %>%
spark_apply(
function(e) data.frame(2.54 * e$Girth, e),
names = c("Girth(cm)", colnames(trees)))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)
iris_tbl %>%
spark_apply(nrow, group_by = "Species")
iris_tbl %>%
spark_apply(
function(e) summary(lm(Petal_Length ~ Petal_Width, e))$r.squared,
names = "r.squared",
group_by = "Species")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")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")) Error in force(code) :
sparklyr worker rscript failure, check worker logs for details
In local mode, sparklyr will retrieve the logs for you. The logs point out the real failure as ERROR sparklyr: RScript (4190) Make this fail as you might expect.
---- Output Log ----
(17/07/27 21:24:18 ERROR sparklyr: Worker (2427) is shutting down with exception ,java.net.SocketException: Socket closed)
17/07/27 21:24:18 WARN TaskSetManager: Lost task 0.0 in stage 389.0 (TID 429, localhost, executor driver): 17/07/27 21:27:21 INFO sparklyr: RScript (4190) retrieved 150 rows
17/07/27 21:27:21 INFO sparklyr: RScript (4190) computing closure
17/07/27 21:27:21 ERROR sparklyr: RScript (4190) 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"]] <- ""
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 precompiled 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.
spark_disconnect(sc)