Introduction to sparklyr

library(sparklyr)
library(dplyr) # sparklyr works with the verbs of dplyr

library(pwt9) # Load data
data("pwt9.0")

Installing and connecting to Spark

Before Spark can be used, it has to be available on your machine. A local version can be installed directly from R with the following command, which downloads Spark into a folder in your home directory. This has to be done only once.²

spark_install()

Every session then begins with a connection to a cluster, which is established with spark_connect. Its master argument describes where that cluster is. For a real cluster this would be its address. The value "local" means that Spark runs on your own computer and treats its cores as if they were the nodes of a cluster. This is the usual way to develop and test code before it is sent to a productive environment.

sc <- spark_connect(master = "local")

The resulting object sc is the connection to Spark. It is the first argument of practically every function of the package, which is why it should always be assigned to an object.

Spark data frames

sparklyr does not work with data.frames or tibbles, but with tables which live in Spark. A local data.frame is copied into the cluster with copy_to, where the name argument gives the name under which the table is registered in Spark:

pwt_spark <- copy_to(sc, pwt9.0, name = "pwt", overwrite = TRUE)

The object pwt_spark prints like a tibble, but it does not contain any data. It is only a reference to the table in Spark.³ The tables which are currently available in the cluster are listed with src_tbls(sc) and an existing table can be referred to again with tbl(sc, "pwt").

In practice, data are rarely copied from R into Spark, because a data set which requires Spark is usually far too large to be loaded into R in the first place. Instead, the data are read into the cluster directly with functions like spark_read_csv(sc, name = "pwt", path = "pwt.csv") or spark_read_parquet.

pipes

Like dplyr, sparklyr is used together with the pipe operator %>%, which forwards the output of the part to its left as input for the part to its right. If you are not familiar with it, take a look at the section on pipes in the introduction to dplyr.

select

The select function can be used to select the columns of a table. For example, the pwt9.0 data set contains 47 columns. But let’s assume that we only want to analyse real GDP and the population size, i.e. variables rgdpna and pop, for all countries and all years. This can be done with the following line:

pwt_select <- pwt_spark %>% select(year, country, rgdpna, pop)

This is exactly the line from the introduction to dplyr, with the only difference that the data do not come from a tibble, but from a table in Spark. And this is the main point of the package: the verbs stay the same, no matter how large the data behind them are.

However, there is one thing which does not work any more. Since a Spark table is not a data.frame, it cannot be subset with the standard R operators. pwt_spark[, c("year", "country", "rgdpna", "pop")] results in an error, whereas the same command would work on the original data.frame.

rename

Let’s assume we are not satisfied with the original column names and want to change them. As in dplyr, this is done with rename, where the new name of the column is followed by the equality sign and the old name of the column:

pwt_rename <- pwt_select %>%
  rename(ctr = country,
         gdp = rgdpna)

Note that it is not necessary to write the commands in one line. In fact, it is good practise to begin a new code line after a pipe.

filter

Observations are selected with the filter function, where the conditions can be added consecutively and separated with a comma. Each condition can be regarded as an and operator. Or conditions must be specified in the same line. The following code uses the data in pwt_rename and omits all observations, where the column gdp contains NA values, where the year is lower than 1980 and where the country name is not Austria, Germany or France.⁴

pwt_filter <- pwt_rename %>%
  filter(!is.na(gdp), # Omit observations with NA values in column "gdp"
         year >= 1980, # Drop observations before 1980
         ctr %in% c("Austria", "Germany", "France")) # Only use obs for AT, DE and FR

Behind the scenes these conditions are not evaluated by R, but translated into the WHERE clause of a SQL query, which is then executed by Spark. This also implies a limitation you should be aware of: only those R functions can be used, for which sparklyr knows a Spark counterpart. Basic arithmetic, the usual mathematical and string functions and the aggregation functions are covered, but a function from an arbitrary R package is not. If a function cannot be translated, you will get an error.

mutate

The mutate function can be used to change the content of an existing column or to generate a new one. In our example we calculate the GDP per capita ratio for the three countries in the filtered data set:

pwt_mutate <- pwt_filter %>%
  mutate(gdp_pc = gdp / pop)

group_by

As in dplyr, the group_by function tells Spark to execute the following commands for each distinct value in the specified columns separately. This should become more clear when we look more closely at the following function…

summarise

Sometimes it might be useful to calculate aggregate values over a group of observations. This can be done with the summarise function in combination with the group_by function. The following code calculates aggregate GDP and population for each year, i.e. for the region of Austria, France and Germany.

data_summarise <- pwt_filter %>%
  group_by(year) %>%
  summarise(gdp = sum(gdp),
            pop = sum(pop)) %>%
  ungroup()

This is the kind of operation which benefits most from Spark. Each node of the cluster can aggregate the part of the data it holds and only the comparatively small partial results have to be combined afterwards.

Laziness and collect

Up to this point not a single number has been calculated. sparklyr is lazy: it collects the verbs, translates them into one SQL query and only sends that query to Spark when the result is actually needed. This is a substantial advantage, because it allows Spark to optimise all the steps together instead of executing them one after another. The query which was built up in the meantime can be inspected with

data_summarise %>% show_query()

## <SQL>
## SELECT `year`, SUM(`gdp`) AS `gdp`, SUM(`pop`) AS `pop`
## FROM (
##   SELECT `year`, `country` AS `ctr`, `rgdpna` AS `gdp`, `pop`
##   FROM `pwt`
##   WHERE
##     (NOT((`rgdpna` IS NULL))) AND
##     (`year` >= 1980.0) AND
##     (`country` IN ('Austria', 'Germany', 'France'))
## ) AS `q01`
## GROUP BY `year`

The result of a query is brought from the cluster into the R session with collect, which returns an ordinary tibble that can be used like any other object in R:

data_local <- data_summarise %>% collect()

Since collect loads its result into the memory of your computer, it should only be used for results which are small enough to fit in there. But this is exactly the way Spark is usually applied: the expensive filtering and aggregation happens in the cluster and only the small final result is pulled back into R, where it is plotted with ggplot2 or written into a report.

If an intermediate result is needed repeatedly, compute stores it as a table in Spark, so that the steps up to that point do not have to be calculated again:

pwt_cached <- pwt_filter %>% compute("pwt_filtered")

Disconnecting

At the end of a session the connection should be closed, which frees the resources of the cluster:

spark_disconnect(sc)

After that, objects like pwt_spark become useless, because the tables they refer to do not exist any more. So everything you want to keep has to be collected before, or written to disk from within Spark with functions like spark_write_csv or spark_write_parquet.

Beside the verbs presented above there are two further families of functions you should know when you work with sparklyr. The functions which begin with sdf_ operate on Spark data frames directly and cover things for which there is no dplyr verb, for example sdf_nrow or sdf_describe. The functions which begin with ml_ give access to the machine learning library of Spark, so that a model like ml_linear_regression can be estimated on data which never enter the memory of R.

Finally, a word of caution. Spark comes with a considerable overhead. Starting a session, distributing the data and translating the queries all take time, which is why an analysis of a moderately sized data set will almost always be faster with dplyr or data.table on a single machine. Spark becomes the better choice when the data do not fit into memory any more – and if you have reached that point, the fact that you can keep using the verbs of dplyr is worth a lot. There is also plenty of additional material out there, which assists in becoming more familiar with the package. The website of the package contains a large collection of examples and, as usual, I find so-called cheatsheets especially useful.⁵