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
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))
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)))
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")
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.
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.
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.
The following table describes relevant parameters while making use of
||The path to the R binary. Useful to select from multiple R versions.|
||The gateway address to use under each worker node. Defaults to
||The gateway port to use under each worker node. Defaults to
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)
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 <- 1 spark_apply(iris_tbl, function(e) e + external_value)
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.
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.